Files
pbf/navsea_audit.py
2026-03-17 19:48:15 +08:00

630 lines
25 KiB
Python

import csv
import json
import os
import tempfile
import time
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
import pymysql
ROOT = Path(__file__).resolve().parent
STYLE_JSON_PATH = ROOT / "src" / "pbf" / "style.json"
REPORT_PATH = ROOT / "navsea_classification_report.md"
@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")
class NavSeaAudit:
def __init__(self, db_config: DbConfig) -> None:
self.db_config = db_config
self.metrics: dict[str, int | float | str] = {}
@contextmanager
def connect(self, *, streaming: bool = False, local_infile: bool = False):
kwargs = {
"host": self.db_config.host,
"port": self.db_config.port,
"user": self.db_config.user,
"password": self.db_config.password,
"database": self.db_config.database,
"charset": "utf8mb4",
"autocommit": False,
"local_infile": local_infile,
"cursorclass": pymysql.cursors.SSCursor if streaming else pymysql.cursors.Cursor,
}
if self.db_config.unix_socket and self.db_config.host in {"localhost", "127.0.0.1"}:
kwargs["unix_socket"] = self.db_config.unix_socket
connection = pymysql.connect(**kwargs)
try:
yield connection
connection.commit()
except Exception:
connection.rollback()
raise
finally:
connection.close()
def run(self) -> None:
start = time.time()
self.validate_inputs()
self.parse_at_attributes()
self.build_feature_semantic()
self.build_object_catalog()
self.build_geometry_consistency()
self.build_object_type_candidates()
self.build_object_type_stats()
self.build_style_layers()
self.build_style_mapping()
self.build_tile_density()
self.build_anomaly_tables()
self.generate_report()
self.metrics["pipeline_runtime_seconds"] = round(time.time() - start, 2)
print(f"Pipeline completed in {self.metrics['pipeline_runtime_seconds']} seconds")
def validate_inputs(self) -> None:
if not STYLE_JSON_PATH.exists():
raise FileNotFoundError(f"style.json not found: {STYLE_JSON_PATH}")
with self.connect() as conn:
with conn.cursor() as cur:
for table in ("features", "properties"):
cur.execute("SHOW TABLES LIKE %s", (table,))
if cur.fetchone() is None:
raise RuntimeError(f"required table missing: {table}")
cur.execute("SELECT COUNT(*) FROM features")
self.metrics["source_feature_count"] = cur.fetchone()[0]
cur.execute("SELECT COUNT(*) FROM properties")
self.metrics["source_property_count"] = cur.fetchone()[0]
cur.execute("SELECT COUNT(*) FROM properties WHERE k='at'")
self.metrics["source_at_count"] = cur.fetchone()[0]
print(
"Validated inputs:",
self.metrics["source_feature_count"],
"features,",
self.metrics["source_property_count"],
"properties,",
self.metrics["source_at_count"],
"at rows",
)
def parse_at_attributes(self) -> None:
print("Parsing at attributes...")
parsed_rows = 0
source_rows = 0
parse_errors = 0
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
newline="",
suffix=".tsv",
delete=False,
dir="/tmp",
) as temp_file:
temp_path = Path(temp_file.name)
writer = csv.writer(
temp_file,
delimiter="\t",
quotechar='"',
escapechar="\\",
lineterminator="\n",
)
with self.connect(streaming=True) as conn:
with conn.cursor() as cur:
cur.execute("SELECT feature_id, v FROM properties WHERE k='at'")
for feature_id, raw_value in cur:
source_rows += 1
try:
pairs = json.loads(raw_value)
except json.JSONDecodeError:
parse_errors += 1
continue
if not isinstance(pairs, list):
parse_errors += 1
continue
for pair in pairs:
if not isinstance(pair, list) or len(pair) != 2:
parse_errors += 1
continue
key, value = pair
writer.writerow(
(
int(feature_id),
"" if key is None else str(key),
"" if value is None else str(value),
)
)
parsed_rows += 1
with self.connect(local_infile=True) as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS at_attributes")
cur.execute(
"""
CREATE TABLE at_attributes (
feature_id BIGINT NOT NULL,
k VARCHAR(100) NOT NULL,
v TEXT NULL,
KEY idx_feature_id (feature_id),
KEY idx_k (k),
KEY idx_feature_k (feature_id, k)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
"""
)
cur.execute(
"""
LOAD DATA LOCAL INFILE %s
INTO TABLE at_attributes
CHARACTER SET utf8mb4
FIELDS TERMINATED BY '\t'
ENCLOSED BY '"'
ESCAPED BY '\\\\'
LINES TERMINATED BY '\n'
(feature_id, k, v)
""",
(str(temp_path),),
)
temp_path.unlink(missing_ok=True)
self.metrics["at_source_rows"] = source_rows
self.metrics["at_attribute_rows"] = parsed_rows
self.metrics["at_parse_errors"] = parse_errors
print(
"Parsed at attributes:",
source_rows,
"source rows ->",
parsed_rows,
"attribute rows,",
parse_errors,
"parse errors",
)
def build_feature_semantic(self) -> None:
print("Building feature_semantic...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS feature_semantic")
cur.execute(
"""
CREATE TABLE feature_semantic AS
SELECT
f.id AS feature_id,
f.z,
f.x,
f.y,
f.vt_layer,
f.geom_type,
MAX(CASE WHEN a.k='分類' THEN a.v END) AS class_name,
MAX(CASE WHEN a.k='形状分類' THEN a.v END) AS shape_name,
MAX(CASE WHEN a.k='レイヤ' THEN a.v END) AS layer_name
FROM features f
LEFT JOIN at_attributes a
ON f.id = a.feature_id
GROUP BY
f.id,
f.z,
f.x,
f.y,
f.vt_layer,
f.geom_type
"""
)
cur.execute("ALTER TABLE feature_semantic ADD PRIMARY KEY (feature_id)")
cur.execute("ALTER TABLE feature_semantic ADD KEY idx_vt_layer (vt_layer)")
cur.execute("ALTER TABLE feature_semantic ADD KEY idx_class_name (class_name(100))")
cur.execute("ALTER TABLE feature_semantic ADD KEY idx_geom_type (geom_type)")
cur.execute("SELECT COUNT(*) FROM feature_semantic")
self.metrics["feature_semantic_rows"] = cur.fetchone()[0]
def build_object_catalog(self) -> None:
print("Building object_catalog...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS object_catalog")
cur.execute(
"""
CREATE TABLE object_catalog AS
SELECT
layer_name,
class_name,
shape_name,
vt_layer,
geom_type,
COUNT(*) AS feature_count
FROM feature_semantic
GROUP BY
layer_name,
class_name,
shape_name,
vt_layer,
geom_type
"""
)
cur.execute("ALTER TABLE object_catalog ADD KEY idx_class_name (class_name(100))")
cur.execute("ALTER TABLE object_catalog ADD KEY idx_vt_layer (vt_layer)")
cur.execute("SELECT COUNT(*) FROM object_catalog")
self.metrics["object_catalog_rows"] = cur.fetchone()[0]
def build_geometry_consistency(self) -> None:
print("Building geometry_consistency...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS geometry_consistency")
cur.execute(
"""
CREATE TABLE geometry_consistency AS
SELECT
class_name,
geom_type,
COUNT(*) AS feature_count
FROM feature_semantic
GROUP BY class_name, geom_type
"""
)
cur.execute("ALTER TABLE geometry_consistency ADD KEY idx_class_name (class_name(100))")
cur.execute("SELECT COUNT(*) FROM geometry_consistency")
self.metrics["geometry_consistency_rows"] = cur.fetchone()[0]
def build_object_type_candidates(self) -> None:
print("Building object_type_candidates...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS object_type_candidates")
cur.execute(
"""
CREATE TABLE object_type_candidates AS
SELECT
feature_id,
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 AS object_type_source,
COALESCE(NULLIF(class_name, ''), NULLIF(shape_name, ''), vt_layer) AS object_type,
vt_layer,
geom_type
FROM feature_semantic
"""
)
cur.execute("ALTER TABLE object_type_candidates ADD PRIMARY KEY (feature_id)")
cur.execute("ALTER TABLE object_type_candidates ADD KEY idx_object_type (object_type(100))")
cur.execute("ALTER TABLE object_type_candidates ADD KEY idx_vt_layer (vt_layer)")
cur.execute("SELECT COUNT(*) FROM object_type_candidates")
self.metrics["object_type_candidate_rows"] = cur.fetchone()[0]
def build_object_type_stats(self) -> None:
print("Building object_type_stats...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS object_type_stats")
cur.execute(
"""
CREATE TABLE object_type_stats AS
SELECT
object_type,
geom_type,
COUNT(*) AS feature_count
FROM object_type_candidates
GROUP BY object_type, geom_type
"""
)
cur.execute("ALTER TABLE object_type_stats ADD KEY idx_object_type (object_type(100))")
cur.execute("SELECT COUNT(*) FROM object_type_stats")
self.metrics["object_type_stats_rows"] = cur.fetchone()[0]
cur.execute("SELECT COUNT(DISTINCT object_type) FROM object_type_candidates")
self.metrics["distinct_object_types"] = cur.fetchone()[0]
def build_style_layers(self) -> None:
print("Building style_layers from style.json...")
style_data = json.loads(STYLE_JSON_PATH.read_text(encoding="utf-8"))
rows = []
for layer in style_data.get("layers", []):
paint = layer.get("paint", {})
layout = layer.get("layout", {})
rows.append(
(
layer.get("id"),
layer.get("source-layer"),
self.stringify_style_value(layout.get("icon-image")),
self.stringify_style_value(paint.get("line-color")),
self.stringify_style_value(paint.get("fill-color")),
layer.get("type"),
)
)
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS style_layers")
cur.execute(
"""
CREATE TABLE style_layers (
layer_id VARCHAR(200) NOT NULL,
source_layer VARCHAR(200) NULL,
icon VARCHAR(2000) NULL,
line_color VARCHAR(2000) NULL,
fill_color VARCHAR(2000) NULL,
layer_type VARCHAR(50) NULL,
KEY idx_source_layer (source_layer),
KEY idx_layer_id (layer_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
"""
)
cur.executemany(
"""
INSERT INTO style_layers
(layer_id, source_layer, icon, line_color, fill_color, layer_type)
VALUES (%s, %s, %s, %s, %s, %s)
""",
rows,
)
self.metrics["style_layer_rows"] = len(rows)
def build_style_mapping(self) -> None:
print("Building style_mapping...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS style_mapping")
cur.execute(
"""
CREATE TABLE style_mapping AS
SELECT
s.layer_id,
s.source_layer,
s.layer_type,
s.icon,
s.line_color,
s.fill_color,
o.object_type,
o.geom_type,
COUNT(*) AS feature_count
FROM style_layers s
JOIN object_type_candidates o
ON s.source_layer = o.object_type
GROUP BY
s.layer_id,
s.source_layer,
s.layer_type,
s.icon,
s.line_color,
s.fill_color,
o.object_type,
o.geom_type
"""
)
cur.execute("ALTER TABLE style_mapping ADD KEY idx_object_type (object_type(100))")
cur.execute("SELECT COUNT(*) FROM style_mapping")
self.metrics["style_mapping_rows"] = cur.fetchone()[0]
def build_tile_density(self) -> None:
print("Building tile_density, tile_layer_density and tile_density_top100...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS tile_density")
cur.execute(
"""
CREATE TABLE tile_density AS
SELECT
f.z,
f.x,
f.y,
COUNT(DISTINCT f.id) AS feature_count
FROM features AS f
GROUP BY
f.z,
f.x,
f.y
"""
)
cur.execute("ALTER TABLE tile_density ADD KEY idx_tile (z, x, y)")
cur.execute("ALTER TABLE tile_density ADD KEY idx_feature_count (feature_count)")
cur.execute("DROP TABLE IF EXISTS tile_layer_density")
cur.execute(
"""
CREATE TABLE tile_layer_density AS
SELECT
f.z,
f.x,
f.y,
f.vt_layer,
COUNT(DISTINCT f.id) AS feature_count
FROM features AS f
GROUP BY
f.z,
f.x,
f.y,
f.vt_layer
"""
)
cur.execute("ALTER TABLE tile_layer_density ADD KEY idx_tile (z, x, y)")
cur.execute("ALTER TABLE tile_layer_density ADD KEY idx_layer (vt_layer)")
cur.execute("ALTER TABLE tile_layer_density ADD KEY idx_feature_count (feature_count)")
cur.execute("DROP TABLE IF EXISTS tile_density_top100")
cur.execute(
"""
CREATE TABLE tile_density_top100 AS
SELECT *
FROM tile_density
ORDER BY feature_count DESC
LIMIT 100
"""
)
cur.execute("SELECT COUNT(*) FROM tile_density")
self.metrics["tile_density_rows"] = cur.fetchone()[0]
cur.execute("SELECT COUNT(*) FROM tile_layer_density")
self.metrics["tile_layer_density_rows"] = cur.fetchone()[0]
def build_anomaly_tables(self) -> None:
print("Building anomaly tables...")
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS anomaly_navigation_geom")
cur.execute(
"""
CREATE TABLE anomaly_navigation_geom AS
SELECT *
FROM feature_semantic
WHERE class_name='灯台'
AND geom_type <> 'Point'
"""
)
cur.execute("DROP TABLE IF EXISTS anomaly_reef_geom")
cur.execute(
"""
CREATE TABLE anomaly_reef_geom AS
SELECT *
FROM feature_semantic
WHERE class_name='魚礁'
AND geom_type NOT IN ('Point', 'Polygon')
"""
)
cur.execute("SELECT COUNT(*) FROM anomaly_navigation_geom")
self.metrics["anomaly_navigation_geom_rows"] = cur.fetchone()[0]
cur.execute("SELECT COUNT(*) FROM anomaly_reef_geom")
self.metrics["anomaly_reef_geom_rows"] = cur.fetchone()[0]
def generate_report(self) -> None:
print("Generating navsea_classification_report.md...")
total_features = self.scalar("SELECT COUNT(*) FROM features")
total_object_types = self.scalar("SELECT COUNT(DISTINCT object_type) FROM object_type_candidates")
top_object_types = self.fetchall(
"""
SELECT object_type, geom_type, feature_count
FROM object_type_stats
ORDER BY feature_count DESC
LIMIT 50
"""
)
geometry_consistency = self.fetchall(
"""
SELECT class_name, geom_type, feature_count
FROM geometry_consistency
ORDER BY feature_count DESC
LIMIT 100
"""
)
style_mapping = self.fetchall(
"""
SELECT layer_id, source_layer, object_type, geom_type, feature_count
FROM style_mapping
ORDER BY feature_count DESC
LIMIT 100
"""
)
tile_density = self.fetchall(
"""
SELECT z, x, y, feature_count
FROM tile_density_top100
ORDER BY feature_count DESC, z, x, y
"""
)
source_breakdown = self.fetchall(
"""
SELECT object_type_source, COUNT(*) AS feature_count
FROM object_type_candidates
GROUP BY object_type_source
ORDER BY feature_count DESC
"""
)
lines = [
"# NavSea Classification Report",
"",
"## Dataset Summary",
"",
f"- Total features: {total_features}",
f"- Total properties: {self.metrics['source_property_count']}",
f"- Total `at` rows: {self.metrics['source_at_count']}",
f"- Parsed `at` attribute rows: {self.metrics['at_attribute_rows']}",
f"- `at` parse errors: {self.metrics['at_parse_errors']}",
f"- Total object types: {total_object_types}",
f"- Style layers parsed: {self.metrics['style_layer_rows']}",
f"- Style mappings found: {self.metrics['style_mapping_rows']}",
"",
"## Object Type Source Breakdown",
"",
self.render_table(
["object_type_source", "feature_count"],
source_breakdown,
),
"",
"## Top Object Types",
"",
self.render_table(["object_type", "geom_type", "feature_count"], top_object_types),
"",
"## Geometry Consistency",
"",
self.render_table(["class_name", "geom_type", "feature_count"], geometry_consistency),
"",
"## Style Mapping",
"",
self.render_table(
["layer_id", "source_layer", "object_type", "geom_type", "feature_count"],
style_mapping,
),
"",
"## Tile Density Top 100",
"",
self.render_table(["z", "x", "y", "feature_count"], tile_density),
"",
"## Spatial Sanity Checks",
"",
f"- anomaly_navigation_geom rows: {self.metrics['anomaly_navigation_geom_rows']}",
f"- anomaly_reef_geom rows: {self.metrics['anomaly_reef_geom_rows']}",
"",
]
REPORT_PATH.write_text("\n".join(lines), encoding="utf-8")
def scalar(self, sql: str):
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute(sql)
return cur.fetchone()[0]
def fetchall(self, sql: str):
with self.connect() as conn:
with conn.cursor() as cur:
cur.execute(sql)
return cur.fetchall()
@staticmethod
def stringify_style_value(value):
if value is None:
return None
if isinstance(value, str):
return value
return json.dumps(value, ensure_ascii=False)
@staticmethod
def render_table(headers, rows) -> str:
lines = [
"| " + " | ".join(headers) + " |",
"| " + " | ".join(["---"] * len(headers)) + " |",
]
for row in rows:
lines.append("| " + " | ".join("" if value is None else str(value) for value in row) + " |")
return "\n".join(lines)
def main() -> None:
NavSeaAudit(DbConfig()).run()
if __name__ == "__main__":
main()