465 lines
13 KiB
Python
465 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import sqlite3
|
|
import subprocess
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
|
|
DB_NAME = "pbf_analysis"
|
|
OUT = Path("out/navsea_kyushu_semantic.sqlite")
|
|
|
|
MYSQL_BASE = [
|
|
"mysql",
|
|
"-h",
|
|
"localhost",
|
|
"-u",
|
|
"root",
|
|
"-p2chi9ks2",
|
|
"-D",
|
|
DB_NAME,
|
|
"-B",
|
|
"-N",
|
|
"--raw",
|
|
"--default-character-set=utf8mb4",
|
|
]
|
|
|
|
COMMON_COLS = """
|
|
d.feature_id,
|
|
gl.fid,
|
|
d.source_layer,
|
|
d.canonical_object_type,
|
|
d.object_family,
|
|
d.detection_class,
|
|
d.capability,
|
|
d.detection_key,
|
|
d.class_name,
|
|
d.shape_name,
|
|
d.layer_name,
|
|
d.z,
|
|
d.x,
|
|
d.y,
|
|
d.geom_type
|
|
"""
|
|
|
|
QUERIES: list[tuple[str, str]] = [
|
|
(
|
|
"land",
|
|
f"""
|
|
SELECT
|
|
{COMMON_COLS}
|
|
FROM navsea_detection_objects d
|
|
JOIN feature_geojson_lookup gl
|
|
ON gl.feature_id = d.feature_id
|
|
WHERE d.canonical_object_type IN ('P陸域', 'P穴')
|
|
""",
|
|
),
|
|
(
|
|
"collision",
|
|
f"""
|
|
SELECT
|
|
{COMMON_COLS}
|
|
FROM navsea_collision_objects d
|
|
JOIN feature_geojson_lookup gl
|
|
ON gl.feature_id = d.feature_id
|
|
""",
|
|
),
|
|
(
|
|
"grounding",
|
|
f"""
|
|
SELECT
|
|
{COMMON_COLS}
|
|
FROM navsea_grounding_objects d
|
|
JOIN feature_geojson_lookup gl
|
|
ON gl.feature_id = d.feature_id
|
|
""",
|
|
),
|
|
(
|
|
"navmark",
|
|
f"""
|
|
SELECT
|
|
{COMMON_COLS}
|
|
FROM navsea_navigation_mark_objects d
|
|
JOIN feature_geojson_lookup gl
|
|
ON gl.feature_id = d.feature_id
|
|
""",
|
|
),
|
|
(
|
|
"boundary",
|
|
f"""
|
|
SELECT
|
|
{COMMON_COLS}
|
|
FROM navsea_boundary_reference_objects d
|
|
JOIN feature_geojson_lookup gl
|
|
ON gl.feature_id = d.feature_id
|
|
""",
|
|
),
|
|
(
|
|
"route",
|
|
f"""
|
|
SELECT
|
|
{COMMON_COLS}
|
|
FROM navsea_route_reference_objects d
|
|
JOIN feature_geojson_lookup gl
|
|
ON gl.feature_id = d.feature_id
|
|
""",
|
|
),
|
|
]
|
|
|
|
|
|
def mysql_stream(query: str):
|
|
proc = subprocess.Popen(
|
|
MYSQL_BASE + ["-e", query],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
bufsize=1,
|
|
)
|
|
assert proc.stdout is not None
|
|
assert proc.stderr is not None
|
|
try:
|
|
for raw_line in proc.stdout:
|
|
line = raw_line.rstrip("\n")
|
|
if not line:
|
|
continue
|
|
yield line.split("\t")
|
|
finally:
|
|
rc = proc.wait()
|
|
if rc != 0:
|
|
raise RuntimeError(proc.stderr.read().strip())
|
|
|
|
|
|
def classify(source_tag: str, canonical_object_type: str, capability: str | None) -> tuple[int, int, int, int, int, int, int, int, int]:
|
|
capability = capability or ""
|
|
is_land = 1 if canonical_object_type == "P陸域" else 0
|
|
is_hole = 1 if canonical_object_type == "P穴" else 0
|
|
is_collision = 1 if source_tag == "collision" or capability == "collision" else 0
|
|
is_grounding = 1 if source_tag == "grounding" or capability == "grounding" else 0
|
|
is_navmark = 1 if source_tag == "navmark" or capability == "navigation_mark" else 0
|
|
is_boundary = 1 if source_tag == "boundary" or capability == "boundary_reference" else 0
|
|
is_route = 1 if source_tag == "route" or capability == "route_reference" else 0
|
|
is_anchorage = 1 if capability == "anchorage_reference" else 0
|
|
is_clearance = 1 if capability == "overhead_clearance" else 0
|
|
return (
|
|
is_land,
|
|
is_hole,
|
|
is_collision,
|
|
is_grounding,
|
|
is_navmark,
|
|
is_boundary,
|
|
is_route,
|
|
is_anchorage,
|
|
is_clearance,
|
|
)
|
|
|
|
|
|
def open_db() -> sqlite3.Connection:
|
|
if OUT.exists():
|
|
OUT.unlink()
|
|
OUT.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
conn = sqlite3.connect(OUT)
|
|
conn.execute("PRAGMA journal_mode=OFF")
|
|
conn.execute("PRAGMA synchronous=OFF")
|
|
conn.execute("PRAGMA temp_store=MEMORY")
|
|
conn.execute("PRAGMA cache_size=200000")
|
|
conn.execute("PRAGMA locking_mode=EXCLUSIVE")
|
|
conn.execute("PRAGMA foreign_keys=OFF")
|
|
return conn
|
|
|
|
|
|
def create_schema(conn: sqlite3.Connection) -> None:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE meta (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE semantic_object (
|
|
feature_id INTEGER PRIMARY KEY,
|
|
fid TEXT,
|
|
source_table TEXT NOT NULL,
|
|
source_layer TEXT,
|
|
canonical_object_type TEXT NOT NULL,
|
|
object_family TEXT NOT NULL,
|
|
detection_class TEXT NOT NULL,
|
|
capability TEXT,
|
|
detection_key TEXT,
|
|
class_name TEXT,
|
|
shape_name TEXT,
|
|
layer_name TEXT,
|
|
z INTEGER,
|
|
x INTEGER,
|
|
y INTEGER,
|
|
geom_type TEXT,
|
|
is_land INTEGER NOT NULL DEFAULT 0,
|
|
is_hole INTEGER NOT NULL DEFAULT 0,
|
|
is_collision INTEGER NOT NULL DEFAULT 0,
|
|
is_grounding INTEGER NOT NULL DEFAULT 0,
|
|
is_navmark INTEGER NOT NULL DEFAULT 0,
|
|
is_boundary INTEGER NOT NULL DEFAULT 0,
|
|
is_route INTEGER NOT NULL DEFAULT 0,
|
|
is_anchorage INTEGER NOT NULL DEFAULT 0,
|
|
is_clearance INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
|
|
CREATE INDEX idx_semantic_object_fid ON semantic_object(fid);
|
|
CREATE INDEX idx_semantic_object_tile ON semantic_object(z, x, y);
|
|
CREATE INDEX idx_semantic_object_capability ON semantic_object(capability, canonical_object_type);
|
|
CREATE INDEX idx_semantic_object_family ON semantic_object(object_family, detection_class);
|
|
|
|
CREATE TABLE grid_cell (
|
|
tile_key TEXT PRIMARY KEY,
|
|
z INTEGER NOT NULL,
|
|
x INTEGER NOT NULL,
|
|
y INTEGER NOT NULL,
|
|
land_count INTEGER NOT NULL DEFAULT 0,
|
|
hole_count INTEGER NOT NULL DEFAULT 0,
|
|
collision_count INTEGER NOT NULL DEFAULT 0,
|
|
grounding_count INTEGER NOT NULL DEFAULT 0,
|
|
navmark_count INTEGER NOT NULL DEFAULT 0,
|
|
boundary_count INTEGER NOT NULL DEFAULT 0,
|
|
route_count INTEGER NOT NULL DEFAULT 0,
|
|
anchorage_count INTEGER NOT NULL DEFAULT 0,
|
|
clearance_count INTEGER NOT NULL DEFAULT 0,
|
|
blocked INTEGER NOT NULL DEFAULT 0,
|
|
blocked_reason TEXT
|
|
);
|
|
|
|
CREATE INDEX idx_grid_cell_zxy ON grid_cell(z, x, y);
|
|
CREATE INDEX idx_grid_cell_blocked ON grid_cell(blocked, z);
|
|
"""
|
|
)
|
|
conn.executemany(
|
|
"INSERT INTO meta(key, value) VALUES (?, ?)",
|
|
[
|
|
("source_db", DB_NAME),
|
|
("source_scope", "九州"),
|
|
("build_time", dt.datetime.now().isoformat(timespec="seconds")),
|
|
("notes", "移动端轻量版语义索引库，保留对象索引和格网占用"),
|
|
(
|
|
"source_filter",
|
|
"land/hole + collision/grounding/navigation_mark/boundary_reference/route_reference",
|
|
),
|
|
],
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def ingest_query(
|
|
conn: sqlite3.Connection,
|
|
source_tag: str,
|
|
query: str,
|
|
seen: dict[int, tuple],
|
|
cells: dict[str, dict[str, int]],
|
|
batch_size: int = 2000,
|
|
) -> int:
|
|
obj_batch: list[tuple] = []
|
|
inserted = 0
|
|
|
|
for cols in mysql_stream(query):
|
|
if len(cols) != 15:
|
|
raise RuntimeError(f"unexpected column count for {source_tag}: {len(cols)}")
|
|
|
|
(
|
|
feature_id,
|
|
fid,
|
|
source_layer,
|
|
canonical_object_type,
|
|
object_family,
|
|
detection_class,
|
|
capability,
|
|
detection_key,
|
|
class_name,
|
|
shape_name,
|
|
layer_name,
|
|
z,
|
|
x,
|
|
y,
|
|
geom_type,
|
|
) = cols
|
|
|
|
feature_id_i = int(feature_id)
|
|
z_i = int(z) if z else 0
|
|
x_i = int(x) if x else 0
|
|
y_i = int(y) if y else 0
|
|
flags = classify(source_tag, canonical_object_type, capability)
|
|
|
|
prev = seen.get(feature_id_i)
|
|
row = (
|
|
feature_id_i,
|
|
fid or None,
|
|
source_tag,
|
|
source_layer,
|
|
canonical_object_type,
|
|
object_family,
|
|
detection_class,
|
|
capability,
|
|
detection_key,
|
|
class_name,
|
|
shape_name,
|
|
layer_name,
|
|
z_i,
|
|
x_i,
|
|
y_i,
|
|
geom_type,
|
|
*flags,
|
|
)
|
|
|
|
if prev is None:
|
|
seen[feature_id_i] = row
|
|
obj_batch.append(row)
|
|
inserted += 1
|
|
else:
|
|
merged = list(prev)
|
|
for idx, flag in enumerate(flags, start=16):
|
|
merged[idx] = int(merged[idx]) or int(flag)
|
|
seen[feature_id_i] = tuple(merged)
|
|
|
|
tile_key = f"{z_i}/{x_i}/{y_i}"
|
|
counts = cells.get(tile_key)
|
|
if counts is None:
|
|
counts = {
|
|
"z": z_i,
|
|
"x": x_i,
|
|
"y": y_i,
|
|
"land_count": 0,
|
|
"hole_count": 0,
|
|
"collision_count": 0,
|
|
"grounding_count": 0,
|
|
"navmark_count": 0,
|
|
"boundary_count": 0,
|
|
"route_count": 0,
|
|
"anchorage_count": 0,
|
|
"clearance_count": 0,
|
|
}
|
|
cells[tile_key] = counts
|
|
counts["land_count"] += flags[0]
|
|
counts["hole_count"] += flags[1]
|
|
counts["collision_count"] += flags[2]
|
|
counts["grounding_count"] += flags[3]
|
|
counts["navmark_count"] += flags[4]
|
|
counts["boundary_count"] += flags[5]
|
|
counts["route_count"] += flags[6]
|
|
counts["anchorage_count"] += flags[7]
|
|
counts["clearance_count"] += flags[8]
|
|
|
|
if len(obj_batch) >= batch_size:
|
|
conn.executemany(
|
|
"""
|
|
INSERT OR REPLACE INTO semantic_object(
|
|
feature_id, fid, source_table, source_layer, canonical_object_type,
|
|
object_family, detection_class, capability, detection_key, class_name,
|
|
shape_name, layer_name, z, x, y, geom_type, is_land, is_hole,
|
|
is_collision, is_grounding, is_navmark, is_boundary, is_route,
|
|
is_anchorage, is_clearance
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
obj_batch,
|
|
)
|
|
conn.commit()
|
|
obj_batch.clear()
|
|
print(f"[{source_tag}] 已写入 {inserted} 条对象")
|
|
|
|
if obj_batch:
|
|
conn.executemany(
|
|
"""
|
|
INSERT OR REPLACE INTO semantic_object(
|
|
feature_id, fid, source_table, source_layer, canonical_object_type,
|
|
object_family, detection_class, capability, detection_key, class_name,
|
|
shape_name, layer_name, z, x, y, geom_type, is_land, is_hole,
|
|
is_collision, is_grounding, is_navmark, is_boundary, is_route,
|
|
is_anchorage, is_clearance
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
obj_batch,
|
|
)
|
|
conn.commit()
|
|
|
|
return inserted
|
|
|
|
|
|
def finalize_grid(conn: sqlite3.Connection, cells: dict[str, dict[str, int]]) -> int:
|
|
rows = []
|
|
for tile_key, c in cells.items():
|
|
blocked = 0
|
|
reason = None
|
|
if c["land_count"] > c["hole_count"] and c["land_count"] > 0:
|
|
blocked = 1
|
|
reason = "land"
|
|
elif c["collision_count"] > 0:
|
|
blocked = 1
|
|
reason = "collision"
|
|
elif c["grounding_count"] > 0:
|
|
blocked = 1
|
|
reason = "grounding"
|
|
elif c["clearance_count"] > 0:
|
|
blocked = 1
|
|
reason = "clearance"
|
|
|
|
rows.append(
|
|
(
|
|
tile_key,
|
|
c["z"],
|
|
c["x"],
|
|
c["y"],
|
|
c["land_count"],
|
|
c["hole_count"],
|
|
c["collision_count"],
|
|
c["grounding_count"],
|
|
c["navmark_count"],
|
|
c["boundary_count"],
|
|
c["route_count"],
|
|
c["anchorage_count"],
|
|
c["clearance_count"],
|
|
blocked,
|
|
reason,
|
|
)
|
|
)
|
|
|
|
conn.executemany(
|
|
"""
|
|
INSERT INTO grid_cell(
|
|
tile_key, z, x, y, land_count, hole_count, collision_count, grounding_count,
|
|
navmark_count, boundary_count, route_count, anchorage_count, clearance_count,
|
|
blocked, blocked_reason
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
rows,
|
|
)
|
|
conn.commit()
|
|
return len(rows)
|
|
|
|
|
|
def main() -> None:
|
|
conn = open_db()
|
|
create_schema(conn)
|
|
|
|
seen: dict[int, tuple] = {}
|
|
cells: dict[str, dict[str, int]] = {}
|
|
total = 0
|
|
|
|
for source_tag, query in QUERIES:
|
|
print(f"开始导出 {source_tag} ...")
|
|
total += ingest_query(conn, source_tag, query, seen, cells)
|
|
print(f"{source_tag} 完成，累计对象 {total}")
|
|
|
|
cell_total = finalize_grid(conn, cells)
|
|
conn.execute("ANALYZE")
|
|
conn.commit()
|
|
|
|
obj_count = conn.execute("SELECT COUNT(*) FROM semantic_object").fetchone()[0]
|
|
cell_count = conn.execute("SELECT COUNT(*) FROM grid_cell").fetchone()[0]
|
|
conn.close()
|
|
|
|
print(f"完成: objects={obj_count}, cells={cell_count}, grid_rows={cell_total}")
|
|
print(f"输出文件: {OUT.resolve()}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|