268 lines
11 KiB
Python
268 lines
11 KiB
Python
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 (
|
||
lookup_condition_matches,
|
||
normalize_cm93_attributes_for_s52,
|
||
normalize_cm93_object_for_s52,
|
||
)
|
||
|
||
|
||
IMPLEMENTED_CSP = frozenset({
|
||
"DEPARE01", "DEPARE02", "DEPCNT02", "LIGHTS05", "SOUNDG02",
|
||
"OBSTRN04", "WRECKS02", "QUAPOS01", "SLCONS03", "RESTRN01",
|
||
"RESARE02", "TOPMAR01",
|
||
})
|
||
|
||
|
||
def _json_object(value: Any) -> dict[str, Any]:
|
||
if value is None:
|
||
return {}
|
||
if isinstance(value, dict):
|
||
return value
|
||
try:
|
||
parsed = json.loads(value)
|
||
except (TypeError, json.JSONDecodeError):
|
||
return {}
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
|
||
|
||
def _geometry_key(geometry_type: str, point_table: str, boundary_table: str):
|
||
return {
|
||
"POINT": (point_table, "Point"),
|
||
"SOUNDING": (point_table, "Point"),
|
||
"LINE": ("Lines", "Line"),
|
||
"AREA": (boundary_table, "Area"),
|
||
}.get(str(geometry_type).upper())
|
||
|
||
|
||
def _prepare_lookups(rows: list[dict[str, Any]]):
|
||
index: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list)
|
||
for row in rows:
|
||
key = (
|
||
str(row.get("object_class") or ""),
|
||
str(row.get("table_name") or ""),
|
||
str(row.get("geometry_type") or ""),
|
||
)
|
||
index[key].append(row)
|
||
for key, candidates in index.items():
|
||
candidates.sort(key=lambda r: (-len(r.get("attribute_conditions_raw") or []), int(r.get("id") or 0)))
|
||
return index
|
||
|
||
|
||
def _resolve(candidates: list[dict[str, Any]], attrs: dict[str, Any]):
|
||
normalized = normalize_cm93_attributes_for_s52(attrs)
|
||
fallback = None
|
||
for row in candidates:
|
||
conditions = row.get("attribute_conditions_raw") or []
|
||
if not conditions:
|
||
if fallback is None:
|
||
fallback = row
|
||
continue
|
||
if all(lookup_condition_matches(c, normalized) for c in conditions):
|
||
return row
|
||
return fallback
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="流式审计整个 CM95 Dataset 的 S-52 Lookup/CSP 覆盖率。")
|
||
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("--point-table", default="Paper")
|
||
parser.add_argument("--boundary-table", default="Plain")
|
||
args = parser.parse_args()
|
||
|
||
lookup_rows = json.loads(args.lookups.read_text(encoding="utf-8"))
|
||
lookup_index = _prepare_lookups(lookup_rows)
|
||
|
||
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,
|
||
cursorclass=pymysql.cursors.SSCursor,
|
||
)
|
||
|
||
total = 0
|
||
matched = 0
|
||
no_lookup = 0
|
||
empty_candidates = 0
|
||
by_geometry = Counter()
|
||
matched_by_geometry = Counter()
|
||
no_lookup_by_key = Counter()
|
||
no_candidate_by_key = Counter()
|
||
condition_miss_by_key = Counter()
|
||
matched_by_key = Counter()
|
||
rcids = Counter()
|
||
csp_counts = Counter()
|
||
implemented_csp_counts = Counter()
|
||
unsupported_csp_counts = Counter()
|
||
opcode_counts = Counter()
|
||
direct_symbol_counts = Counter()
|
||
pattern_counts = Counter()
|
||
complex_line_counts = Counter()
|
||
area_color_counts = Counter()
|
||
simple_line_counts = Counter()
|
||
tx_rule_counts = Counter()
|
||
te_rule_counts = Counter()
|
||
substitution_counts = Counter()
|
||
|
||
sql = (
|
||
"SELECT object_class, geometry_type, attributes_json "
|
||
"FROM chart_feature WHERE dataset_id=%s ORDER BY id"
|
||
)
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql, (args.dataset_id,))
|
||
for object_class, geometry_type, attrs_raw in cur:
|
||
total += 1
|
||
object_class = str(object_class)
|
||
geometry_type = str(geometry_type)
|
||
attrs = _json_object(attrs_raw)
|
||
lookup_class, lookup_attrs = normalize_cm93_object_for_s52(
|
||
object_class, geometry_type, attrs
|
||
)
|
||
if lookup_class != object_class:
|
||
substitution_counts[(object_class, lookup_class, geometry_type)] += 1
|
||
by_geometry[geometry_type] += 1
|
||
tg = _geometry_key(geometry_type, args.point_table, args.boundary_table)
|
||
if tg is None:
|
||
no_lookup += 1
|
||
empty_candidates += 1
|
||
no_lookup_by_key[(object_class, geometry_type)] += 1
|
||
no_candidate_by_key[(object_class, geometry_type)] += 1
|
||
continue
|
||
table_name, lookup_geometry = tg
|
||
candidates = lookup_index.get((lookup_class, table_name, lookup_geometry), [])
|
||
if not candidates:
|
||
no_lookup += 1
|
||
empty_candidates += 1
|
||
no_lookup_by_key[(object_class, geometry_type)] += 1
|
||
no_candidate_by_key[(object_class, geometry_type)] += 1
|
||
continue
|
||
hit = _resolve(candidates, lookup_attrs)
|
||
if hit is None:
|
||
no_lookup += 1
|
||
no_lookup_by_key[(object_class, geometry_type)] += 1
|
||
condition_miss_by_key[(object_class, geometry_type)] += 1
|
||
continue
|
||
matched += 1
|
||
matched_by_geometry[geometry_type] += 1
|
||
matched_by_key[(object_class, geometry_type)] += 1
|
||
rcids[str(hit.get("rcid") or "")] += 1
|
||
rules = hit.get("rules") or []
|
||
for rule in rules:
|
||
opcode = str(rule.get("opcode") or "")
|
||
if opcode:
|
||
opcode_counts[opcode] += 1
|
||
args_list = rule.get("args") or []
|
||
if opcode == "SY" and args_list:
|
||
direct_symbol_counts[str(args_list[0]).strip("'\"")[:8]] += 1
|
||
elif opcode == "AP" and args_list:
|
||
pattern_counts[str(args_list[0]).strip("'\"")[:8]] += 1
|
||
elif opcode == "LC" and args_list:
|
||
complex_line_counts[str(args_list[0]).strip("'\"")[:8]] += 1
|
||
elif opcode == "AC" and args_list:
|
||
area_color_counts[str(args_list[0]).strip("'\"")] += 1
|
||
elif opcode == "LS" and len(args_list) >= 3:
|
||
simple_line_counts[tuple(str(x).strip("'\"") for x in args_list[:3])] += 1
|
||
elif opcode == "TX":
|
||
tx_rule_counts[tuple(str(x) for x in args_list)] += 1
|
||
elif opcode == "TE":
|
||
te_rule_counts[tuple(str(x) for x in args_list)] += 1
|
||
if opcode == "CS" and args_list:
|
||
name = str(args_list[0])[:8]
|
||
csp_counts[name] += 1
|
||
if name in IMPLEMENTED_CSP:
|
||
implemented_csp_counts[name] += 1
|
||
else:
|
||
unsupported_csp_counts[name] += 1
|
||
conn.close()
|
||
|
||
top_no_lookup = [
|
||
{"object_class": k[0], "geometry_type": k[1], "count": v}
|
||
for k, v in no_lookup_by_key.most_common()
|
||
]
|
||
top_matched = [
|
||
{"object_class": k[0], "geometry_type": k[1], "count": v}
|
||
for k, v in matched_by_key.most_common()
|
||
]
|
||
report = {
|
||
"schema": "navsea.cm95.s52-full-database-audit/v1",
|
||
"dataset_id": args.dataset_id,
|
||
"point_table": args.point_table,
|
||
"boundary_table": args.boundary_table,
|
||
"feature_count": total,
|
||
"matched_lookup_count": matched,
|
||
"no_lookup_count": no_lookup,
|
||
"no_candidate_count": empty_candidates,
|
||
"condition_no_match_count": no_lookup - empty_candidates,
|
||
"match_rate": matched / total if total else None,
|
||
"geometry_counts": dict(by_geometry),
|
||
"matched_geometry_counts": dict(matched_by_geometry),
|
||
"lookup_rcid_top100": rcids.most_common(100),
|
||
"opcode_counts": dict(opcode_counts),
|
||
"csp_counts": dict(csp_counts),
|
||
"implemented_csp_counts": dict(implemented_csp_counts),
|
||
"unsupported_csp_counts": dict(unsupported_csp_counts),
|
||
"direct_symbol_top100": direct_symbol_counts.most_common(100),
|
||
"symbol_resource_counts": direct_symbol_counts.most_common(),
|
||
"pattern_resource_counts": pattern_counts.most_common(),
|
||
"complex_line_resource_counts": complex_line_counts.most_common(),
|
||
"area_color_counts": area_color_counts.most_common(),
|
||
"simple_line_rule_counts": [[list(k), v] for k, v in simple_line_counts.most_common()],
|
||
"tx_rule_counts": [[list(k), v] for k, v in tx_rule_counts.most_common()],
|
||
"te_rule_counts": [[list(k), v] for k, v in te_rule_counts.most_common()],
|
||
"class_substitution_counts": [
|
||
{"source_object_class": k[0], "s52_object_class": k[1], "geometry_type": k[2], "count": v}
|
||
for k, v in substitution_counts.most_common()
|
||
],
|
||
"no_lookup_by_object_geometry": top_no_lookup,
|
||
"no_candidate_by_object_geometry": [
|
||
{"object_class": k[0], "geometry_type": k[1], "count": v}
|
||
for k, v in no_candidate_by_key.most_common()
|
||
],
|
||
"condition_no_match_by_object_geometry": [
|
||
{"object_class": k[0], "geometry_type": k[1], "count": v}
|
||
for k, v in condition_miss_by_key.most_common()
|
||
],
|
||
"matched_by_object_geometry": top_matched,
|
||
"notes": {
|
||
"context_dependent_csp": "本报告只统计 Lookup 实际触发量，不执行需要邻接几何、sounding depth、TOPMAR 平台关系等上下文的 CSP。",
|
||
"implemented_csp": sorted(IMPLEMENTED_CSP),
|
||
},
|
||
}
|
||
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({
|
||
"feature_count": total,
|
||
"matched_lookup_count": matched,
|
||
"no_lookup_count": no_lookup,
|
||
"no_candidate_count": empty_candidates,
|
||
"condition_no_match_count": no_lookup - empty_candidates,
|
||
"match_rate": report["match_rate"],
|
||
"csp_counts": report["csp_counts"],
|
||
"unsupported_csp_counts": report["unsupported_csp_counts"],
|
||
"class_substitution_counts": report["class_substitution_counts"],
|
||
"top_no_lookup": top_no_lookup[:30],
|
||
}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|