630 lines
25 KiB
Python
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()
|