451 lines
14 KiB
Python
451 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime as dt
|
||
import json
|
||
import math
|
||
import sqlite3
|
||
from dataclasses import asdict, dataclass
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
|
||
import pymysql
|
||
|
||
|
||
DB_USER = "root"
|
||
DB_PASSWORD = "2chi9ks2"
|
||
DB_HOST = "localhost"
|
||
DB_SOCKET = "/tmp/mysql.sock"
|
||
|
||
DEFAULT_OUTPUT_DB = Path("out/navsea_national_navigation_fusion.sqlite")
|
||
DEFAULT_MANIFEST = Path("out/navsea_national_navigation_fusion.manifest.json")
|
||
|
||
DEFAULT_SOURCES = (
|
||
{
|
||
"source_id": "coast_200m",
|
||
"layer_key": "coast_200m",
|
||
"grid_level": 200,
|
||
"role": "base_coast",
|
||
"description": "全国海岸线 200m 硬阻塞格",
|
||
"source_type": "mysql_grid",
|
||
"database_name": "navsea_japan_coast_grid",
|
||
"source_table": "navsea_grid_cell",
|
||
"source_layer": "coast_200m",
|
||
"source_path": "coastline/C23-06_*_GML.zip",
|
||
"notes": "全国 200x200 海岸线底盘",
|
||
},
|
||
{
|
||
"source_id": "fish_port_20m",
|
||
"layer_key": "fish_port_20m",
|
||
"grid_level": 20,
|
||
"role": "port_detail",
|
||
"description": "全国渔港 20m 陆基/海面格",
|
||
"source_type": "mysql_grid",
|
||
"database_name": "navsea_fukuoka_saga_grid",
|
||
"source_table": "navsea_grid_cell",
|
||
"source_layer": "fish_port_20m",
|
||
"source_path": "coastline/C09-06.zip",
|
||
"notes": "当前仓库里的渔港 20m 生成结果",
|
||
},
|
||
{
|
||
"source_id": "hazard_50m",
|
||
"layer_key": "hazard_50m",
|
||
"grid_level": 50,
|
||
"role": "overlay",
|
||
"description": "海上障碍 50m 黄格",
|
||
"source_type": "mysql_grid",
|
||
"database_name": "navsea_fukuoka_saga_grid",
|
||
"source_table": "navsea_grid_cell",
|
||
"source_layer": "hazard_50m",
|
||
"source_path": "pbf-delivery-kyushu-reencoded z12",
|
||
"notes": "当前仓库里的障碍叠加层",
|
||
},
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SourceSpec:
|
||
source_id: str
|
||
layer_key: str
|
||
grid_level: int
|
||
role: str
|
||
description: str
|
||
source_type: str
|
||
database_name: str
|
||
source_table: str
|
||
source_layer: str
|
||
source_path: str
|
||
notes: str = ""
|
||
|
||
|
||
def mysql_connect(database: str):
|
||
kwargs = {
|
||
"host": DB_HOST,
|
||
"user": DB_USER,
|
||
"password": DB_PASSWORD,
|
||
"charset": "utf8mb4",
|
||
"autocommit": False,
|
||
"cursorclass": pymysql.cursors.SSCursor,
|
||
}
|
||
if DB_SOCKET:
|
||
kwargs["unix_socket"] = DB_SOCKET
|
||
kwargs["database"] = database
|
||
return pymysql.connect(**kwargs)
|
||
|
||
|
||
def open_output_db(path: Path, recreate: bool) -> sqlite3.Connection:
|
||
if recreate and path.exists():
|
||
path.unlink()
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
conn = sqlite3.connect(path)
|
||
conn.execute("PRAGMA journal_mode=WAL")
|
||
conn.execute("PRAGMA synchronous=OFF")
|
||
conn.execute("PRAGMA temp_store=MEMORY")
|
||
conn.execute("PRAGMA cache_size=200000")
|
||
conn.execute("PRAGMA busy_timeout=5000")
|
||
return conn
|
||
|
||
|
||
def create_schema(conn: sqlite3.Connection) -> None:
|
||
conn.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS meta (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS source_registry (
|
||
source_id TEXT PRIMARY KEY,
|
||
source_type TEXT NOT NULL,
|
||
database_name TEXT NOT NULL,
|
||
source_table TEXT NOT NULL,
|
||
source_layer TEXT NOT NULL,
|
||
source_path TEXT NOT NULL,
|
||
status TEXT NOT NULL,
|
||
notes TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS layer_registry (
|
||
layer_key TEXT PRIMARY KEY,
|
||
grid_level INTEGER NOT NULL,
|
||
role TEXT NOT NULL,
|
||
description TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS layer_summary (
|
||
layer_key TEXT PRIMARY KEY,
|
||
source_id TEXT NOT NULL,
|
||
database_name TEXT NOT NULL,
|
||
source_table TEXT NOT NULL,
|
||
source_layer TEXT NOT NULL,
|
||
grid_level INTEGER NOT NULL,
|
||
cell_size_m REAL NOT NULL,
|
||
cell_count INTEGER NOT NULL,
|
||
bbox_min_lon REAL,
|
||
bbox_min_lat REAL,
|
||
bbox_max_lon REAL,
|
||
bbox_max_lat REAL,
|
||
generated_at TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS fusion_cell (
|
||
layer_key TEXT NOT NULL,
|
||
cell_id TEXT NOT NULL,
|
||
source_id TEXT NOT NULL,
|
||
source_type TEXT NOT NULL,
|
||
database_name TEXT NOT NULL,
|
||
source_table TEXT NOT NULL,
|
||
source_layer TEXT NOT NULL,
|
||
grid_level INTEGER NOT NULL,
|
||
row_idx INTEGER NOT NULL,
|
||
col_idx INTEGER NOT NULL,
|
||
cell_size_m REAL NOT NULL,
|
||
state_name TEXT NOT NULL,
|
||
class_name TEXT NOT NULL,
|
||
source_name TEXT NOT NULL,
|
||
min_lon REAL NOT NULL,
|
||
min_lat REAL NOT NULL,
|
||
max_lon REAL NOT NULL,
|
||
max_lat REAL NOT NULL,
|
||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
PRIMARY KEY (layer_key, cell_id)
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_fusion_cell_layer_grid
|
||
ON fusion_cell(layer_key, grid_level);
|
||
CREATE INDEX IF NOT EXISTS idx_fusion_cell_layer_state
|
||
ON fusion_cell(layer_key, state_name);
|
||
CREATE INDEX IF NOT EXISTS idx_fusion_cell_bbox
|
||
ON fusion_cell(layer_key, min_lon, min_lat, max_lon, max_lat);
|
||
"""
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def seed_registry(conn: sqlite3.Connection, sources: Iterable[SourceSpec]) -> None:
|
||
conn.executemany(
|
||
"INSERT OR REPLACE INTO source_registry VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
||
[
|
||
(
|
||
src.source_id,
|
||
src.source_type,
|
||
src.database_name,
|
||
src.source_table,
|
||
src.source_layer,
|
||
src.source_path,
|
||
"ready",
|
||
src.notes,
|
||
)
|
||
for src in sources
|
||
],
|
||
)
|
||
conn.executemany(
|
||
"INSERT OR REPLACE INTO layer_registry VALUES (?, ?, ?, ?)",
|
||
[
|
||
(
|
||
src.layer_key,
|
||
src.grid_level,
|
||
src.role,
|
||
src.description,
|
||
)
|
||
for src in sources
|
||
],
|
||
)
|
||
conn.executemany(
|
||
"INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)",
|
||
[
|
||
("fusion_version", "v1"),
|
||
("fusion_scope", "japan"),
|
||
("fusion_note", "全国 200m / 20m / 50m 障碍层统一汇聚"),
|
||
("source_count", str(len(tuple(sources)))),
|
||
("generated_at", dt.datetime.now().isoformat(timespec="seconds")),
|
||
("layer_keys", json.dumps([src.layer_key for src in sources], ensure_ascii=False)),
|
||
],
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def insert_batch(conn: sqlite3.Connection, rows: list[tuple]) -> None:
|
||
conn.executemany(
|
||
"""
|
||
INSERT OR REPLACE INTO fusion_cell(
|
||
layer_key, cell_id, source_id, source_type, database_name, source_table, source_layer,
|
||
grid_level, row_idx, col_idx, cell_size_m, state_name, class_name, source_name,
|
||
min_lon, min_lat, max_lon, max_lat
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
rows,
|
||
)
|
||
|
||
|
||
def import_layer(conn: sqlite3.Connection, spec: SourceSpec, *, batch_size: int = 5000) -> dict[str, object]:
|
||
mysql_conn = mysql_connect(spec.database_name)
|
||
try:
|
||
with mysql_conn.cursor() as cur:
|
||
cur.execute(
|
||
f"""
|
||
SELECT cell_id, row_idx, col_idx, cell_size_m, state_name, class_name,
|
||
source_name, min_lon, min_lat, max_lon, max_lat
|
||
FROM {spec.source_table}
|
||
WHERE layer_name=%s
|
||
ORDER BY row_idx, col_idx
|
||
""",
|
||
(spec.source_layer,),
|
||
)
|
||
|
||
count = 0
|
||
batch: list[tuple] = []
|
||
bbox_min_lon = bbox_min_lat = math.inf
|
||
bbox_max_lon = bbox_max_lat = -math.inf
|
||
|
||
for row in cur:
|
||
(
|
||
cell_id,
|
||
row_idx,
|
||
col_idx,
|
||
cell_size_m,
|
||
state_name,
|
||
class_name,
|
||
source_name,
|
||
min_lon,
|
||
min_lat,
|
||
max_lon,
|
||
max_lat,
|
||
) = row
|
||
batch.append(
|
||
(
|
||
spec.layer_key,
|
||
str(cell_id),
|
||
spec.source_id,
|
||
spec.source_type,
|
||
spec.database_name,
|
||
spec.source_table,
|
||
spec.source_layer,
|
||
int(spec.grid_level),
|
||
int(row_idx),
|
||
int(col_idx),
|
||
float(cell_size_m),
|
||
str(state_name),
|
||
str(class_name),
|
||
str(source_name),
|
||
float(min_lon),
|
||
float(min_lat),
|
||
float(max_lon),
|
||
float(max_lat),
|
||
)
|
||
)
|
||
bbox_min_lon = min(bbox_min_lon, float(min_lon))
|
||
bbox_min_lat = min(bbox_min_lat, float(min_lat))
|
||
bbox_max_lon = max(bbox_max_lon, float(max_lon))
|
||
bbox_max_lat = max(bbox_max_lat, float(max_lat))
|
||
count += 1
|
||
|
||
if len(batch) >= batch_size:
|
||
insert_batch(conn, batch)
|
||
conn.commit()
|
||
batch.clear()
|
||
|
||
if batch:
|
||
insert_batch(conn, batch)
|
||
conn.commit()
|
||
|
||
finally:
|
||
mysql_conn.close()
|
||
|
||
bbox: tuple[float, float, float, float] | None
|
||
if count == 0:
|
||
bbox = None
|
||
else:
|
||
bbox = (bbox_min_lon, bbox_min_lat, bbox_max_lon, bbox_max_lat)
|
||
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT OR REPLACE INTO layer_summary(
|
||
layer_key, source_id, database_name, source_table, source_layer,
|
||
grid_level, cell_size_m, cell_count, bbox_min_lon, bbox_min_lat,
|
||
bbox_max_lon, bbox_max_lat, generated_at
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
spec.layer_key,
|
||
spec.source_id,
|
||
spec.database_name,
|
||
spec.source_table,
|
||
spec.source_layer,
|
||
spec.grid_level,
|
||
float(spec.grid_level),
|
||
count,
|
||
bbox[0] if bbox else None,
|
||
bbox[1] if bbox else None,
|
||
bbox[2] if bbox else None,
|
||
bbox[3] if bbox else None,
|
||
dt.datetime.now().isoformat(timespec="seconds"),
|
||
),
|
||
)
|
||
conn.commit()
|
||
|
||
return {
|
||
"source_id": spec.source_id,
|
||
"layer_key": spec.layer_key,
|
||
"grid_level": spec.grid_level,
|
||
"count": count,
|
||
"bbox_lonlat": list(bbox) if bbox else None,
|
||
"database_name": spec.database_name,
|
||
"source_table": spec.source_table,
|
||
"source_layer": spec.source_layer,
|
||
"cell_size_m": float(spec.grid_level),
|
||
"source_path": spec.source_path,
|
||
}
|
||
|
||
|
||
def build_manifest(conn: sqlite3.Connection, import_summaries: list[dict[str, object]]) -> dict[str, object]:
|
||
meta = dict(conn.execute("SELECT key, value FROM meta").fetchall())
|
||
layer_rows = conn.execute(
|
||
"""
|
||
SELECT layer_key, grid_level, role, description
|
||
FROM layer_registry
|
||
ORDER BY grid_level, layer_key
|
||
"""
|
||
).fetchall()
|
||
return {
|
||
"meta": meta,
|
||
"layers": [
|
||
{
|
||
"layer_key": row[0],
|
||
"grid_level": row[1],
|
||
"role": row[2],
|
||
"description": row[3],
|
||
}
|
||
for row in layer_rows
|
||
],
|
||
"imports": import_summaries,
|
||
"cell_counts": [
|
||
{
|
||
"layer_key": row[0],
|
||
"count": row[1],
|
||
}
|
||
for row in conn.execute(
|
||
"SELECT layer_key, COUNT(*) FROM fusion_cell GROUP BY layer_key ORDER BY layer_key"
|
||
).fetchall()
|
||
],
|
||
}
|
||
|
||
|
||
def write_template(path: Path) -> None:
|
||
payload = {
|
||
"fusion_version": "v1",
|
||
"sources": [asdict(SourceSpec(**spec)) for spec in DEFAULT_SOURCES],
|
||
"notes": "这个模板只是记录默认汇聚层，不会执行导入。",
|
||
}
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="把全国 200m / 20m / 50m 障碍层汇聚到一个 SQLite 数据库。")
|
||
parser.add_argument("--output-db", type=Path, default=DEFAULT_OUTPUT_DB)
|
||
parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
|
||
parser.add_argument("--recreate", action="store_true", help="重建输出数据库")
|
||
parser.add_argument("--emit-template", action="store_true", help="只输出默认模板 JSON，不执行汇聚")
|
||
parser.add_argument("--batch-size", type=int, default=5000, help="每批写入多少条格网记录")
|
||
args = parser.parse_args()
|
||
|
||
if args.emit_template:
|
||
write_template(args.manifest)
|
||
print(f"已写出模板: {args.manifest}")
|
||
return
|
||
|
||
sources = [SourceSpec(**spec) for spec in DEFAULT_SOURCES]
|
||
|
||
conn = open_output_db(args.output_db, recreate=args.recreate)
|
||
try:
|
||
create_schema(conn)
|
||
seed_registry(conn, sources)
|
||
|
||
import_summaries = []
|
||
for spec in sources:
|
||
summary = import_layer(conn, spec, batch_size=args.batch_size)
|
||
import_summaries.append(summary)
|
||
print(
|
||
f"已汇聚 {spec.layer_key}: count={summary['count']} "
|
||
f"source={spec.database_name}.{spec.source_table}/{spec.source_layer}"
|
||
)
|
||
if int(summary["count"]) == 0:
|
||
print(f"警告: {spec.layer_key} 没有导入到任何记录，请检查源库或 source_layer")
|
||
|
||
manifest = build_manifest(conn, import_summaries)
|
||
args.manifest.parent.mkdir(parents=True, exist_ok=True)
|
||
args.manifest.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f"已生成融合库: {args.output_db}")
|
||
print(f"已写出清单: {args.manifest}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|