c7a115c928
이전(PR#5)은 비루프 AnalogPoint에만, 또 realtime=False로 보조를 넣어 적산 QV가 크롤러 폴링집합(is_active AND realtime_enabled)에서 빠져 라이브로 흐르지 않았다. - is_analog 판정을 분기 앞으로 올려 모든 AnalogPoint(루프 포함)에 보조 처리 적용. 루프 엔트리와 겹치면 seen_tags가 중복 흡수. - 슬롯명 있는 A1~A4를 일반적으로 모두 어드레싱(공백 슬롯만 제외). - realtime 기본값(True) 사용 → 보조도 폴링 대상. QV=연속 누적값만 archive 유지. - FlexibleParameter(사용자정의)는 기존대로 realtime=False(폴링 제외) 유지. 맵 재생성(c1~c4): 순수 추가. C3 +6(FICQ-8118/LICA-5113 GAIN/RATE/RESET 등 루프 보조 누락분), 리포트 스팀적산 FIQ-*.QV 전부 포함, 제거 0. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
709 lines
31 KiB
Python
709 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Build register-map-cN.json from Sinam_Tag_all.xlsx ALONE.
|
|
|
|
No HC Designer CSVs needed at run time; pass --validate-csv only for cross-check.
|
|
|
|
Output: register-map-cN.json (per controller), plus optional DB upsert:
|
|
- tag_metadata (desc, area, state labels, units, range)
|
|
- hc900_map_master (is_active, realtime_enabled, archive_enabled)
|
|
|
|
Usage:
|
|
python3 scripts/build_register_map_from_sinam.py \
|
|
--sinam docs/Sinam_Tag_all.xlsx \
|
|
--controller C3 \
|
|
-o docs/register-map-c3.json \
|
|
[--db-conn "Host=localhost;Database=hc900;Username=postgres;Password=postgres"]
|
|
# optional CSV cross-check:
|
|
python3 scripts/build_register_map_from_sinam.py \
|
|
--sinam docs/Sinam_Tag_all.xlsx \
|
|
--controller C4 \
|
|
--validate-csv docs/C4-All-Modbus-Map.csv \
|
|
-o docs/register-map-c4.json
|
|
"""
|
|
|
|
import re
|
|
import csv
|
|
import json
|
|
import argparse
|
|
import datetime
|
|
from pathlib import Path
|
|
from collections import Counter
|
|
|
|
import openpyxl
|
|
|
|
|
|
# ─────────────────────── HC900 fixed loop layout (Table 6-3) ───────────────────────
|
|
|
|
LOOP_LAYOUT = {
|
|
0x00: ('PV', 2, 'float32', 'R'),
|
|
0x02: ('RSP_SP2', 2, 'float32', 'RW'),
|
|
0x04: ('WSP', 2, 'float32', 'RW'),
|
|
0x06: ('Output', 2, 'float32', 'RW'),
|
|
0x08: ('PV_B', 2, 'float32', 'R'),
|
|
0x0A: ('CarbonPotTemp', 2, 'float32', 'R'),
|
|
0x0C: ('Gain1', 2, 'float32', 'RW'),
|
|
0x0E: ('Direction', 2, 'float32', 'R'),
|
|
0x10: ('Reset1', 2, 'float32', 'RW'),
|
|
0x12: ('Rate1', 2, 'float32', 'RW'),
|
|
0x14: ('CycleTime1', 2, 'float32', 'R'),
|
|
0x16: ('PV_LowRange', 2, 'float32', 'R'),
|
|
0x18: ('PV_HighRange', 2, 'float32', 'R'),
|
|
0x1A: ('Alarm1SP1', 2, 'float32', 'RW'),
|
|
0x1C: ('Alarm1SP2', 2, 'float32', 'RW'),
|
|
0x20: ('Gain2', 2, 'float32', 'RW'),
|
|
0x22: ('StepDeadband', 2, 'float32', 'RW'),
|
|
0x24: ('Reset2', 2, 'float32', 'RW'),
|
|
0x26: ('Rate2', 2, 'float32', 'RW'),
|
|
0x28: ('CycleTime2', 2, 'float32', 'R'),
|
|
0x2A: ('LSP1', 2, 'float32', 'RW'),
|
|
0x2C: ('LSP2', 2, 'float32', 'RW'),
|
|
0x2E: ('Alarm2SP1', 2, 'float32', 'RW'),
|
|
0x30: ('Alarm2SP2', 2, 'float32', 'RW'),
|
|
0x34: ('SP_LowLimit', 2, 'float32', 'RW'),
|
|
0x36: ('SP_HighLimit', 2, 'float32', 'RW'),
|
|
0x38: ('WSP_B', 2, 'float32', 'RW'),
|
|
0x3A: ('Output_LowLimit', 2, 'float32', 'RW'),
|
|
0x3C: ('Output_HighLimit',2, 'float32', 'RW'),
|
|
0x3E: ('OPWORK', 2, 'float32', 'RW'),
|
|
0x46: ('Ratio', 2, 'float32', 'RW'),
|
|
0x48: ('Bias', 2, 'float32', 'RW'),
|
|
0x4A: ('Deviation', 2, 'float32', 'R'),
|
|
0x4E: ('ManualReset', 2, 'float32', 'RW'),
|
|
0x50: ('FeedforwardGain', 2, 'float32', 'RW'),
|
|
0x52: ('LocalPctCO', 2, 'float32', 'RW'),
|
|
0x54: ('FurnaceFactor', 2, 'float32', 'RW'),
|
|
0x56: ('PercentHydrogen', 2, 'float32', 'RW'),
|
|
0x58: ('OnOffHysteresis', 2, 'float32', 'RW'),
|
|
0x5A: ('CarbPotDewpt', 2, 'float32', 'RW'),
|
|
0x5C: ('StepMotorTime', 2, 'float32', 'RW'),
|
|
0xB7: ('FuzzyEnable', 1, 'uint16', 'RW'),
|
|
0xB8: ('DemandTune', 1, 'uint16', 'RW'),
|
|
0xB9: ('AntiSootEnable', 1, 'uint16', 'RW'),
|
|
0xBA: ('AutoManState', 1, 'uint16', 'RW'),
|
|
0xBB: ('SP_SelectState', 1, 'uint16', 'RW'),
|
|
0xBC: ('RemLocSPState', 1, 'uint16', 'RW'),
|
|
0xBD: ('TuneSetState', 1, 'uint16', 'RW'),
|
|
0xBE: ('LoopStatus', 1, 'uint16', 'R'),
|
|
}
|
|
|
|
MNEMONIC_OFFSET = {
|
|
'PV': 0x00, 'RSP': 0x02, 'SP2': 0x02, 'WSP': 0x04, 'OP': 0x06, 'PVB': 0x08,
|
|
'TEMP': 0x0A, 'GAIN1': 0x0C, 'PROP1': 0x0C, 'DIR': 0x0E, 'RESET1': 0x10,
|
|
'RATE1': 0x12, 'CYCLE1': 0x14, 'PVLOW': 0x16, 'PVHIGH': 0x18, 'AL1SP1': 0x1A,
|
|
'AL1SP2': 0x1C, 'GAIN2': 0x20, 'PROP2': 0x20, 'DB': 0x22, 'RESET2': 0x24,
|
|
'RATE2': 0x26, 'CYCLE2': 0x28, 'LSP1': 0x2A, 'LSP2': 0x2C, 'AL2SP1': 0x2E,
|
|
'AL2SP2': 0x30, 'SPLOW': 0x34, 'SPHIGH': 0x36, 'SPWORK': 0x38, 'OPLOW': 0x3A,
|
|
'OPHIGH': 0x3C, 'OPWORK': 0x3E, 'RATIO': 0x46, 'BIAS': 0x48, 'DEV': 0x4A,
|
|
'MAN_RESET': 0x4E, 'FF': 0x50, 'PCTCO': 0x52, 'FFCTR': 0x54, 'H2': 0x56,
|
|
'OUT_HYST': 0x58, 'CPD': 0x5A, 'MOTOR': 0x5C, 'AMSTAT': 0xBA, 'LOOPSTAT': 0xBE,
|
|
'MODEIN': 0xBA,
|
|
}
|
|
|
|
EXPERION_ATTR = {
|
|
'PV': 'PV',
|
|
'WSP': 'SP', 'SPWORK': 'SP',
|
|
'OP': 'OP', 'OPWORK': 'OP',
|
|
'GAIN1': 'GAIN',
|
|
'RESET1': 'RESET',
|
|
'RATE1': 'RATE',
|
|
'LOOPSTAT': 'MD',
|
|
}
|
|
|
|
MD_READ_OFFSET = 0xBE
|
|
MD_WRITE_OFFSET = 0xBA
|
|
|
|
SIGNAL_TAG_BASE = 0x2000
|
|
MATH_VAR_BASE = 0x18C0
|
|
|
|
|
|
def loop_base(n: int) -> int:
|
|
if 1 <= n <= 24:
|
|
return 0x0040 + (n - 1) * 0x0100
|
|
if 25 <= n <= 32:
|
|
return 0x7840 + (n - 25) * 0x0100
|
|
raise ValueError(f'loop number {n} out of range 1-32')
|
|
|
|
|
|
# ─────────────────────────── archive flag logic ───────────────────────────
|
|
|
|
def should_archive(kind: str, param_name: str | None, is_signal_tag: bool) -> bool:
|
|
"""Determine whether this tag should be archived to history_table.
|
|
|
|
Rules (docs/베이직아키텍처-태그-디자인-전면재설계.md §7.2):
|
|
- Loop PV/SP/OP/MD/QV → True
|
|
- Non-loop AnalogPoint/StatusPoint TAG source (PV) → True
|
|
- Loop other params (GAIN, RESET, AL1SP1, DEV, ...) → False
|
|
- Variable (MATH_VAR) → False
|
|
- FlexibleParameter → False (except QV handled in loop entry)
|
|
"""
|
|
if kind == 'loop':
|
|
return param_name in ('PV', 'SP', 'OP', 'MD', 'QV')
|
|
if kind == 'tag':
|
|
return True
|
|
return False
|
|
|
|
|
|
# ─────────────────────── Sinam (Experion export) parsing ───────────────────────
|
|
|
|
_RE_LOOP = re.compile(r'^C(\d+)\s+(LOOPX?)\s+(\d+)\s+(\w+)\s*$', re.I)
|
|
_RE_TAG = re.compile(r'^C(\d+)\s+TAG\s+(\d+)\s+VALUE\s*$', re.I)
|
|
_RE_VAR = re.compile(r'^C(\d+)\s+MATH_VAR\s+(\d+)\s+VALUE\s*$', re.I)
|
|
_RE_RAW = re.compile(r'^C(\d+)\s+(\d+):0x([0-9a-fA-F]+)(?:\s+(\w+))?\s*$', re.I)
|
|
|
|
|
|
def _raw_type(fmt: str | None) -> tuple[int, str]:
|
|
f = (fmt or '').upper()
|
|
if f in ('', 'IEEEFP'):
|
|
return 2, 'float32'
|
|
return 1, 'uint16'
|
|
|
|
|
|
def scan_point(cells: list[str], controller_num: str) -> dict | None:
|
|
"""Classify an Experion point from all of its non-empty cells.
|
|
|
|
Priority: LOOP/LOOPX > TAG > MATH_VAR > raw.
|
|
"""
|
|
loop_refs: list[tuple[int, str]] = []
|
|
tag_n = var_n = None
|
|
raw = None
|
|
|
|
for cell in cells:
|
|
s = cell.strip()
|
|
m = _RE_LOOP.match(s)
|
|
if m and m.group(1) == controller_num:
|
|
loop_refs.append((int(m.group(3)), m.group(4).upper()))
|
|
continue
|
|
m = _RE_TAG.match(s)
|
|
if m and m.group(1) == controller_num:
|
|
tag_n = int(m.group(2))
|
|
continue
|
|
m = _RE_VAR.match(s)
|
|
if m and m.group(1) == controller_num:
|
|
var_n = int(m.group(2))
|
|
continue
|
|
m = _RE_RAW.match(s)
|
|
if m and m.group(1) == controller_num:
|
|
count, dtype = _raw_type(m.group(4))
|
|
raw = {'addr': int(m.group(3), 16), 'table': int(m.group(2)),
|
|
'count': count, 'type': dtype}
|
|
continue
|
|
|
|
if loop_refs:
|
|
counts = Counter(n for n, _ in loop_refs)
|
|
pv_loops = [n for n, mn in loop_refs if mn == 'PV']
|
|
primary = (max(pv_loops, key=lambda n: counts[n]) if pv_loops
|
|
else counts.most_common(1)[0][0])
|
|
mnems = {mn for n, mn in loop_refs if n == primary}
|
|
return {'kind': 'loop', 'n': primary, 'mnemonics': mnems}
|
|
if tag_n is not None:
|
|
return {'kind': 'tag', 'n': tag_n}
|
|
if var_n is not None:
|
|
return {'kind': 'var', 'n': var_n}
|
|
if raw is not None:
|
|
return {'kind': 'raw', **raw}
|
|
return None
|
|
|
|
|
|
def build_loop_entries(item_name: str, n: int, mnemonics: set[str]) -> list[dict]:
|
|
"""Register every parameter of loop n as individual entries.
|
|
|
|
Each entry carries an ``archive`` flag (True only for PV, SP, OP, MD, QV).
|
|
.MD addr=LoopStatus(0xBE), write_addr=AutoManState(0xBA).
|
|
"""
|
|
base = loop_base(n)
|
|
|
|
offset_attr: dict[int, str] = {}
|
|
for mn in mnemonics:
|
|
attr = EXPERION_ATTR.get(mn)
|
|
if attr and mn in MNEMONIC_OFFSET:
|
|
offset_attr[MNEMONIC_OFFSET[mn]] = attr
|
|
has_mode = 'LOOPSTAT' in mnemonics or 'MODEIN' in mnemonics
|
|
|
|
entries = []
|
|
for off, (suffix, count, dtype, access) in sorted(LOOP_LAYOUT.items()):
|
|
if off == MD_READ_OFFSET and has_mode:
|
|
continue
|
|
name = offset_attr.get(off, suffix)
|
|
entries.append({
|
|
'tag': f'{item_name}.{name}',
|
|
'addr': base + off,
|
|
'write_addr': base + off,
|
|
'count': count,
|
|
'type': dtype,
|
|
'access': access,
|
|
'description': f'LOOP #{n} {suffix}',
|
|
'archive': should_archive('loop', name, False),
|
|
})
|
|
|
|
if has_mode:
|
|
entries.append({
|
|
'tag': f'{item_name}.MD',
|
|
'addr': base + MD_READ_OFFSET,
|
|
'write_addr': base + MD_WRITE_OFFSET,
|
|
'count': 1,
|
|
'type': 'uint16',
|
|
'access': 'R',
|
|
'description': f'LOOP #{n} Mode status (read=LoopStatus 0xBE, write=AutoManState 0xBA)',
|
|
'archive': True,
|
|
})
|
|
return entries
|
|
|
|
|
|
def build_point_entry(item_name: str, desc: dict) -> dict | None:
|
|
"""Build a register entry for a signal-tag / variable / raw point."""
|
|
kind = desc['kind']
|
|
if kind == 'tag':
|
|
addr = SIGNAL_TAG_BASE + (desc['n'] - 1) * 2
|
|
return {
|
|
'tag': item_name, 'addr': addr, 'write_addr': addr,
|
|
'count': 2, 'type': 'float32', 'access': 'R',
|
|
'description': f'Signal Tag #{desc["n"]}',
|
|
'archive': should_archive('tag', None, True),
|
|
}
|
|
if kind == 'var':
|
|
addr = MATH_VAR_BASE + (desc['n'] - 1) * 2
|
|
return {
|
|
'tag': item_name, 'addr': addr, 'write_addr': addr,
|
|
'count': 2, 'type': 'float32', 'access': 'RW',
|
|
'description': f'Variable (MATH_VAR) #{desc["n"]}',
|
|
'archive': False,
|
|
}
|
|
if kind == 'raw':
|
|
if desc['table'] != 4:
|
|
print(f' \u26a0 {item_name}: non-named table {desc["table"]} (not holding registers) \u2014 skipping')
|
|
return None
|
|
return {
|
|
'tag': item_name, 'addr': desc['addr'], 'write_addr': desc['addr'],
|
|
'count': desc['count'], 'type': desc['type'], 'access': 'R',
|
|
'description': f'Custom address 0x{desc["addr"]:04X}',
|
|
'archive': False,
|
|
}
|
|
return None
|
|
|
|
|
|
def resolve_one(s: str, controller_num: str) -> dict | None:
|
|
"""Resolve a single source-address for FlexibleParameters."""
|
|
s = s.strip()
|
|
m = _RE_LOOP.match(s)
|
|
if m and m.group(1) == controller_num:
|
|
n, mn = int(m.group(3)), m.group(4).upper()
|
|
off = MNEMONIC_OFFSET.get(mn)
|
|
if off is None:
|
|
return None
|
|
if off in LOOP_LAYOUT:
|
|
_suf, count, dtype, access = LOOP_LAYOUT[off]
|
|
else:
|
|
count, dtype, access = 2, 'float32', 'RW'
|
|
return {'addr': loop_base(n) + off, 'count': count, 'type': dtype, 'access': access}
|
|
m = _RE_TAG.match(s)
|
|
if m and m.group(1) == controller_num:
|
|
return {'addr': SIGNAL_TAG_BASE + (int(m.group(2)) - 1) * 2,
|
|
'count': 2, 'type': 'float32', 'access': 'R'}
|
|
m = _RE_VAR.match(s)
|
|
if m and m.group(1) == controller_num:
|
|
return {'addr': MATH_VAR_BASE + (int(m.group(2)) - 1) * 2,
|
|
'count': 2, 'type': 'float32', 'access': 'RW'}
|
|
m = _RE_RAW.match(s)
|
|
if m and m.group(1) == controller_num and int(m.group(2)) == 4:
|
|
count, dtype = _raw_type(m.group(4))
|
|
return {'addr': int(m.group(3), 16), 'count': count, 'type': dtype, 'access': 'R'}
|
|
return None
|
|
|
|
|
|
# ─────────────────────── metadata helpers ───────────────────────
|
|
|
|
def _s(val) -> str:
|
|
"""Safely stringify a cell value."""
|
|
return str(val).strip() if val is not None else ''
|
|
|
|
|
|
def _get_desc(row: list, col: dict) -> str:
|
|
"""ItemDescription or DownloadedName."""
|
|
for key in ('DownloadedName', 'ItemDescription'):
|
|
idx = col.get(key)
|
|
if idx is not None:
|
|
v = _s(row[idx])
|
|
if v:
|
|
return v
|
|
return ''
|
|
|
|
|
|
def _upsert_tag_metadata(conn, base_tag: str, controller_id: str,
|
|
attr: str, value: str) -> None:
|
|
if not value:
|
|
return
|
|
with conn.cursor() as cur:
|
|
cur.execute("""
|
|
INSERT INTO tag_metadata (base_tag, attribute, value, controller_id)
|
|
VALUES (%s, %s, %s, %s)
|
|
ON CONFLICT (base_tag, attribute, controller_id)
|
|
DO UPDATE SET value = EXCLUDED.value, loaded_at = NOW()
|
|
""", (base_tag, attr, value, controller_id))
|
|
|
|
|
|
def _upsert_map_master(conn, tagname: str, hc900_tag: str, modbus_addr: int,
|
|
data_type: str, access: str, controller_id: str,
|
|
realtime_enabled: bool, archive_enabled: bool) -> None:
|
|
# param_type = tagname suffix (PV/SP/OP/MD/QV/Gain1/...). 모든 태그가 {base}.{param}
|
|
# 형식이므로 마지막 '.' 뒤가 param. UI 필터(paramType)가 이 컬럼을 쓴다.
|
|
param_type = tagname.rsplit('.', 1)[1] if '.' in tagname else ''
|
|
with conn.cursor() as cur:
|
|
cur.execute("""
|
|
INSERT INTO hc900_map_master
|
|
(tagname, hc900_tag, modbus_addr, data_type, access,
|
|
controller_id, is_active, realtime_enabled, archive_enabled, param_type)
|
|
VALUES (%s, %s, %s, %s, %s, %s, TRUE, %s, %s, %s)
|
|
ON CONFLICT (controller_id, tagname)
|
|
DO UPDATE SET
|
|
hc900_tag = EXCLUDED.hc900_tag,
|
|
modbus_addr = EXCLUDED.modbus_addr,
|
|
data_type = EXCLUDED.data_type,
|
|
access = EXCLUDED.access,
|
|
is_active = TRUE,
|
|
realtime_enabled = EXCLUDED.realtime_enabled,
|
|
archive_enabled = EXCLUDED.archive_enabled,
|
|
param_type = EXCLUDED.param_type
|
|
""", (tagname, hc900_tag, modbus_addr, data_type, access,
|
|
controller_id, realtime_enabled, archive_enabled, param_type))
|
|
|
|
|
|
def _build_db_conn(dsn: str):
|
|
"""Lazy psycopg2 import + connect."""
|
|
import psycopg2
|
|
return psycopg2.connect(dsn)
|
|
|
|
|
|
# ─────────────────────── main build ───────────────────────
|
|
|
|
def build_registers(sinam_path: Path, controller: str,
|
|
db_conn_str: str | None) -> list[dict]:
|
|
controller_num = controller.lstrip('Cc')
|
|
wb = openpyxl.load_workbook(sinam_path, read_only=True, data_only=True)
|
|
ws = wb['Sheet1']
|
|
db_conn = _build_db_conn(db_conn_str) if db_conn_str else None
|
|
|
|
registers: list[dict] = []
|
|
seen_tags: set[str] = set()
|
|
seen_loops: set[int] = set()
|
|
meta_upserted: set[str] = set()
|
|
|
|
# Sinam_Tag_all.xlsx 는 다중 섹션 구조다. 첫 셀이 'ItemName'/'ParentItemName' 인
|
|
# 행이 헤더이고, 그 다음 행들은 그 헤더의 컬럼 레이아웃을 따른다. Class(StatusPoint/
|
|
# AnalogPoint)·FlexibleParameters 블록마다 컬럼 위치가 다르고, 등록 태그 수에 따라
|
|
# 행 범위가 가변하므로 행번호로 고정하지 않고 "헤더를 만날 때마다" 컬럼맵을 갱신해
|
|
# 항상 현재 헤더의 컬럼명으로 값을 읽는다.
|
|
header: dict[str, int] = {}
|
|
section: str | None = None # 'item' | 'parent'
|
|
|
|
def gv(r, name):
|
|
i = header.get(name)
|
|
if i is None or i >= len(r):
|
|
return ''
|
|
return _s(r[i])
|
|
|
|
def add_entry(tag, addr, write_addr, count, dtype, access,
|
|
desc_txt, archive, realtime=True):
|
|
if tag in seen_tags:
|
|
return
|
|
seen_tags.add(tag)
|
|
registers.append({
|
|
'tag': tag, 'addr': addr, 'write_addr': write_addr,
|
|
'count': count, 'type': dtype, 'access': access,
|
|
'description': desc_txt, 'archive': archive,
|
|
})
|
|
if db_conn:
|
|
_upsert_map_master(db_conn, tag, tag, addr, dtype, access, controller,
|
|
realtime_enabled=realtime, archive_enabled=archive)
|
|
|
|
def upsert_meta(base, attr, value):
|
|
if db_conn and value:
|
|
_upsert_tag_metadata(db_conn, base, controller, attr, value)
|
|
|
|
for row in ws.iter_rows(min_row=1, values_only=True):
|
|
c0 = row[0]
|
|
if c0 in ('ItemName', 'ParentItemName'):
|
|
header = {str(h).strip(): i for i, h in enumerate(row) if h is not None}
|
|
section = 'parent' if c0 == 'ParentItemName' else 'item'
|
|
continue
|
|
if not c0 or section is None:
|
|
continue
|
|
item_name = str(c0).strip()
|
|
cls = gv(row, 'Class')
|
|
|
|
# ── ParentItemName 섹션: FlexibleParameters → {parent}.{ParamName} ──
|
|
# HistoryParameters 블록은 SourceAddressPVUDSP 컬럼이 없어 자동 스킵된다.
|
|
if section == 'parent':
|
|
param = gv(row, 'ParamName')
|
|
src = gv(row, 'SourceAddressPVUDSP')
|
|
if not param or not src:
|
|
continue
|
|
res = resolve_one(src, controller_num)
|
|
if not res:
|
|
continue
|
|
dst = gv(row, 'DestinationAddressPVUDSP')
|
|
access = 'RW' if dst else res['access']
|
|
write_addr = res['addr']
|
|
if dst:
|
|
dres = resolve_one(dst, controller_num)
|
|
if dres:
|
|
write_addr = dres['addr']
|
|
tag = f'{item_name}.{param}'
|
|
archive = (param.upper() == 'QV')
|
|
# FlexibleParameter 는 기본적으로 폴링 대상에서 제외(realtime_enabled=False).
|
|
add_entry(tag, res['addr'], write_addr, res['count'], res['type'],
|
|
access, f'FlexibleParameter {param}', archive, realtime=False)
|
|
upsert_meta(tag, 'units', gv(row, 'UnitsUDSP'))
|
|
upsert_meta(tag, 'eulo', gv(row, 'RangeLowUDSP'))
|
|
upsert_meta(tag, 'euhi', gv(row, 'RangeHighUDSP'))
|
|
for i in range(8):
|
|
upsert_meta(tag, f'state{i}', gv(row, f'DescriptorState{i}UDSP'))
|
|
continue
|
|
|
|
# ── ItemName 섹션: AnalogPoint / StatusPoint ──
|
|
if cls not in ('AnalogPoint', 'StatusPoint'):
|
|
continue
|
|
pv_src = gv(row, 'SourceAddressPV')
|
|
sp_src = gv(row, 'SourceAddressSP')
|
|
op_src = gv(row, 'SourceAddressOP')
|
|
md_src = gv(row, 'SourceAddressMD')
|
|
is_analog = (cls == 'AnalogPoint')
|
|
|
|
# 루프 판정 + mnemonic 수집은 행의 모든 셀을 스캔(scan_point)한다. GAIN1/RESET1/
|
|
# RATE1 등 SourceAddressPV/SP/OP/MD 외 컬럼에 흩어진 파라미터까지 포착해 doc 5.2
|
|
# 의 Experion 속성명(.GAIN/.RESET/.RATE/.MD)을 정확히 매핑하기 위함.
|
|
cells = [str(v) for v in row if v is not None]
|
|
desc = scan_point(cells, controller_num)
|
|
if desc and desc['kind'] == 'loop':
|
|
if desc['n'] in seen_loops:
|
|
continue
|
|
seen_loops.add(desc['n'])
|
|
for e in build_loop_entries(item_name, desc['n'], desc['mnemonics']):
|
|
add_entry(e['tag'], e['addr'], e['write_addr'], e['count'],
|
|
e['type'], e['access'], e['description'], e['archive'])
|
|
produced = True
|
|
else:
|
|
# 신호/상태 포인트: PV=bare(태그명 자체), SP/OP/MD=suffix 엔트리.
|
|
# 빈칸이면 생성 안 함, 주소가 있으면 생성(빈칸/주소 판정은 소스 컬럼 유무).
|
|
produced = False
|
|
if pv_src:
|
|
r = resolve_one(pv_src, controller_num)
|
|
if r:
|
|
# 일관성: PV 도 항상 .PV suffix (bare 금지). 모든 레지스터는
|
|
# {base}.{param} 형식. AnalogPoint PV = 연속 측정값 → archive,
|
|
# StatusPoint PV = 디지털 상태(state 라벨) → archive 제외(event_history).
|
|
add_entry(f'{item_name}.PV', r['addr'], r['addr'], r['count'],
|
|
r['type'], r['access'], f'{cls} PV', is_analog)
|
|
produced = True
|
|
for attr, src, dstname in (('SP', sp_src, 'DestinationAddressSP'),
|
|
('OP', op_src, 'DestinationAddressOP'),
|
|
('MD', md_src, 'DestinationAddressMD')):
|
|
if not src:
|
|
continue
|
|
r = resolve_one(src, controller_num)
|
|
if not r:
|
|
continue
|
|
dst = gv(row, dstname)
|
|
access = 'RW' if dst else r['access']
|
|
write_addr = r['addr']
|
|
if dst:
|
|
dres = resolve_one(dst, controller_num)
|
|
if dres:
|
|
write_addr = dres['addr']
|
|
# StatusPoint(디지털 상태점)의 .SP/.OP/.MD 도 이산값 → archive 제외
|
|
# (event_history 로 감). AnalogPoint 의 .SP/.OP/.MD 만 archive.
|
|
add_entry(f'{item_name}.{attr}', r['addr'], write_addr, r['count'],
|
|
r['type'], access, f'{cls} {attr}', is_analog)
|
|
produced = True
|
|
|
|
# ── 보조 파라미터 A1~A4 (모든 AnalogPoint, 루프 포함) ──────────────────────
|
|
# FIQ 류 적산기는 QV(적산)가 FlexibleParameter가 아니라 AnalogPoint 보조슬롯
|
|
# (AuxiliaryNameA{n} ↔ SourceAddressA{n})에 노출된다. 슬롯명이 있는 A1~A4를
|
|
# 일반적으로 모두 어드레싱(공백 슬롯만 제외). 루프 엔트리와 겹치면 seen_tags가
|
|
# 중복을 흡수. QV=연속 누적값→archive. 보조도 폴링 대상(realtime 기본 True).
|
|
if is_analog:
|
|
for n in (1, 2, 3, 4):
|
|
aux_name = gv(row, f'AuxiliaryNameA{n}')
|
|
aux_src = gv(row, f'SourceAddressA{n}')
|
|
if not aux_name or not aux_src: # 공백 슬롯 제외
|
|
continue
|
|
r = resolve_one(aux_src, controller_num)
|
|
if not r:
|
|
continue
|
|
dst = gv(row, f'DestinationAddressA{n}')
|
|
access = 'RW' if dst else r['access']
|
|
write_addr = r['addr']
|
|
if dst:
|
|
dres = resolve_one(dst, controller_num)
|
|
if dres:
|
|
write_addr = dres['addr']
|
|
add_entry(f'{item_name}.{aux_name}', r['addr'], write_addr, r['count'],
|
|
r['type'], access, f'{cls} aux {aux_name}', aux_name.upper() == 'QV')
|
|
produced = True
|
|
|
|
# 베이스 태그 메타데이터 (이 컨트롤러에 속해 엔트리가 생성된 경우 1회)
|
|
if produced and db_conn and item_name not in meta_upserted:
|
|
meta_upserted.add(item_name)
|
|
upsert_meta(item_name, 'desc',
|
|
gv(row, 'DownloadedName') or gv(row, 'ItemDescription'))
|
|
upsert_meta(item_name, 'area', gv(row, 'AreaCode'))
|
|
upsert_meta(item_name, 'units', gv(row, 'Units'))
|
|
upsert_meta(item_name, 'eulo', gv(row, 'RangeLow'))
|
|
upsert_meta(item_name, 'euhi', gv(row, 'RangeHigh'))
|
|
for i in range(8):
|
|
upsert_meta(item_name, f'state{i}', gv(row, f'DescriptorState{i}'))
|
|
|
|
wb.close()
|
|
|
|
# ── 흡수(dedup) ──────────────────────────────────────────────────────────
|
|
# 짧은 지시계 prefix(>=2자)가 같은 번호의 더 긴 prefix(그 prefix로 시작하는
|
|
# 컨트롤러/적산계, 동일 측정)에 흡수된다. 예) FI-6101 → FICQ-6101/FIQ-6101 존재 시
|
|
# 제거, LI-6211 → LICA-6211, PI → PICA. P-(펌프, 1자)는 >=2자 제한으로 보호.
|
|
def _pfx_num(tag):
|
|
# base 가 '{문자}-{숫자}' 또는 '{문자}-{숫자}{단일문자}' 형태일 때 흡수 후보.
|
|
# 단일 trailing letter(다점/리보일러 지시계: TI-6111A↔TICA-6111A)는 번호키에
|
|
# 포함해 같은 suffix끼리만 그룹화 → TI-6111A 가 TICA-6111A 로 흡수된다.
|
|
# 인터록/복합 suffix(LIC-9113-IL-RST, TICA-6111A-HI-IL 등)는 fullmatch로 제외.
|
|
b = tag.split('.', 1)[0]
|
|
m = re.fullmatch(r'([A-Za-z]+)-(\d+[A-Za-z]?)', b)
|
|
return (m.group(1), m.group(2), b) if m else (None, None, b)
|
|
num_prefixes: dict[str, set[str]] = {}
|
|
for r in registers:
|
|
p, n, _ = _pfx_num(r['tag'])
|
|
if p and n:
|
|
num_prefixes.setdefault(n, set()).add(p)
|
|
absorbed_bases: set[str] = set()
|
|
for r in registers:
|
|
p, n, b = _pfx_num(r['tag'])
|
|
if p and n and len(p) >= 2 and any(q != p and q.startswith(p) for q in num_prefixes[n]):
|
|
absorbed_bases.add(b)
|
|
if absorbed_bases:
|
|
registers = [r for r in registers
|
|
if r['tag'].split('.', 1)[0] not in absorbed_bases]
|
|
print(f' 흡수(중복 신호점 제거): {len(absorbed_bases)}개 base '
|
|
f'(예: {sorted(absorbed_bases)[:6]})')
|
|
|
|
if db_conn:
|
|
if absorbed_bases:
|
|
with db_conn.cursor() as cur:
|
|
for b in absorbed_bases:
|
|
cur.execute("DELETE FROM hc900_map_master WHERE controller_id=%s "
|
|
"AND split_part(tagname,'.',1)=%s", (controller, b))
|
|
cur.execute("DELETE FROM tag_metadata WHERE controller_id=%s "
|
|
"AND split_part(base_tag,'.',1)=%s", (controller, b))
|
|
db_conn.commit()
|
|
db_conn.close()
|
|
print(f' DB upsert: {len(meta_upserted)} base tags metadata, '
|
|
f'{len(registers)} map_master entries (흡수 {len(absorbed_bases)} base)')
|
|
|
|
registers.sort(key=lambda r: r['addr'])
|
|
return registers, absorbed_bases
|
|
|
|
|
|
# ─────────────────────── CSV cross-check ───────────────────────
|
|
|
|
def validate_against_csv(csv_path: Path) -> None:
|
|
rows = list(csv.reader(open(csv_path, encoding='utf-8-sig')))
|
|
custom = any(len(r) > 1 and r[0].strip() == 'Hex Addr' and 'Partition Name' in r
|
|
for r in rows)
|
|
tag_col = 3 if custom else 2
|
|
type_col = 5 if custom else 4
|
|
|
|
def norm(s):
|
|
return re.sub(r'[^a-z0-9]', '', s.lower())
|
|
|
|
expected = {off: norm(suf) for off, (suf, *_) in LOOP_LAYOUT.items()}
|
|
aliases = {0x3e: {'opwork', 'outputb'}, 0x38: {'wspb', 'spwork'},
|
|
0x0a: {'carbonpottemp', 'temp'}, 0x14: {'cycletime1', 'scancycletime'},
|
|
0x28: {'cycletime2', 'cycletimescan', 'scancycletimeb'},
|
|
0x52: {'localpctco', 'localpercentcarbmonoxide'},
|
|
0x58: {'onoffhysteresis', 'onoffouthysterisis'},
|
|
0x5a: {'carbpotdewpt', 'carbpotdewpt'}, 0x5c: {'stepmotortime', '3posstepmotortime'},
|
|
0x22: {'stepdeadband', '3posstepdeadband'},
|
|
0xbb: {'spselectstate', 'lspselectstate'},
|
|
0xbe: {'loopstatus', 'loopstatusregister'},
|
|
0xb9: {'antisootenable', 'antisootsplimenable'},
|
|
0xb7: {'fuzzyenable', 'enabledisablefuzzy'},
|
|
0xb8: {'demandtune', 'demandtunereq'}, 0x4e: {'manualreset'},
|
|
0x50: {'feedforwardgain'}, 0x08: {'pvb', 'pv'}}
|
|
mism = 0
|
|
for r in rows:
|
|
if len(r) <= type_col or not r[0].startswith('0x'):
|
|
continue
|
|
if r[type_col].strip() != 'PID':
|
|
continue
|
|
addr = int(r[0], 16)
|
|
off = (addr - 0x40) % 0x100
|
|
if off not in LOOP_LAYOUT:
|
|
continue
|
|
suff = r[tag_col].split('.', 1)[1] if '.' in r[tag_col] else r[tag_col]
|
|
got = norm(suff)
|
|
ok = got == expected[off] or got in aliases.get(off, set()) \
|
|
or expected[off] in got or got in expected[off]
|
|
if not ok:
|
|
mism += 1
|
|
if mism <= 12:
|
|
print(f' \u26a0 offset 0x{off:02X}: layout={expected[off]!r} '
|
|
f'csv={got!r} ({r[0]})')
|
|
print(f' CSV cross-check: {"OK" if mism == 0 else f"{mism} mismatch(es)"} '
|
|
f'({"Custom" if custom else "Fixed"} map)')
|
|
|
|
|
|
# ─────────────────────── main ───────────────────────
|
|
|
|
def build(sinam_path: Path, controller: str, output_path: Path,
|
|
validate_csv: Path | None, db_conn_str: str | None) -> None:
|
|
if validate_csv:
|
|
print(f'Validating embedded loop layout against {validate_csv}...')
|
|
validate_against_csv(validate_csv)
|
|
|
|
print(f'Parsing {controller} points from {sinam_path}...')
|
|
registers, absorbed_bases = build_registers(sinam_path, controller, db_conn_str)
|
|
|
|
output = {
|
|
'controller': controller,
|
|
'report_generated': datetime.date.today().isoformat(),
|
|
'float_format': 'FP_B',
|
|
'notes': f'Register map built from {sinam_path.name} ({controller} only). '
|
|
'Experion point names are keys; .MD reads LoopStatus(0xBE) and '
|
|
'writes AutoManState(0xBA). archive flag indicates history target.',
|
|
'register_count': len(registers),
|
|
'registers': registers,
|
|
}
|
|
output_path.write_text(json.dumps(output, indent=2, ensure_ascii=False),
|
|
encoding='utf-8')
|
|
|
|
n_loops = len({r['description'].split()[1] for r in registers
|
|
if r['description'].startswith('LOOP')})
|
|
n_archive = sum(1 for r in registers if r.get('archive'))
|
|
print(f'\n\u2713 Wrote {output_path}')
|
|
print(f' {len(registers)} registers ({n_loops} loops expanded)')
|
|
print(f' archive=true: {n_archive}')
|
|
by_access = {}
|
|
for r in registers:
|
|
by_access[r['access']] = by_access.get(r['access'], 0) + 1
|
|
print(f' by access: {by_access}')
|
|
|
|
# machine-readable JSON summary for C# caller
|
|
print(f'SINAM_SUMMARY:{json.dumps({"registers": len(registers), "archive_true": n_archive, "absorbed": len(absorbed_bases), "loops": n_loops}, ensure_ascii=False)}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
p = argparse.ArgumentParser(
|
|
description='Build register-map-cN.json from Sinam_Tag_all.xlsx alone')
|
|
p.add_argument('--controller', required=True,
|
|
help='Experion controller prefix, e.g. C3')
|
|
p.add_argument('--sinam', required=True,
|
|
help='Path to Sinam_Tag_all.xlsx')
|
|
p.add_argument('-o', '--output', default='docs/register-map.json',
|
|
help='Output JSON path')
|
|
p.add_argument('--validate-csv', default=None,
|
|
help='Optional HC Designer CSV to cross-check loop layout')
|
|
p.add_argument('--db-conn', default=None,
|
|
help='PostgreSQL DSN; if set, upsert tag_metadata + hc900_map_master')
|
|
args = p.parse_args()
|
|
|
|
build(Path(args.sinam), args.controller.upper(), Path(args.output),
|
|
Path(args.validate_csv) if args.validate_csv else None,
|
|
args.db_conn)
|