import json from collections import Counter, defaultdict from pathlib import Path import pymysql ROOT = Path(__file__).resolve().parent STYLE_JSON_PATH = ROOT / "src" / "pbf" / "style.json" MANUAL_OBJECT_RULES = { "P754ククリ": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Source-preserved clip/outline layer.", }, "P危険界ククリ": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Source-preserved clip/outline layer.", }, "P基本線ククリ": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Source-preserved clip/outline layer.", }, "P投錨注意障害物ククリ": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Source-preserved clip/outline layer.", }, "P施設・境界線等ククリ": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Source-preserved clip/outline layer.", }, "P航路ククリ": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Source-preserved clip/outline layer.", }, "P穴": { "family": "surface", "allowed_geom_types": ["Polygon", "MultiPolygon"], "notes": "Source-preserved area layer.", }, "P陸域": { "family": "surface", "allowed_geom_types": ["Polygon", "MultiPolygon"], "notes": "Source-preserved area layer.", }, "L海底地形": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Source-preserved bathymetry line layer.", }, "p地名": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Source-preserved place label layer.", }, "p地名陸": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Source-preserved place label layer.", }, "p高さ制限": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Source-preserved point annotation layer.", }, "p錨泊地等": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Source-preserved point annotation layer.", }, "サンドウェーブ": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Same object appears in multiple source layers but remains a point hazard.", }, "海底設置物、放水口、取水口": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Stable point object despite multiple source layers.", }, "全沈没船 (危険なし)": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Stable point object despite multiple source layers.", }, "測定済みの沈船": { "family": "symbol", "allowed_geom_types": ["Point"], "notes": "Stable point object despite multiple source layers.", }, "道路": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Road centerlines legitimately appear in multi-part line form.", }, "等深線": { "family": "line", "allowed_geom_types": ["LineString", "MultiLineString"], "notes": "Depth contours legitimately appear in multi-part line form.", }, "防波堤": { "family": "surface", "allowed_geom_types": ["Polygon", "MultiPolygon"], "notes": "Breakwaters are modeled as areas, including multipart polygons.", }, "障害物": { "family": "mixed", "allowed_geom_types": ["Point", "Polygon", "MultiPolygon"], "notes": "Obstacle objects can be rendered as symbols or area extents.", }, "未測海域": { "family": "mixed", "allowed_geom_types": ["LineString", "Polygon", "MultiPolygon"], "notes": "Unsurveyed regions appear as boundaries and areas.", }, "浮施設・桟橋": { "family": "mixed", "allowed_geom_types": ["LineString", "MultiLineString", "Polygon", "MultiPolygon"], "notes": "Floating facilities and piers appear as line and area geometry.", }, "航路 (法律による航路)": { "family": "mixed", "allowed_geom_types": ["LineString", "Polygon", "MultiPolygon"], "notes": "Legally defined routes appear as lines and areas.", }, "険悪物": { "family": "mixed", "allowed_geom_types": ["Point", "Polygon", "MultiPolygon"], "notes": "Hazards may be represented as point symbols or area extents.", }, "魚礁": { "family": "mixed", "allowed_geom_types": ["Point", "Polygon", "MultiPolygon"], "notes": "Artificial reefs may be represented as point symbols or area extents.", }, "養殖場": { "family": "surface", "allowed_geom_types": ["Polygon", "MultiPolygon"], "notes": "Aquaculture zones are area features.", }, "漁網": { "family": "surface", "allowed_geom_types": ["Polygon", "MultiPolygon"], "notes": "Fishing nets are area extents in this dataset.", }, "錨泊 (指定)地": { "family": "mixed", "allowed_geom_types": ["Point", "Polygon", "MultiPolygon"], "notes": "Anchorage can appear as symbol or area depending on source layer.", }, "ケーソン仮置き場": { "family": "surface", "allowed_geom_types": ["Polygon", "MultiPolygon"], "notes": "Caisson storage areas are polygons.", }, "土砂捨て場": { "family": "surface", "allowed_geom_types": ["Polygon", "MultiPolygon"], "notes": "Spoil grounds are polygons.", }, } DEFAULT_GEOMETRIES = { "symbol": ["Point"], "line": ["LineString", "MultiLineString"], "surface": ["Polygon", "MultiPolygon"], "mixed": [], } def connect(): return pymysql.connect( host="localhost", user="root", password="2chi9ks2", database="pbf_analysis", charset="utf8mb4", unix_socket="/tmp/mysql.sock", autocommit=False, ) def load_style_source_layers(): style = json.loads(STYLE_JSON_PATH.read_text(encoding="utf-8")) layers = {} for layer in style.get("layers", []): source_layer = layer.get("source-layer") if not source_layer: continue layers.setdefault(source_layer, {"style_rows": 0, "layer_types": set()}) layers[source_layer]["style_rows"] += 1 if layer.get("type"): layers[source_layer]["layer_types"].add(layer["type"]) return layers def geom_bucket(geom_type: str | None) -> str: if geom_type in {"Point", "MultiPoint"}: return "symbol" if geom_type in {"LineString", "MultiLineString"}: return "line" if geom_type in {"Polygon", "MultiPolygon"}: return "surface" return "mixed" def infer_family(geom_types: set[str]) -> str: buckets = {geom_bucket(geom_type) for geom_type in geom_types if geom_type} if not buckets: return "mixed" if len(buckets) == 1: return next(iter(buckets)) return "mixed" def choose_allowed_geometries(object_type: str, family: str, observed_geom_types: list[str]) -> tuple[list[str], str]: manual = MANUAL_OBJECT_RULES.get(object_type) if manual: return manual["allowed_geom_types"], "manual_override" defaults = DEFAULT_GEOMETRIES.get(family, []) if defaults: return defaults, "family_default" return observed_geom_types, "observed_fallback" def build_canonical_layer_rules(cur, style_layers): cur.execute("DROP TABLE IF EXISTS canonical_layer_rules") cur.execute( """ CREATE TABLE canonical_layer_rules ( source_layer VARCHAR(100) NOT NULL, semantic_granularity VARCHAR(50) NOT NULL, canonical_family VARCHAR(100) NOT NULL, render_strategy VARCHAR(50) NOT NULL, preserve_source_layer TINYINT(1) NOT NULL DEFAULT 1, style_bound TINYINT(1) NOT NULL DEFAULT 1, style_rows INT NOT NULL DEFAULT 0, style_types VARCHAR(200) NULL, notes TEXT NULL, PRIMARY KEY (source_layer) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) cur.execute( """ SELECT vt_layer, SUM(feature_count) AS feature_count, COUNT(DISTINCT COALESCE(NULLIF(class_name, ''), NULLIF(shape_name, ''), vt_layer)) AS semantic_type_count, SUM(CASE WHEN class_name IS NULL OR class_name='' THEN feature_count ELSE 0 END) AS missing_class_count, SUM(CASE WHEN shape_name IS NULL OR shape_name='' THEN feature_count ELSE 0 END) AS missing_shape_count FROM object_catalog GROUP BY vt_layer """ ) inserts = [] for vt_layer, feature_count, semantic_type_count, missing_class_count, missing_shape_count in cur.fetchall(): style_info = style_layers.get(vt_layer, {"style_rows": 0, "layer_types": set()}) if vt_layer.startswith("p"): canonical_family = "symbol" elif vt_layer.startswith("L"): canonical_family = "line" elif vt_layer.startswith("P"): canonical_family = "surface" else: canonical_family = "mixed" if semantic_type_count >= 5 and (vt_layer.startswith("P") or vt_layer.startswith("L")): semantic_granularity = "container_layer" notes = ( f"Layer contains {semantic_type_count} semantic types across {feature_count} features; " "retain source layer for rendering and derive semantic overlays for analysis." ) elif missing_class_count == feature_count and missing_shape_count == feature_count: semantic_granularity = "style_or_source_layer" notes = "Layer has no semantic class/shape annotations; source layer remains the stable identity." else: semantic_granularity = "semantic_layer" notes = "Layer is close to a stable semantic object and can be preserved as-is." inserts.append( ( vt_layer, semantic_granularity, canonical_family, "keep_source_layer", 1, 1 if style_info["style_rows"] > 0 else 0, style_info["style_rows"], ",".join(sorted(style_info["layer_types"])) or None, notes, ) ) cur.executemany( """ INSERT INTO canonical_layer_rules ( source_layer, semantic_granularity, canonical_family, render_strategy, preserve_source_layer, style_bound, style_rows, style_types, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, inserts, ) def build_canonical_object_rules(cur): cur.execute("DROP TABLE IF EXISTS canonical_object_rules") cur.execute( """ CREATE TABLE canonical_object_rules ( canonical_object_type VARCHAR(191) NOT NULL, canonical_family VARCHAR(100) NOT NULL, classification_basis VARCHAR(50) NOT NULL, source_layer_scope VARCHAR(50) NOT NULL, source_layer_count INT NOT NULL, geom_type_count INT NOT NULL, object_type_sources VARCHAR(100) NOT NULL, source_layers TEXT NULL, observed_geom_types VARCHAR(200) NULL, allowed_geom_types VARCHAR(200) NULL, preferred_geom_type VARCHAR(20) NULL, geometry_rule_source VARCHAR(50) NOT NULL, notes TEXT NULL, PRIMARY KEY (canonical_object_type) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) cur.execute( """ SELECT COALESCE(NULLIF(class_name, ''), NULLIF(shape_name, ''), vt_layer) AS object_type, GROUP_CONCAT( DISTINCT CASE WHEN class_name IS NOT NULL AND class_name <> '' THEN 'class_name' WHEN shape_name IS NOT NULL AND shape_name <> '' THEN 'shape_name' ELSE 'vt_layer' END ORDER BY 1 SEPARATOR ',' ) AS object_type_sources, SUM(feature_count) AS feature_count, COUNT(DISTINCT vt_layer) AS source_layer_count, GROUP_CONCAT(DISTINCT vt_layer ORDER BY vt_layer SEPARATOR ',') AS source_layers, COUNT(DISTINCT geom_type) AS geom_type_count, GROUP_CONCAT(DISTINCT geom_type ORDER BY geom_type SEPARATOR ',') AS observed_geom_types FROM object_catalog GROUP BY COALESCE(NULLIF(class_name, ''), NULLIF(shape_name, ''), vt_layer) """ ) object_rows = cur.fetchall() cur.execute( """ SELECT object_type, geom_type, feature_count FROM object_type_stats """ ) geom_counts: dict[str, Counter[str]] = defaultdict(Counter) for object_type, geom_type, feature_count in cur.fetchall(): geom_counts[object_type][geom_type] = feature_count inserts = [] for ( object_type, object_type_sources, feature_count, source_layer_count, source_layers, geom_type_count, observed_geom_types, ) in object_rows: observed_geom_list = [part for part in (observed_geom_types or "").split(",") if part] observed_geom_set = set(observed_geom_list) manual = MANUAL_OBJECT_RULES.get(object_type) canonical_family = manual["family"] if manual else infer_family(observed_geom_set) classification_basis = ( "source_layer_preserved" if object_type_sources == "vt_layer" else "semantic_object" ) source_layer_scope = "multi_source" if source_layer_count > 1 else "single_source" allowed_geom_types, geometry_rule_source = choose_allowed_geometries( object_type, canonical_family, observed_geom_list, ) preferred_geom_type = None if geom_counts.get(object_type): preferred_geom_type = geom_counts[object_type].most_common(1)[0][0] notes = ( manual["notes"] if manual else ( "Source-preserved object type derived from vt_layer only." if classification_basis == "source_layer_preserved" else "Semantic object type derived from class_name/shape_name." ) ) inserts.append( ( object_type, canonical_family, classification_basis, source_layer_scope, source_layer_count, geom_type_count, object_type_sources, source_layers, ",".join(observed_geom_list) or None, ",".join(allowed_geom_types) or None, preferred_geom_type, geometry_rule_source, notes, ) ) cur.executemany( """ INSERT INTO canonical_object_rules ( canonical_object_type, canonical_family, classification_basis, source_layer_scope, source_layer_count, geom_type_count, object_type_sources, source_layers, observed_geom_types, allowed_geom_types, preferred_geom_type, geometry_rule_source, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, inserts, ) def build_object_geometry_allowlist(cur): cur.execute("DROP TABLE IF EXISTS object_geometry_allowlist") cur.execute( """ CREATE TABLE object_geometry_allowlist ( canonical_object_type VARCHAR(191) NOT NULL, geometry_type VARCHAR(20) NOT NULL, rule_source VARCHAR(50) NOT NULL, is_allowed TINYINT(1) NOT NULL DEFAULT 1, notes TEXT NULL, PRIMARY KEY (canonical_object_type, geometry_type) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) cur.execute( """ SELECT canonical_object_type, geometry_rule_source, allowed_geom_types, notes FROM canonical_object_rules """ ) inserts = [] for canonical_object_type, geometry_rule_source, allowed_geom_types, notes in cur.fetchall(): for geometry_type in [part for part in (allowed_geom_types or "").split(",") if part]: inserts.append( ( canonical_object_type, geometry_type, geometry_rule_source, 1, notes, ) ) if inserts: cur.executemany( """ INSERT INTO object_geometry_allowlist ( canonical_object_type, geometry_type, rule_source, is_allowed, notes ) VALUES (%s, %s, %s, %s, %s) """, inserts, ) def build_relayer_outputs(cur): cur.execute("DROP TABLE IF EXISTS pbf_relayer_candidates") cur.execute( """ CREATE TABLE pbf_relayer_candidates AS SELECT o.feature_id, s.z, s.x, s.y, s.vt_layer AS source_layer, s.geom_type, s.class_name, s.shape_name, s.layer_name, o.object_type_source, o.object_type AS canonical_object_type, r.semantic_granularity, obj.canonical_family, obj.classification_basis, obj.source_layer_scope, r.render_strategy, r.preserve_source_layer, r.style_bound, CASE WHEN obj.classification_basis = 'source_layer_preserved' THEN CONCAT('source:', s.vt_layer) ELSE CONCAT(obj.canonical_family, ':', o.object_type) END AS semantic_key, CASE WHEN obj.classification_basis = 'source_layer_preserved' THEN CONCAT(obj.canonical_family, ':', s.vt_layer) ELSE CONCAT(obj.canonical_family, ':', o.object_type) END AS detection_key, CASE WHEN r.preserve_source_layer = 1 THEN s.vt_layer ELSE o.object_type END AS render_layer FROM feature_semantic s JOIN object_type_candidates o ON s.feature_id = o.feature_id JOIN canonical_layer_rules r ON s.vt_layer = r.source_layer JOIN canonical_object_rules obj ON o.object_type = obj.canonical_object_type """ ) cur.execute("DROP TABLE IF EXISTS pbf_source_object_stats") cur.execute( """ CREATE TABLE pbf_source_object_stats AS SELECT source_layer, render_layer, canonical_family, classification_basis, source_layer_scope, semantic_granularity, style_bound, canonical_object_type, geom_type, detection_key, semantic_key, COUNT(*) AS feature_count FROM pbf_relayer_candidates GROUP BY source_layer, render_layer, canonical_family, classification_basis, source_layer_scope, semantic_granularity, style_bound, canonical_object_type, geom_type, detection_key, semantic_key """ ) cur.execute("ALTER TABLE pbf_source_object_stats ADD KEY idx_source_layer (source_layer(100))") cur.execute("ALTER TABLE pbf_source_object_stats ADD KEY idx_render_layer (render_layer(100))") cur.execute("ALTER TABLE pbf_source_object_stats ADD KEY idx_detection_key (detection_key(50))") cur.execute("DROP TABLE IF EXISTS pbf_render_compatibility") cur.execute( """ CREATE TABLE pbf_render_compatibility AS SELECT s.source_layer, s.render_layer, s.canonical_family, s.semantic_granularity, s.style_bound, SUM(s.feature_count) AS feature_count, COUNT(*) AS canonical_object_types FROM pbf_source_object_stats s GROUP BY s.source_layer, s.render_layer, s.canonical_family, s.semantic_granularity, s.style_bound """ ) cur.execute("DROP TABLE IF EXISTS pbf_detection_catalog") cur.execute( """ CREATE TABLE pbf_detection_catalog AS SELECT detection_key, canonical_family, canonical_object_type, geom_type, SUM(feature_count) AS feature_count, COUNT(DISTINCT source_layer) AS source_layers FROM pbf_source_object_stats GROUP BY detection_key, canonical_family, canonical_object_type, geom_type """ ) def refresh_relayer_tables(): style_layers = load_style_source_layers() with connect() as conn: with conn.cursor() as cur: build_canonical_layer_rules(cur, style_layers) build_canonical_object_rules(cur) build_object_geometry_allowlist(cur) build_relayer_outputs(cur) conn.commit() def main(): refresh_relayer_tables() print( "Built canonical_layer_rules, canonical_object_rules, object_geometry_allowlist, " "pbf_relayer_candidates, pbf_render_compatibility, pbf_detection_catalog" ) if __name__ == "__main__": main()