224 lines
7.5 KiB
Python
224 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Import field history data from field_hist DB into hc900.history_table
|
|
for C3 and C4 controller tags only. 2026-02-05 ~ 2026-06-05 (~30s interval).
|
|
|
|
Usage:
|
|
# Preview-only
|
|
python3 scripts/import_field_hist_C3_C4.py --dry-run
|
|
|
|
# Full import
|
|
python3 scripts/import_field_hist_C3_C4.py
|
|
|
|
# Partial date range
|
|
python3 scripts/import_field_hist_C3_C4.py --from-date 2026-03-01 --to-date 2026-03-07
|
|
|
|
Logic:
|
|
1. Load C3/C4 base tag names from `register-map-c3.json` / `register-map-c4.json`.
|
|
2. Decode field_hist ptlist+mapping+tblist → {tagname_upper: (tblname, oit)}.
|
|
3. Filter to C3/C4 (by base name match), tag names stored UPPERCASE.
|
|
4. For each cont table in daily chunks: DELETE existing history overlap,
|
|
then COPY new long-format data.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import psycopg
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
REG_MAP_DIR = Path(__file__).resolve().parent.parent / "docs"
|
|
|
|
FIELD_DSN = "host=localhost port=5432 dbname=field_hist user=postgres password=postgres"
|
|
TARGET_DSN = (
|
|
"host=localhost port=5432 dbname=iiot_platform "
|
|
"user=postgres password=postgres options=-csearch_path=hc900"
|
|
)
|
|
|
|
BATCH_DAYS = 1
|
|
|
|
|
|
def load_controller_bases() -> dict[str, str]:
|
|
"""Return {base_name_lower: controller_id} from register-map-c3/c4.json."""
|
|
result = {}
|
|
for ctrl in ("C3", "C4"):
|
|
path = REG_MAP_DIR / f"register-map-{ctrl.lower()}.json"
|
|
data = json.loads(path.read_text())
|
|
for entry in data["registers"]:
|
|
base = entry["tag"].split(".")[0].lower()
|
|
if base not in result:
|
|
result[base] = ctrl
|
|
return result
|
|
|
|
|
|
def build_tag_map(allowed_bases: dict[str, str]) -> dict[str, tuple[str, int, str]]:
|
|
"""Return {tagname_upper: (tblname, oit, controller_id)} for C3/C4 tags."""
|
|
tag_map = {}
|
|
with psycopg.connect(FIELD_DSN) as conn:
|
|
cur = conn.execute("""
|
|
SELECT p.shortptname, t.tblname, m.oit
|
|
FROM ptlist p
|
|
JOIN mapping m ON m.pid = p.pid
|
|
JOIN tblist t ON t.tid = m.tid
|
|
WHERE p.shortptname IS NOT NULL
|
|
""")
|
|
for short, tbl, oit in cur:
|
|
upper = short.strip().upper()
|
|
base = upper.split(".")[0].lower()
|
|
ctrl = allowed_bases.get(base)
|
|
if ctrl is not None:
|
|
tag_map[upper] = (tbl, int(oit), ctrl)
|
|
return tag_map
|
|
|
|
|
|
def process_all(dry_run: bool, from_date: str | None, to_date: str | None) -> int:
|
|
"""Main processing loop."""
|
|
print("Loading controller tag bases from register maps...", flush=True)
|
|
bases = load_controller_bases()
|
|
c3_ct = sum(1 for v in bases.values() if v == "C3")
|
|
c4_ct = sum(1 for v in bases.values() if v == "C4")
|
|
print(f" C3: {c3_ct} base tags, C4: {c4_ct} base tags", flush=True)
|
|
|
|
print("Building tag map from field_hist...", flush=True)
|
|
tag_map = build_tag_map(bases)
|
|
c3_tags = sum(1 for _, _, c in tag_map.values() if c == "C3")
|
|
c4_tags = sum(1 for _, _, c in tag_map.values() if c == "C4")
|
|
print(f" Matched: C3={c3_tags}, C4={c4_tags} (total={len(tag_map)})", flush=True)
|
|
|
|
if not tag_map:
|
|
print("ERROR: no tags matched — aborting", flush=True)
|
|
return 0
|
|
|
|
# Group by cont table
|
|
groups: dict[str, list[tuple[int, str, str]]] = defaultdict(list)
|
|
for tagname, (tbl, oit, ctrl) in tag_map.items():
|
|
groups[tbl].append((oit, tagname, ctrl))
|
|
|
|
print(f" Across {len(groups)} cont tables", flush=True)
|
|
|
|
# Load tblist for ordered processing
|
|
with psycopg.connect(FIELD_DSN) as conn:
|
|
cur = conn.execute("SELECT tblname FROM tblist ORDER BY tid")
|
|
tbl_order = [r[0] for r in cur]
|
|
|
|
# Time range
|
|
t_from = (
|
|
datetime.strptime(from_date, "%Y-%m-%d")
|
|
if from_date
|
|
else datetime(2026, 2, 5)
|
|
)
|
|
t_to = (
|
|
datetime.strptime(to_date, "%Y-%m-%d") if to_date else datetime(2026, 6, 6)
|
|
)
|
|
|
|
total_rows = 0
|
|
for tbl in tbl_order:
|
|
col_info = groups.get(tbl)
|
|
if not col_info:
|
|
continue
|
|
|
|
# Build dynamic SELECT for mapped columns
|
|
oits = sorted({oit for oit, _, _ in col_info})
|
|
oit_map: dict[int, list[tuple[str, str]]] = defaultdict(list)
|
|
for oit, tag, ctrl in col_info:
|
|
oit_map[oit].append((tag, ctrl))
|
|
|
|
col_sel = ", ".join(f"col{c:02d}" for c in oits)
|
|
sql = f"SELECT dtat, {col_sel} FROM {tbl} WHERE dtat >= %s AND dtat < %s ORDER BY dtat"
|
|
|
|
print(f"\n{tbl} ({len(col_info)} cols, {len(oits)} unique oits) ...", flush=True)
|
|
|
|
with psycopg.connect(FIELD_DSN) as src_conn:
|
|
with src_conn.cursor(name="fetch_cursor") as src_cur:
|
|
src_cur.itersize = 50000
|
|
src_cur.execute(sql, (t_from, t_to))
|
|
|
|
batch = []
|
|
|
|
for row in src_cur:
|
|
dtat, *vals = row
|
|
recorded_at = dtat - timedelta(hours=9)
|
|
for i, val in enumerate(vals):
|
|
if val is None:
|
|
continue
|
|
oit = oits[i]
|
|
for tagname, ctrl_id in oit_map[oit]:
|
|
batch.append((tagname, str(val), recorded_at, ctrl_id))
|
|
|
|
if len(batch) >= 50000:
|
|
total_rows += _flush_batch(batch, dry_run, tbl)
|
|
batch.clear()
|
|
|
|
if batch:
|
|
total_rows += _flush_batch(batch, dry_run, tbl)
|
|
|
|
return total_rows
|
|
|
|
|
|
def _flush_batch(rows: list, dry_run: bool, tbl: str) -> int:
|
|
"""DELETE overlap + COPY batch into history_table. Returns row count."""
|
|
n = len(rows)
|
|
if dry_run:
|
|
print(f" [{tbl}] {n:>8} rows (dry-run)", flush=True)
|
|
return n
|
|
|
|
# Collect unique tagnames + time range for DELETE
|
|
tagnames = list({r[0] for r in rows})
|
|
ts_min = min(r[2] for r in rows)
|
|
ts_max = max(r[2] for r in rows)
|
|
|
|
with psycopg.connect(TARGET_DSN) as tgt:
|
|
with tgt.cursor() as cur:
|
|
del_sql = """
|
|
DELETE FROM history_table
|
|
WHERE tagname = ANY(%s) AND recorded_at >= %s AND recorded_at <= %s
|
|
"""
|
|
cur.execute(del_sql, (tagnames, ts_min, ts_max))
|
|
deleted = cur.rowcount
|
|
|
|
with tgt.cursor() as cur:
|
|
with cur.copy(
|
|
"COPY history_table (tagname, value, recorded_at, controller_id) FROM STDIN"
|
|
) as copy:
|
|
for r in rows:
|
|
copy.write_row(r)
|
|
|
|
tgt.commit()
|
|
|
|
print(f" [{tbl}] {n:>8} inserted ({deleted} deleted)", flush=True)
|
|
return n
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(
|
|
description="Import field_hist data into hc900.history_table (C3/C4 only)"
|
|
)
|
|
p.add_argument(
|
|
"--dry-run", action="store_true", help="Preview only, no writes"
|
|
)
|
|
p.add_argument(
|
|
"--from-date", help="Start date (YYYY-MM-DD), default 2026-02-05"
|
|
)
|
|
p.add_argument(
|
|
"--to-date", help="End date (YYYY-MM-DD), default 2026-06-06"
|
|
)
|
|
args = p.parse_args()
|
|
|
|
t0 = time.time()
|
|
total = process_all(args.dry_run, args.from_date, args.to_date)
|
|
elapsed = time.time() - t0
|
|
|
|
print(f"\n{'=' * 60}")
|
|
label = "dry-run rows" if args.dry_run else "rows imported"
|
|
print(f"Total {label}: {total:,}")
|
|
print(f"Elapsed: {elapsed:.1f}s ({total / max(elapsed, 1):,.0f} rows/s)")
|
|
print(f"{'=' * 60}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|