139 lines
4.3 KiB
Python
139 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from dotenv import load_dotenv
|
|
import pymysql
|
|
|
|
from cm95pbf.s52 import (
|
|
conditional_procedures,
|
|
find_best_lookup,
|
|
primary_symbol_code,
|
|
)
|
|
|
|
|
|
DEFAULT_CLASSES = ("LIGHTS", "BCNLAT", "BOYLAT", "TOPMAR", "LNDMRK")
|
|
|
|
|
|
def _json_object(value: Any) -> dict[str, Any]:
|
|
if value is None:
|
|
return {}
|
|
if isinstance(value, str):
|
|
value = json.loads(value)
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Audit CM95 DB features against S-52 lookups.")
|
|
parser.add_argument("--lookups", required=True, type=Path)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
parser.add_argument("--env-file", default=Path(".env"), type=Path)
|
|
parser.add_argument("--dataset-id", default=1, type=int)
|
|
parser.add_argument("--table", default="Paper")
|
|
parser.add_argument("--classes", default=",".join(DEFAULT_CLASSES))
|
|
args = parser.parse_args()
|
|
|
|
rows = json.loads(args.lookups.read_text(encoding="utf-8"))
|
|
by_class: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for row in rows:
|
|
by_class[str(row.get("object_class") or "")].append(row)
|
|
|
|
load_dotenv(args.env_file)
|
|
conn = pymysql.connect(
|
|
host=os.environ["CM93_DB_HOST"],
|
|
port=int(os.environ.get("CM93_DB_PORT", "3306")),
|
|
user=os.environ["CM93_DB_USER"],
|
|
password=os.environ["CM93_DB_PASSWORD"],
|
|
database=os.environ["CM93_DB_NAME"],
|
|
charset=os.environ.get("CM93_DB_CHARSET", "utf8mb4"),
|
|
autocommit=True,
|
|
)
|
|
|
|
classes = [x.strip() for x in args.classes.split(",") if x.strip()]
|
|
summary: dict[str, Any] = {}
|
|
placeholders = ",".join(["%s"] * len(classes))
|
|
query = (
|
|
"SELECT object_class, attributes_json "
|
|
"FROM chart_feature "
|
|
"WHERE dataset_id=%s AND object_class IN (" + placeholders + ") "
|
|
"ORDER BY object_class, id"
|
|
)
|
|
|
|
with conn.cursor() as cur:
|
|
cur.execute(query, [args.dataset_id, *classes])
|
|
db_rows = cur.fetchall()
|
|
conn.close()
|
|
|
|
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for object_class, attrs_raw in db_rows:
|
|
grouped[str(object_class)].append(_json_object(attrs_raw))
|
|
|
|
for object_class in classes:
|
|
stats = Counter()
|
|
symbols = Counter()
|
|
csps = Counter()
|
|
rcids = Counter()
|
|
for attrs in grouped.get(object_class, []):
|
|
hit = find_best_lookup(
|
|
by_class.get(object_class, []),
|
|
attrs,
|
|
table_name=args.table,
|
|
geometry_type="Point",
|
|
)
|
|
if hit is None:
|
|
stats["no_lookup"] += 1
|
|
continue
|
|
|
|
stats["matched"] += 1
|
|
rcids[str(hit.get("rcid"))] += 1
|
|
|
|
symbol = primary_symbol_code(hit)
|
|
if symbol:
|
|
stats["static_symbol"] += 1
|
|
symbols[symbol] += 1
|
|
|
|
procedures = conditional_procedures(hit)
|
|
if procedures:
|
|
stats["conditional"] += 1
|
|
for procedure in procedures:
|
|
csps[procedure] += 1
|
|
|
|
if not symbol and not procedures:
|
|
stats["empty_instruction_or_non_symbol"] += 1
|
|
|
|
feature_count = len(grouped.get(object_class, []))
|
|
summary[object_class] = {
|
|
"feature_count": feature_count,
|
|
"stats": dict(stats),
|
|
"match_rate": (
|
|
stats["matched"] / feature_count if feature_count else None
|
|
),
|
|
"top_symbols": symbols.most_common(20),
|
|
"conditional_procedures": dict(csps),
|
|
"top_lookup_rcids": rcids.most_common(20),
|
|
}
|
|
|
|
report = {
|
|
"schema": "navsea.cm95.s52-database-match-audit/v1",
|
|
"dataset_id": args.dataset_id,
|
|
"table_name": args.table,
|
|
"classes": classes,
|
|
"summary": summary,
|
|
}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(
|
|
json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|