from __future__ import annotations import os from dataclasses import dataclass import pymysql @dataclass(frozen=True) class DbConfig: host: str = os.getenv("NAVSEA_DB_HOST", "localhost") port: int = int(os.getenv("NAVSEA_DB_PORT", "3306")) user: str = os.getenv("NAVSEA_DB_USER", "root") password: str = os.getenv("NAVSEA_DB_PASSWORD", "2chi9ks2") database: str = os.getenv("NAVSEA_DB_NAME", "pbf_analysis") unix_socket: str | None = os.getenv("NAVSEA_DB_SOCKET", "/tmp/mysql.sock") FAMILY_DEFAULT_CAPABILITIES = { "hazard": {"collision"}, "navigation_aid": {"navigation_mark"}, "bathymetry": {"depth_reference"}, "seabed": {"bottom_reference"}, "fishery": {"entangle"}, "route": {"route_reference"}, "boundary": {"boundary_reference"}, "anchorage": {"anchorage_reference"}, "restricted_area": {"boundary_reference"}, "infrastructure": {"structure_reference"}, "utility": {"utility_hazard"}, "landmark": {"landmark_reference"}, "place": {"place_reference"}, "water_area": {"water_reference"}, "monitoring": {"traffic_monitoring"}, "overlay_support": {"render_support"}, "unknown": {"review_required"}, } class DetectionPipeline: def __init__(self, config: DbConfig) -> None: self.config = config def connect(self): kwargs = { "host": self.config.host, "port": self.config.port, "user": self.config.user, "password": self.config.password, "database": self.config.database, "charset": "utf8mb4", "autocommit": False, } if self.config.unix_socket and self.config.host in {"localhost", "127.0.0.1"}: kwargs["unix_socket"] = self.config.unix_socket return pymysql.connect(**kwargs) def run(self) -> None: with self.connect() as conn: with conn.cursor() as cur: object_types = self.fetch_object_types(cur) taxonomy_rows = [self.classify_object_type(name) for name in object_types] capability_rows = self.build_capabilities(taxonomy_rows) self.create_taxonomy_table(cur, taxonomy_rows) self.create_capabilities_table(cur, capability_rows) self.create_detection_view(cur) self.create_capability_views(cur) self.ensure_query_indexes(cur) conn.commit() print( "Built navsea_object_taxonomy, navsea_object_capabilities, " "navsea_detection_objects, capability views and supporting query indexes" ) @staticmethod def fetch_object_types(cur) -> list[str]: cur.execute( """ SELECT canonical_object_type FROM canonical_object_rules ORDER BY canonical_object_type """ ) return [row[0] for row in cur.fetchall()] @staticmethod def is_depth_band(name: str) -> bool: compact = name.replace(" ", "") return compact.endswith("m") or "m以深" in compact @staticmethod def classify_object_type(name: str) -> tuple[str, str, str, str]: if name.startswith(("P", "L", "p")) and ( name.endswith("ククリ") or name in { "P穴", "P陸域", "P危険界ククリ", "P投錨注意障害物ククリ", "P施設・境界線等ククリ", "P航路ククリ", "P錨泊地等ククリ", "P754ククリ", "P721ククリ", "P730ククリ", "L海底地形", "p地名", "p地名陸", "p高さ制限", } ): return ( name, "overlay_support", "render_support", "保留原始 source_layer 的叠加支持对象，不作为独立语义实体替换原始渲染层。", ) if DetectionPipeline.is_depth_band(name): return (name, "bathymetry", "depth_zone", "水深分带对象，用于浅滩、深度范围和搁浅风险分析。") if any(token in name for token in ("等深線",)): return (name, "bathymetry", "depth_contour", "等深线对象，用于水深变化与航线安全分析。") if any(token in name for token in ("底質", "海底地形", "サンドウェーブ", "海底火山")): return (name, "seabed", "seabed_feature", "海底形态或底质对象，用于海底环境识别。") if any( token in name for token in ( "険悪物", "障害物", "危険物", "危険全沈没船", "沈船", "全沈没船", "沈木", "暗岩", "洗岩", "干出岩", "水上岩", "孤立危険物", "魚礁", "浅所危険界", "撤去跡", "掃海済み", "サンゴ礁", ) ): return (name, "hazard", "obstacle", "对航行存在碰撞或搁浅风险的危险物对象。") if any( token in name for token in ( "灯", "灯台", "灯標", "灯浮標", "浮標", "立標", "導灯", "指向灯", "V-AIS", "管制信号所", ) ): return (name, "navigation_aid", "navigation_mark", "航标、灯标或导助航对象。") if any(token in name for token in ("航路", "進路矢印", "分離通航方式", "誘導線", "指導線")): return (name, "route", "route_reference", "航路、导向线或通航组织对象。") if any( token in name for token in ( "境界", "危険界", "境界線", "制限区域", "専用用途海域", "航泊禁止区域", "錨泊禁止区域", "航空機進入区域", ) ): return (name, "boundary", "boundary_control", "边界、限制区或规则控制对象。") if "錨泊" in name or "錨地" in name: return (name, "anchorage", "anchorage", "锚地或锚泊控制对象。") if any(token in name for token in ("漁", "養殖場", "漁網", "海草")): return (name, "fishery", "fishery_area", "渔业、养殖或缠绕风险相关对象。") if any( token in name for token in ( "防波堤", "潜提", "桟橋", "海上バース", "ドルフィン", "ポンツーン", "橋", "道路", "ビル", "タンク", "煙突", "塔", "やぐら", "ケーソン", "土砂捨て場", "石油開発台", ) ): return (name, "infrastructure", "structure", "港口、岸线或陆上海工构造物对象。") if any(token in name for token in ("海底線", "輸送管", "送電線", "架空線", "放水口", "取水口")): return (name, "utility", "utility_line", "海底线缆、管道或公用设施对象。") if any(token in name for token in ("河川域", "湖沼域", "陸上水域", "干潮帯", "未測海域", "平水境界")): return (name, "water_area", "water_area", "水域、潮滩或未测区域对象。") if any( token in name for token in ( "その他 (記念碑等)", "陸上顕著物", "地名", "山頂", "港湾", "漁港", "マリーナ", "フィッシャリーナ", "海の駅", "漁業協同組合", "海事関係署", "税関", "パイロットステーション", ) ): return (name, "place", "place_label", "地名、显著地物或港区服务设施类参考对象。") if any(token in name for token in ("廃棄物捨て場", "貯木場")): return (name, "infrastructure", "structure", "作业、堆置或弃置区域类设施对象。") if any(token in name for token in ("無線局", "レーダー局")): return (name, "monitoring", "monitoring_station", "通信、雷达或监测设施对象。") if any(token in name for token in ("急潮", "波紋", "激潮", "渦流", "水源方向")): return (name, "hazard", "current_hazard", "流态、水流或潮流风险相关对象。") return (name, "unknown", "review_required", "尚未细化归类的对象，需要人工复核。") @staticmethod def build_capabilities( taxonomy_rows: list[tuple[str, str, str, str]] ) -> list[tuple[str, str]]: capability_rows: set[tuple[str, str]] = set() for canonical_object_type, object_family, detection_class, _ in taxonomy_rows: capabilities = set(FAMILY_DEFAULT_CAPABILITIES.get(object_family, {"review_required"})) if detection_class in {"obstacle"}: capabilities.add("collision") if detection_class in {"depth_zone", "depth_contour"}: capabilities.add("grounding") if "魚礁" in canonical_object_type or "浅所" in canonical_object_type or "干潮帯" in canonical_object_type: capabilities.add("grounding") if any(token in canonical_object_type for token in ("漁", "養殖場", "漁網", "海草")): capabilities.add("entangle") if any(token in canonical_object_type for token in ("防波堤",)): capabilities.add("wave_barrier") if any(token in canonical_object_type for token in ("架空線", "送電線", "高さ制限")): capabilities.add("overhead_clearance") if any(token in canonical_object_type for token in ("航路", "誘導線", "進路矢印", "分離通航方式")): capabilities.add("route_reference") if any(token in canonical_object_type for token in ("錨泊", "錨地")): capabilities.add("anchorage_reference") if any(token in canonical_object_type for token in ("境界", "区域", "危険界")): capabilities.add("boundary_reference") if any(token in canonical_object_type for token in ("海底線", "輸送管")): capabilities.add("snag_risk") if any(token in canonical_object_type for token in ("灯", "標", "浮標", "導灯", "V-AIS", "管制信号所")): capabilities.add("navigation_mark") if object_family in {"place", "infrastructure", "monitoring"}: capabilities.add("landmark_reference") for capability in sorted(capabilities): capability_rows.add((canonical_object_type, capability)) return sorted(capability_rows) @staticmethod def create_taxonomy_table(cur, rows: list[tuple[str, str, str, str]]) -> None: cur.execute("DROP TABLE IF EXISTS navsea_object_taxonomy") cur.execute( """ CREATE TABLE navsea_object_taxonomy ( canonical_object_type VARCHAR(191) NOT NULL, object_family VARCHAR(64) NOT NULL, detection_class VARCHAR(64) NOT NULL, description TEXT NULL, PRIMARY KEY (canonical_object_type), KEY idx_family_class (object_family, detection_class) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) cur.executemany( """ INSERT INTO navsea_object_taxonomy ( canonical_object_type, object_family, detection_class, description ) VALUES (%s, %s, %s, %s) """, rows, ) @staticmethod def create_capabilities_table(cur, rows: list[tuple[str, str]]) -> None: cur.execute("DROP TABLE IF EXISTS navsea_object_capabilities") cur.execute( """ CREATE TABLE navsea_object_capabilities ( canonical_object_type VARCHAR(191) NOT NULL, capability VARCHAR(64) NOT NULL, PRIMARY KEY (canonical_object_type, capability), KEY idx_capability (capability, canonical_object_type) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) cur.executemany( """ INSERT INTO navsea_object_capabilities ( canonical_object_type, capability ) VALUES (%s, %s) """, rows, ) @staticmethod def create_detection_view(cur) -> None: cur.execute("DROP VIEW IF EXISTS navsea_detection_objects") cur.execute( """ CREATE VIEW navsea_detection_objects AS SELECT r.feature_id, CAST(NULL AS CHAR(1)) AS geometry, r.z, r.x, r.y, r.geom_type, r.source_layer, r.class_name, r.shape_name, r.layer_name, t.canonical_object_type, t.object_family, t.detection_class, c.capability, r.detection_key FROM pbf_relayer_candidates r JOIN navsea_object_taxonomy t ON r.canonical_object_type = t.canonical_object_type LEFT JOIN navsea_object_capabilities c ON r.canonical_object_type = c.canonical_object_type """ ) @staticmethod def create_capability_views(cur) -> None: views = { "navsea_collision_objects": "collision", "navsea_grounding_objects": "grounding", "navsea_entangle_objects": "entangle", "navsea_navigation_mark_objects": "navigation_mark", "navsea_route_reference_objects": "route_reference", "navsea_boundary_reference_objects": "boundary_reference", } for view_name, capability in views.items(): cur.execute(f"DROP VIEW IF EXISTS {view_name}") cur.execute( f""" CREATE VIEW {view_name} AS SELECT * FROM navsea_detection_objects WHERE capability = %s """, (capability,), ) @staticmethod def add_index_if_missing(cur, table: str, index_name: str, ddl: str) -> None: cur.execute(f"SHOW INDEX FROM {table} WHERE Key_name=%s", (index_name,)) if cur.fetchone() is None: cur.execute(ddl) def ensure_query_indexes(self, cur) -> None: self.add_index_if_missing( cur, "pbf_relayer_candidates", "idx_navsea_tile", "ALTER TABLE pbf_relayer_candidates ADD KEY idx_navsea_tile (z, x, y)", ) self.add_index_if_missing( cur, "pbf_relayer_candidates", "idx_navsea_object", "ALTER TABLE pbf_relayer_candidates ADD KEY idx_navsea_object (canonical_object_type(100), geom_type)", ) self.add_index_if_missing( cur, "pbf_relayer_candidates", "idx_navsea_detection_key", "ALTER TABLE pbf_relayer_candidates ADD KEY idx_navsea_detection_key (detection_key(64))", ) self.add_index_if_missing( cur, "pbf_relayer_candidates", "idx_navsea_source_layer", "ALTER TABLE pbf_relayer_candidates ADD KEY idx_navsea_source_layer (source_layer(100), geom_type)", ) def main() -> None: DetectionPipeline(DbConfig()).run() if __name__ == "__main__": main()