From 2a6de41d12eeecf0a64810957c2e37620de48cc8 Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Sat, 3 Oct 2026 08:44:35 +0800 Subject: [PATCH] feat: add segmented global CM95 pipeline --- AGENTS.md | 6 + README.md | 6 + .../cm95_global_2014_quarantine_v1.json | 24 + config/global/cm95_global_2014_v1.json | 40 + config/pbf/cm95_tile_policy_global_v1.yaml | 102 ++ deploy/systemd/cm95-global.service | 19 + deploy/systemd/cm95-global.timer | 12 + docs/Global_CM95_Pipeline_v1.md | 64 + pyproject.toml | 1 + scripts/cm95_global_install_service.sh | 21 + src/cm93db/audit.py | 79 +- src/cm93db/cli.py | 9 +- src/cm93db/importer.py | 10 +- src/cm93db/scanner.py | 5 + src/cm95global/__init__.py | 2 + src/cm95global/cli.py | 59 + src/cm95global/config.py | 108 ++ src/cm95global/pipeline.py | 1411 +++++++++++++++++ src/cm95global/state.py | 157 ++ src/cm95pbf/production.py | 12 +- 20 files changed, 2120 insertions(+), 27 deletions(-) create mode 100644 config/global/cm95_global_2014_quarantine_v1.json create mode 100644 config/global/cm95_global_2014_v1.json create mode 100644 config/pbf/cm95_tile_policy_global_v1.yaml create mode 100644 deploy/systemd/cm95-global.service create mode 100644 deploy/systemd/cm95-global.timer create mode 100644 docs/Global_CM95_Pipeline_v1.md create mode 100755 scripts/cm95_global_install_service.sh create mode 100644 src/cm95global/__init__.py create mode 100644 src/cm95global/cli.py create mode 100644 src/cm95global/config.py create mode 100644 src/cm95global/pipeline.py create mode 100644 src/cm95global/state.py diff --git a/AGENTS.md b/AGENTS.md index eceac50..6a06aff 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -9,3 +9,9 @@ 7. 运行配置只使用 CM95_PG_* 与 CM95_SOURCE_DIR，不回退旧 MySQL 环境变量。 8. 全国全量导入前必须先通过单 Cell、小样本、DB audit 和单 tile PBF 验证。 9. 文档与日志不得记录数据库密码。 +10. Global CM95 使用独立 `cm95global` pipeline；输入 `/mnt/sda1/CM93_2014.zip`，数据库固定与日本基线分离为 `cm95_global_2014`。 +11. Global pipeline 稳定优先：每个 segment 独立进程、checkpoint/tag 后退出；systemd timer 启动下一段，不允许单 Python coordinator 连续驻留数天。 +12. X220 Global pipeline 只使用 CPU0/1、最多 2 PBF worker；低内存、高 load、低磁盘时必须 WAIT_RESOURCE，不领取新 segment。 +13. Global PBF 只使用 coverage/sparse 枚举，不允许 full-world aoi-complete z0-z12；日期变更线/异常超宽 coverage 必须 fail-closed。 +14. `/root/sourceserver/cm95` 与 NAS `cm95_chart` 日本基线继续保持只读参考，不得被 Global pipeline 修改。 +15. Global source quarantine 只能按精确 relative path + SHA256 + file size 绑定；必须在 shard 分配后过滤，并同时作用于 source scan / import / DB audit。禁止修改原始 ZIP、禁止为坏 cell 放宽全局 parser、禁止静默忽略 quarantine。 diff --git a/README.md b/README.md index 633a89d..9cf6b98 100644 --- a/README.md +++ b/README.md @@ -38,3 +38,9 @@ NAS 已预先建立数据库时，第 7 步不需要 --create-database。 导入完成后先建立 FID manifest，再创建 production plan 并生成 PBF。 mapping、tile policy、S-52、sprite/style 继续沿用参考工程已经验证的配置。 + +## Global CM95 2014 segmented pipeline + +全球输入使用 `/mnt/sda1/CM93_2014.zip`，独立数据库 `cm95_global_2014`。运行器 `cm95global` 每次只执行一个 segment 后退出，由 systemd timer 启动下一段；默认固定 CPU0/1、低优先级、资源健康闸门、64 个 source/import/audit shard，以及 coverage/sparse z0-z12 PBF ledger。 + +详见 `docs/Global_CM95_Pipeline_v1.md`。全球基础 PBF 与 bathymetry v5 保持两个独立长任务/产物。已确认损坏且无法可靠恢复的极少数 source cell 使用 path+SHA256+size 绑定的 quarantine 清单隔离；不修改原始 ZIP，也不放宽 CM93 parser。 diff --git a/config/global/cm95_global_2014_quarantine_v1.json b/config/global/cm95_global_2014_quarantine_v1.json new file mode 100644 index 0000000..ee27bcb --- /dev/null +++ b/config/global/cm95_global_2014_quarantine_v1.json @@ -0,0 +1,24 @@ +{ + "schema": "navsea.cm95.global-quarantine/v1", + "archive_sha256": "d78dc8aedbbce9686ae2fbef6fb84f860a9e4b324c9fc08201b9b40d48e48463", + "entries": [ + { + "source_relative_path": "03900060/E/03940087.E", + "sha256": "6079fb710505aec80adbcbca43491198170f7b4b7cbd30b0909a2cb878bbfbac", + "file_size": 13692, + "scale_code": "E", + "region": "Turkey / Black Sea near northern Bosporus", + "reason": "feature 0 record length overruns file; no unique trustworthy suffix reconstruction" + }, + { + "source_relative_path": "02700420/F/02870471.F", + "sha256": "6c4f83d8740074d0037a16ea364e9f11067e9f83c84e4413b726dc5545d2c88d", + "file_size": 8715, + "scale_code": "F", + "region": "FSM / Pohnpei State, Sapwuahfik (Ngatik) Atoll area", + "reason": "feature 41 INFORM string lacks NUL terminator and subsequent record boundary is not trustworthy", + "alternate_source_relative_path": "00300420/02700420/F/02870471.F", + "alternate_sha256": "72e1dbca14f20102a210ff0274e247ebd0423991e081460b4efd5e13d2f38a28" + } + ] +} diff --git a/config/global/cm95_global_2014_v1.json b/config/global/cm95_global_2014_v1.json new file mode 100644 index 0000000..3c839ff --- /dev/null +++ b/config/global/cm95_global_2014_v1.json @@ -0,0 +1,40 @@ +{ + "schema": "navsea.cm95.global-pipeline/v1", + "source_archive": "/mnt/sda1/CM93_2014.zip", + "work_root": "/mnt/sda1/cm95-global", + "source_dir_name": "CM93_2014", + "database": "cm95_global_2014", + "shard_count": 64, + "cpu_affinity": [ + 0, + 1 + ], + "min_free_gib": 30, + "min_mem_available_mib": 900, + "max_load1": 6.0, + "max_database_gib": 80, + "segment_retry_limit": 3, + "pbf_workers": 2, + "pbf_segment_jobs": 100, + "pbf_max_attempts": 3, + "pbf_zooms": [ + 0, + 1, + 2, + 3, + 4, + 5, + 6, + 7, + 8, + 9, + 10, + 11, + 12 + ], + "mapping": "config/mappings/cm95_to_navsea_v1.yaml", + "policy": "config/pbf/cm95_tile_policy_global_v1.yaml", + "s52_lookups": "resources/s52/lookup_rule_ir.json", + "s52_resource_contract": "resources/s52/resource_contract.json", + "quarantine": "config/global/cm95_global_2014_quarantine_v1.json" +} diff --git a/config/pbf/cm95_tile_policy_global_v1.yaml b/config/pbf/cm95_tile_policy_global_v1.yaml new file mode 100644 index 0000000..28a9634 --- /dev/null +++ b/config/pbf/cm95_tile_policy_global_v1.yaml @@ -0,0 +1,102 @@ +schema: navsea.cm95.tile-policy/v1 +version: v1.4-global-coverage-sparse +status: PRODUCTION_CANDIDATE +xyz: true +nominal_tile_pixels: 256 +buffer_tile_pixels: 8 +default_extent: 4096 +extent_by_zoom: + '10': 4096 + '11': 4096 + '12': 1048576 +enumeration_scales: + '0': + - 20000000 + '1': + - 20000000 + '2': + - 20000000 + '3': + - 20000000 + '4': + - 20000000 + '5': + - 3000000 + '6': + - 3000000 + '7': + - 3000000 + '8': + - 1000000 + '9': + - 1000000 + '10': + - 200000 + '11': + - 50000 + - 100000 + '12': + - 7500 + - 20000 + - 50000 +scale_ladders: + '0': + - 20000000 + - 3000000 + '1': + - 20000000 + - 3000000 + '2': + - 20000000 + - 3000000 + '3': + - 20000000 + - 3000000 + '4': + - 20000000 + - 3000000 + '5': + - 3000000 + - 20000000 + '6': + - 3000000 + - 20000000 + '7': + - 3000000 + - 20000000 + '8': + - 1000000 + - 3000000 + - 20000000 + '9': + - 1000000 + - 3000000 + - 20000000 + '10': + - 200000 + - 1000000 + - 3000000 + - 20000000 + '11': + - 50000 + - 100000 + - 200000 + - 1000000 + - 3000000 + - 20000000 + '12': + - 7500 + - 20000 + - 50000 + - 100000 + - 200000 + - 1000000 + - 3000000 + - 20000000 +notes: +- Global coverage-sparse: z0-z4 使用 1:20M 物化范围；scale ladder 仍允许 1:3M 回退。 +- Global production 使用 coverage/sparse z0-z12；不物化 full-world 空 tile，不新增 z13/z14。 +- z12 enumeration 加入 1:7,500（G），避免 G 级细节被生产策略直接忽略。 +- z12 scale ladder 优先 7,500，再回退 20k/50k/100k/...。 +- z0-z11 extent=4096；z12 extent=1048576，保持已验证 NavSea/MapLibre 契约。 +- enumeration_scales 仅决定物化范围；scale_ladders 仅决定 tile 内选择/回退。 diff --git a/deploy/systemd/cm95-global.service b/deploy/systemd/cm95-global.service new file mode 100644 index 0000000..f1b0aba --- /dev/null +++ b/deploy/systemd/cm95-global.service @@ -0,0 +1,19 @@ +[Unit] +Description=CM95 Global segmented pipeline - one segment +Wants=network-online.target +After=network-online.target +ConditionPathExists=/mnt/sda1/CM93_2014.zip + +[Service] +Type=oneshot +User=root +WorkingDirectory=/root/sourceserver/cm95-pg +ExecStart=/root/sourceserver/cm95-pg/.venv/bin/cm95global --config config/global/cm95_global_2014_v1.json run-next +CPUAffinity=0 1 +Nice=10 +IOSchedulingClass=best-effort +IOSchedulingPriority=7 +MemoryHigh=3G +MemoryMax=4G +TimeoutStartSec=4h +KillMode=control-group diff --git a/deploy/systemd/cm95-global.timer b/deploy/systemd/cm95-global.timer new file mode 100644 index 0000000..476a96b --- /dev/null +++ b/deploy/systemd/cm95-global.timer @@ -0,0 +1,12 @@ +[Unit] +Description=CM95 Global segmented pipeline scheduler + +[Timer] +OnBootSec=5min +OnUnitInactiveSec=2min +AccuracySec=15s +Persistent=true +Unit=cm95-global.service + +[Install] +WantedBy=timers.target diff --git a/docs/Global_CM95_Pipeline_v1.md b/docs/Global_CM95_Pipeline_v1.md new file mode 100644 index 0000000..3e1d8b8 --- /dev/null +++ b/docs/Global_CM95_Pipeline_v1.md @@ -0,0 +1,64 @@ +# Global CM95 Pipeline v1 + +## 目标 + +把 `/mnt/sda1/CM93_2014.zip` 的全球 CM93/CM95 数据稳定导入独立 PostGIS 数据库，并生成 coverage/sparse 的 z0-z12 基础 PBF。 + +稳定跑完优先于速度。流水线禁止依赖一个长寿命 Python 进程连续运行数天。 + +## 固定资源边界 + +- X220：2 物理核 / 4 逻辑 CPU。 +- systemd service 固定 `CPUAffinity=0 1`，完整保留 CPU2/3 所在物理核给其他工作。 +- `Nice=10`，I/O priority=7。 +- 每个 segment 为独立 oneshot 进程，默认 4 小时 hard timeout。 +- MemoryHigh=3 GiB，MemoryMax=4 GiB。 +- PBF 最多 2 worker；单次最多 100 tile jobs。 +- `/mnt/sda1` 空闲低于 30 GiB 时不领取新 segment。 +- MemAvailable 低于 900 MiB 或 load1 高于 6 时进入 WAIT_RESOURCE，不破坏 checkpoint。 + +## 阶段 + +1. `PRECHECK`：ZIP CRC、SHA256、cell 数、代码 fingerprint。 +2. `EXTRACT`：原子解压到 `/mnt/sda1/cm95-global/source/CM93_2014`。 +3. `INVENTORY`：冻结 source manifest 与 64-shard 分配。 +4. `QUARANTINE_VALIDATE`：验证已知坏源的 relative path / SHA256 / size；不改写原始 source。 +5. `SOURCE_SCAN_000..063`：每 shard 独立 source-only parse audit；shard 分配完成后再过滤 quarantine，避免重排。 +6. `DB_INIT`：创建独立 `cm95_global_2014` + PostGIS schema。 +7. `POSTGIS_IMPORT_000..063`：每 shard 一个全新 importer 进程，并使用同一 quarantine。 +8. `DB_AUDIT_000..063`：每 shard 独立 source↔DB audit，并使用同一 quarantine。 +9. `FID_MANIFEST`：检查 int32 FID 容量后生成全球 deterministic manifest。 +10. `COVERAGE_PLAN_Z00..Z12`：每 zoom 独立生成 sparse tile ledger。 +11. `BASE_PBF_SEGMENT_xxxxxx`：每次领取一小批 tile；结束后释放全部内存/连接。 +12. `FINAL_AUDIT`：数据库 fingerprint、PBF ledger、实际文件数与冻结资源 hash。 +13. `PIPELINE_DONE`：基础 PostGIS + sparse PBF 完成。 + +Bathymetry v5 为第二套独立 PBF，不与基础 PBF 放在同一长任务故障域中。 + +## Source quarantine + +Global source 保持原始字节不变；parser 继续严格 fail-closed。若极少数 CM93 cell 经二进制检查确认无法可靠恢复，可进入 `config/global/cm95_global_2014_quarantine_v1.json`。每条 quarantine 必须绑定 source relative path、SHA256、file size 与原因；运行时任一项不匹配立即停止，禁止按文件名模糊跳过。 + +Quarantine 在既有 shard 取模分配完成后过滤，因此不会改变其他 cell 的 shard 归属。SOURCE_SCAN、POSTGIS_IMPORT、DB_AUDIT 三阶段必须消费同一清单；最终 release/report 必须显式记录 quarantine 数量与字节数，不得把隔离源伪装成成功解析。 + +## Checkpoint / Tag + +运行状态位于 `/mnt/sda1/cm95-global/state/`。每段 PASS 写 `tags/.PASS.json`；失败尝试写 `tags/.attemptN.FAIL.json`。 + +Tag 固定记录开始/结束机器状态、elapsed、peak RSS、磁盘、swap、DB 大小和本段业务统计。 + +同一 segment 最多自动尝试 3 次。source/import/audit shard 连续失败 3 次后继续其他 shard，到阶段边界 fail-closed；PBF 单 tile 最多尝试 3 次。 + +## 代码冻结 + +`PRECHECK` 会对 parser、PostGIS/PBF/global pipeline、config、S-52 resource 与 schema 计算代码 fingerprint。后续任一 segment 发现 fingerprint 改变，自动进入 `CODE_VERSION_CHANGED / PAUSED`，禁止一套 release 静默混用两个算法版本。 + +## 调度 + +安装并启动：`scripts/cm95_global_install_service.sh --enable`。 + +查看：`.venv/bin/cm95global status`、`systemctl status cm95-global.timer --no-pager`、`journalctl -u cm95-global.service`。 + +暂停/恢复：`.venv/bin/cm95global pause` / `.venv/bin/cm95global resume`。 + +timer 只触发 `run-next`。每次只执行一个 segment，然后整个 Python 进程退出；下一次 timer 再启动全新进程。 diff --git a/pyproject.toml b/pyproject.toml index 8764852..106919d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,7 @@ dependencies = [ [project.scripts] cm95pg = "cm93db.cli:main" cm95pbf = "cm95pbf.cli:main" +cm95global = "cm95global.cli:main" [tool.setuptools] package-dir = {"" = "src"} diff --git a/scripts/cm95_global_install_service.sh b/scripts/cm95_global_install_service.sh new file mode 100755 index 0000000..c1a5c34 --- /dev/null +++ b/scripts/cm95_global_install_service.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +set -euo pipefail + +PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$PROJECT_ROOT" + +install -m 0644 deploy/systemd/cm95-global.service /etc/systemd/system/cm95-global.service +install -m 0644 deploy/systemd/cm95-global.timer /etc/systemd/system/cm95-global.timer +mkdir -p /mnt/sda1/cm95-global/{state/tags,state/locks,logs,reports,manifests,pbf} + +systemctl daemon-reload + +if [[ "${1:-}" == "--enable" ]]; then + systemctl enable --now cm95-global.timer +fi + +echo "Installed: /etc/systemd/system/cm95-global.service" +echo "Installed: /etc/systemd/system/cm95-global.timer" +echo "Enable: systemctl enable --now cm95-global.timer" +echo "Status: systemctl status cm95-global.timer --no-pager" +echo "Pipeline: .venv/bin/cm95global status" diff --git a/src/cm93db/audit.py b/src/cm93db/audit.py index 1cffee3..12700ff 100644 --- a/src/cm93db/audit.py +++ b/src/cm93db/audit.py @@ -18,7 +18,7 @@ from .database import ( ) from .dictionaries import load_object_dictionary, load_attribute_dictionary from .geometry import build_coverages, build_feature -from .scanner import scan_cm93, CellFile +from .scanner import scan_cm93, CellFile, normalize_relative_path @dataclass(frozen=True) @@ -52,6 +52,13 @@ class PrecisionSample: class AuditResult: source_cells: int = 0 source_parse_failed: int = 0 + source_features: int = 0 + source_attributes: int = 0 + source_points: int = 0 + source_lines: int = 0 + source_areas: int = 0 + source_coverages: int = 0 + source_sounding_points: int = 0 db_missing_cells: int = 0 db_not_done_cells: int = 0 count_mismatches: int = 0 @@ -74,6 +81,13 @@ class AuditResult: "passed": self.passed, "source_cells": self.source_cells, "source_parse_failed": self.source_parse_failed, + "source_features": self.source_features, + "source_attributes": self.source_attributes, + "source_points": self.source_points, + "source_lines": self.source_lines, + "source_areas": self.source_areas, + "source_coverages": self.source_coverages, + "source_sounding_points": self.source_sounding_points, "db_missing_cells": self.db_missing_cells, "db_not_done_cells": self.db_not_done_cells, "count_mismatches": self.count_mismatches, @@ -93,13 +107,23 @@ def _rel(cf: CellFile) -> str: return str(cf.relative_path).replace("/", "\\") -def _select_cells(scan, limit: int | None, scales: set[str] | None, cell_name: str | None): +def _select_cells(scan, limit: int | None, scales: set[str] | None, cell_name: str | None, shard_count: int | None = None, shard_index: int | None = None, exclude_relative_paths: set[str] | None = None): cells=list(scan.cells) if scales: cells=[c for c in cells if c.scale_code in scales] if cell_name: t=cell_name.lower() cells=[c for c in cells if c.path.name.lower()==t or c.path.stem.lower()==t] + if shard_count is not None: + if shard_count < 1: + raise ValueError("shard_count must be >= 1") + index = 0 if shard_index is None else shard_index + if index < 0 or index >= shard_count: + raise ValueError("shard_index must satisfy 0 <= index < shard_count") + cells=[c for i,c in enumerate(cells) if i % shard_count == index] + if exclude_relative_paths: + excluded={normalize_relative_path(x) for x in exclude_relative_paths} + cells=[c for c in cells if normalize_relative_path(c.relative_path) not in excluded] if limit is not None: cells=cells[:limit] return cells @@ -126,13 +150,15 @@ def _source_stat(cell, cf: CellFile) -> SourceCellStat: def collect_source(root: Path, limit: int | None = None, scales: set[str] | None = None, cell_name: str | None = None, - precision_samples: int = 0, validate_geometry: bool = False): + precision_samples: int = 0, validate_geometry: bool = False, + shard_count: int | None = None, shard_index: int | None = None, + exclude_relative_paths: set[str] | None = None): scan=scan_cm93(root) if not scan.object_dictionary or not scan.attribute_dictionary: raise RuntimeError("CM93 字典未找到") objs={x.object_id:x for x in load_object_dictionary(scan.object_dictionary)} attrs={x.attribute_id:x for x in load_attribute_dictionary(scan.attribute_dictionary)} - cells=_select_cells(scan,limit,scales,cell_name) + cells=_select_cells(scan,limit,scales,cell_name,shard_count,shard_index,exclude_relative_paths) stats: dict[str,SourceCellStat]={} failures: list[str]=[] samples: list[PrecisionSample]=[] @@ -193,17 +219,26 @@ def _precision_db_geom(conn, cell_id: int, sample: PrecisionSample): def run_audit(cfg: DbConfig, root: Path, limit: int | None = None, scales: set[str] | None = None, cell_name: str | None = None, - precision_samples: int = 128, tolerance_deg: float = 1e-12, validate_geometry: bool = False, source_only: bool = False, require_raw: bool = True) -> AuditResult: + precision_samples: int = 128, tolerance_deg: float = 1e-12, validate_geometry: bool = False, source_only: bool = False, require_raw: bool = True, + shard_count: int | None = None, shard_index: int | None = None, exclude_relative_paths: set[str] | None = None) -> AuditResult: result=AuditResult() - stats,failures,samples,geom_issues=collect_source(root,limit,scales,cell_name,precision_samples if not source_only else 0,validate_geometry) + stats,failures,samples,geom_issues=collect_source(root,limit,scales,cell_name,precision_samples if not source_only else 0,validate_geometry,shard_count,shard_index,exclude_relative_paths) result.source_cells=len(stats)+len(failures) result.source_parse_failed=len(failures) result.messages.extend(f"SOURCE_PARSE_FAILED {m}" for m in failures[:100]) + result.source_features=sum(s.features for s in stats.values()) + result.source_attributes=sum(s.attributes for s in stats.values()) + result.source_points=sum(s.points for s in stats.values()) + result.source_lines=sum(s.lines for s in stats.values()) + result.source_areas=sum(s.areas for s in stats.values()) + result.source_coverages=sum(s.coverages for s in stats.values()) + result.source_sounding_points=sum(s.sounding_points for s in stats.values()) result.unknown_objects=sum(s.unknown_objects for s in stats.values()) result.unknown_attributes=sum(s.unknown_attributes for s in stats.values()) print(f"Source cells parsed: {len(stats)}, failed: {len(failures)}") - print(f"Source features: {sum(s.features for s in stats.values())}") - print(f"Source attributes: {sum(s.attributes for s in stats.values())}") + print(f"Source features: {result.source_features}") + print(f"Source attributes: {result.source_attributes}") + print(f"Source sounding points: {result.source_sounding_points}") result.geometry_validity_issues=geom_issues if validate_geometry: print(f"Source geometry issues (Shapely validity/structure): {geom_issues}") @@ -218,19 +253,26 @@ def run_audit(cfg: DbConfig, root: Path, limit: int | None = None, scales: set[s result.db_missing_cells=len(stats) result.messages.append("数据库中找不到该 source_root 对应 dataset") return result + rels=list(stats) with conn.cursor() as cur: - cur.execute("SELECT id,source_relative_path,import_status,source_sha256,feature_count,attribute_count FROM chart_cell WHERE dataset_id=%s", (did,)) + cur.execute("SELECT id,source_relative_path,import_status,source_sha256,feature_count,attribute_count FROM chart_cell WHERE dataset_id=%s AND source_relative_path = ANY(%s)", (did,rels)) rows=cur.fetchall() db_by_rel={str(r["source_relative_path"]):r for r in rows} cellids=[int(r["id"]) for r in rows] - fcnt=_group_counts(conn,f"SELECT cell_id,COUNT(*) n FROM chart_feature WHERE cell_id IN (SELECT id FROM chart_cell WHERE dataset_id={did}) GROUP BY cell_id") - acnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_feature_attribute a JOIN chart_feature f ON f.id=a.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id") - pcnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_point_geometry g JOIN chart_feature f ON f.id=g.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id") - lcnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_line_geometry g JOIN chart_feature f ON f.id=g.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id") - arcnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_area_geometry g JOIN chart_feature f ON f.id=g.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id") - ccnt=_group_counts(conn,f"SELECT cell_id,COUNT(*) n FROM chart_coverage WHERE cell_id IN (SELECT id FROM chart_cell WHERE dataset_id={did}) GROUP BY cell_id") - scnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_sounding_point s JOIN chart_feature f ON f.id=s.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id") - rawbytes=_group_counts(conn,f"SELECT cell_id,SUM(OCTET_LENGTH(raw_blob)) n FROM chart_raw_record WHERE cell_id IN (SELECT id FROM chart_cell WHERE dataset_id={did}) GROUP BY cell_id") + def grouped(sql: str) -> dict[int,int]: + if not cellids: + return {} + with conn.cursor() as cur: + cur.execute(sql,(cellids,)) + return {int(r["cell_id"]):int(r["n"]) for r in cur.fetchall()} + fcnt=grouped("SELECT cell_id,COUNT(*) n FROM chart_feature WHERE cell_id = ANY(%s) GROUP BY cell_id") + acnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_feature_attribute a JOIN chart_feature f ON f.id=a.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id") + pcnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_point_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id") + lcnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_line_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id") + arcnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_area_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id") + ccnt=grouped("SELECT cell_id,COUNT(*) n FROM chart_coverage WHERE cell_id = ANY(%s) GROUP BY cell_id") + scnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_sounding_point s JOIN chart_feature f ON f.id=s.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id") + rawbytes=grouped("SELECT cell_id,SUM(OCTET_LENGTH(raw_blob)) n FROM chart_raw_record WHERE cell_id = ANY(%s) GROUP BY cell_id") for rel,s in stats.items(): r=db_by_rel.get(rel) @@ -282,6 +324,9 @@ def print_result(r: AuditResult) -> None: print("\n=== CM93 AUDIT ===") print(f"source_cells : {r.source_cells}") print(f"source_parse_failed : {r.source_parse_failed}") + print(f"source_features : {r.source_features}") + print(f"source_attributes : {r.source_attributes}") + print(f"source_sounding_points: {r.source_sounding_points}") print(f"db_missing_cells : {r.db_missing_cells}") print(f"db_not_done_cells : {r.db_not_done_cells}") print(f"count_mismatches : {r.count_mismatches}") diff --git a/src/cm93db/cli.py b/src/cm93db/cli.py index ac87c50..745f5fb 100644 --- a/src/cm93db/cli.py +++ b/src/cm93db/cli.py @@ -129,7 +129,7 @@ def cmd_import(args): opt=ImportOptions(root=root,limit=args.limit,scales=_scales(args.scale),cell_name=args.cell, reimport=args.reimport,retry_failed=not args.skip_failed,fail_fast=args.fail_fast, store_raw_tables=not args.no_raw_tables,shard_count=args.shard_count, - shard_index=args.shard_index) + shard_index=args.shard_index,exclude_relative_paths=set(args.exclude_relative_path or [])) s=run_import(cfg,opt) print("\n=== IMPORT SUMMARY ===") for k in ("cells_discovered","cells_done","cells_skipped","cells_failed","features_source","features_imported","attributes_source","attributes_imported","unknown_object_count","unknown_attribute_count","geometry_issue_count","sounding_points"): @@ -142,7 +142,8 @@ def cmd_audit(args): cfg=load_config() if not args.source_only else load_config() r=run_audit(cfg,_root(args.root),limit=args.limit,scales=_scales(args.scale),cell_name=args.cell, precision_samples=args.precision_samples,tolerance_deg=args.tolerance, - validate_geometry=args.validate_geometry,source_only=args.source_only,require_raw=not args.allow_missing_raw) + validate_geometry=args.validate_geometry,source_only=args.source_only,require_raw=not args.allow_missing_raw, + shard_count=args.shard_count,shard_index=args.shard_index,exclude_relative_paths=set(args.exclude_relative_path or [])) print_result(r) report = Path(args.report) if args.report else (project_root() / "logs" / f"cm93_audit_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json") report.parent.mkdir(parents=True,exist_ok=True) @@ -179,6 +180,7 @@ def build_parser(): s.add_argument("--no-raw-tables", action="store_true", help="不保存 decoded 原始 table；不建议正式母库使用") s.add_argument("--shard-count", type=int, help="可选并行分片总数；默认不分片") s.add_argument("--shard-index", type=int, help="当前分片编号，从 0 开始；需与 --shard-count 配合") + s.add_argument("--exclude-relative-path", action="append", default=[], help="按 source relative path 显式排除 cell；可重复，仅用于已审计 quarantine") s.set_defaults(func=cmd_import) s = sub.add_parser("audit", help="重新解析源 CM93 并与数据库逐 cell 审计") @@ -191,6 +193,9 @@ def build_parser(): s.add_argument("--validate-geometry", action="store_true", help="额外执行 Shapely validity 全量检查，较慢") s.add_argument("--source-only", action="store_true", help="只审计源文件解析，不查询数据库") s.add_argument("--allow-missing-raw", action="store_true", help="不把 raw table 未保存视为审计失败") + s.add_argument("--shard-count", type=int, help="可选稳定分片总数；与 import 使用相同排序/取模规则") + s.add_argument("--shard-index", type=int, help="当前分片编号，从 0 开始；需与 --shard-count 配合") + s.add_argument("--exclude-relative-path", action="append", default=[], help="按 source relative path 显式排除 cell；可重复，仅用于已审计 quarantine") s.add_argument("--report", help="JSON 审计报告路径；默认写入项目 logs/") s.set_defaults(func=cmd_audit) return p diff --git a/src/cm93db/importer.py b/src/cm93db/importer.py index c081e1c..e892474 100644 --- a/src/cm93db/importer.py +++ b/src/cm93db/importer.py @@ -21,7 +21,7 @@ from .database import ( from psycopg.types.json import Jsonb from .dictionaries import load_object_dictionary, load_attribute_dictionary, ObjectDef, AttributeDef from .geometry import build_coverages, build_feature, BuiltFeature -from .scanner import scan_cm93, CellFile +from .scanner import scan_cm93, CellFile, normalize_relative_path RAW_CHUNK = 512 * 1024 @@ -50,6 +50,7 @@ class ImportOptions: store_raw_tables: bool = True shard_count: int | None = None shard_index: int | None = None + exclude_relative_paths: set[str] | None = None @dataclass @@ -861,6 +862,13 @@ def _selected_cells( raise ValueError("shard_index must satisfy 0 <= index < shard_count") cells = [c for i, c in enumerate(cells) if i % opt.shard_count == shard_index] + if opt.exclude_relative_paths: + excluded = {normalize_relative_path(x) for x in opt.exclude_relative_paths} + cells = [ + c for c in cells + if normalize_relative_path(c.relative_path) not in excluded + ] + if opt.limit is not None: cells = cells[:opt.limit] diff --git a/src/cm93db/scanner.py b/src/cm93db/scanner.py index 0a5140e..7e93928 100644 --- a/src/cm93db/scanner.py +++ b/src/cm93db/scanner.py @@ -28,6 +28,11 @@ class ScanResult: object_dictionary: Path | None attribute_dictionary: Path | None +def normalize_relative_path(value: str | Path) -> str: + """Canonical source-relative path used only for selection/exclusion matching.""" + return str(value).replace("\\", "/").lstrip("./").casefold() + + def _find_case_insensitive(root: Path, names: set[str]) -> Path | None: lowers = {n.lower() for n in names} for p in root.rglob("*"): diff --git a/src/cm95global/__init__.py b/src/cm95global/__init__.py new file mode 100644 index 0000000..01f83e3 --- /dev/null +++ b/src/cm95global/__init__.py @@ -0,0 +1,2 @@ +__version__ = "1.0.0" +PIPELINE_CONTRACT_VERSION = "cm95-global-pipeline/1.0.0" diff --git a/src/cm95global/cli.py b/src/cm95global/cli.py new file mode 100644 index 0000000..451a44f --- /dev/null +++ b/src/cm95global/cli.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from .config import load_global_config +from .pipeline import GlobalPipeline +from .state import PipelineLock + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="cm95global", + description="Segmented, restart-safe Global CM95 -> PostGIS -> sparse PBF pipeline", + ) + p.add_argument( + "--config", + default="config/global/cm95_global_2014_v1.json", + help="global pipeline JSON config", + ) + sub = p.add_subparsers(dest="command", required=True) + sub.add_parser("run-next", help="run exactly one segment, write checkpoint, then exit") + sub.add_parser("status", help="show current persistent pipeline status") + sub.add_parser("pause", help="pause future timer-driven segments") + r = sub.add_parser("resume", help="resume future timer-driven segments") + r.add_argument( + "--retry-exhausted", + action="store_true", + help="clear exhausted segment retry markers", + ) + return p + + +def main() -> None: + args = build_parser().parse_args() + cfg = load_global_config(Path(args.config)) + pipe = GlobalPipeline(cfg) + + if args.command == "status": + print(json.dumps(pipe.status(), ensure_ascii=False, indent=2)) + return + if args.command == "pause": + pipe.pause("manual pause") + print(json.dumps(pipe.status(), ensure_ascii=False, indent=2)) + return + if args.command == "resume": + pipe.resume(retry_exhausted=bool(args.retry_exhausted)) + print(json.dumps(pipe.status(), ensure_ascii=False, indent=2)) + return + + lock = cfg.state_root / "locks" / "pipeline.lock" + with PipelineLock(lock): + result = pipe.run_next() + print(json.dumps(result, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/src/cm95global/config.py b/src/cm95global/config.py new file mode 100644 index 0000000..ee98823 --- /dev/null +++ b/src/cm95global/config.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +from dataclasses import dataclass +import json +from pathlib import Path + + +@dataclass(frozen=True) +class GlobalConfig: + project_root: Path + config_path: Path + source_archive: Path + work_root: Path + source_dir_name: str + database: str + shard_count: int + cpu_affinity: tuple[int, ...] + min_free_gib: float + min_mem_available_mib: int + max_load1: float + max_database_gib: float + segment_retry_limit: int + pbf_workers: int + pbf_segment_jobs: int + pbf_max_attempts: int + pbf_zooms: tuple[int, ...] + mapping: Path + policy: Path + s52_lookups: Path + s52_resource_contract: Path + quarantine: Path + + @property + def source_root(self) -> Path: + return self.work_root / "source" / self.source_dir_name + + @property + def state_root(self) -> Path: + return self.work_root / "state" + + @property + def tag_root(self) -> Path: + return self.state_root / "tags" + + @property + def log_root(self) -> Path: + return self.work_root / "logs" + + @property + def report_root(self) -> Path: + return self.work_root / "reports" + + @property + def manifest_root(self) -> Path: + return self.work_root / "manifests" + + @property + def fid_manifest(self) -> Path: + return self.manifest_root / "cm95_global_fid_manifest_v1.sqlite" + + @property + def pbf_root(self) -> Path: + return self.work_root / "pbf" + + @property + def pbf_ledger(self) -> Path: + return self.state_root / "global_pbf_ledger_v1.sqlite" + + @property + def source_manifest(self) -> Path: + return self.state_root / "source_manifest_v1.sqlite" + + +def _resolve(project_root: Path, raw: str) -> Path: + p = Path(raw) + return p if p.is_absolute() else project_root / p + + +def load_global_config(path: Path) -> GlobalConfig: + path = Path(path).resolve() + project_root = path.parents[2] + obj = json.loads(path.read_text(encoding="utf-8")) + if obj.get("schema") != "navsea.cm95.global-pipeline/v1": + raise RuntimeError(f"unsupported global config schema: {obj.get('schema')}") + return GlobalConfig( + project_root=project_root, + config_path=path, + source_archive=Path(obj["source_archive"]), + work_root=Path(obj["work_root"]), + source_dir_name=str(obj["source_dir_name"]), + database=str(obj["database"]), + shard_count=int(obj["shard_count"]), + cpu_affinity=tuple(int(x) for x in obj["cpu_affinity"]), + min_free_gib=float(obj["min_free_gib"]), + min_mem_available_mib=int(obj["min_mem_available_mib"]), + max_load1=float(obj["max_load1"]), + max_database_gib=float(obj["max_database_gib"]), + segment_retry_limit=int(obj["segment_retry_limit"]), + pbf_workers=int(obj["pbf_workers"]), + pbf_segment_jobs=int(obj["pbf_segment_jobs"]), + pbf_max_attempts=int(obj["pbf_max_attempts"]), + pbf_zooms=tuple(int(x) for x in obj["pbf_zooms"]), + mapping=_resolve(project_root, obj["mapping"]), + policy=_resolve(project_root, obj["policy"]), + s52_lookups=_resolve(project_root, obj["s52_lookups"]), + s52_resource_contract=_resolve(project_root, obj["s52_resource_contract"]), + quarantine=_resolve(project_root, obj["quarantine"]), + ) diff --git a/src/cm95global/pipeline.py b/src/cm95global/pipeline.py new file mode 100644 index 0000000..685a2dd --- /dev/null +++ b/src/cm95global/pipeline.py @@ -0,0 +1,1411 @@ +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import replace +from datetime import datetime, timezone +import hashlib +import json +import math +import os +from pathlib import Path +import shutil +import sqlite3 +import subprocess +import threading +import time +import traceback +from typing import Any +from zipfile import ZipFile + +from cm93db.config import load_config +from cm93db.database import connect, create_database, apply_schema +from cm93db.scanner import scan_cm93, normalize_relative_path +from cm95pbf import BUILDER_CONTRACT_VERSION +from cm95pbf.builder import Cm95TileBuilder, TileJob +from cm95pbf.identity import dataset_fingerprint, validate_manifest, ROLE_FACTOR, MAX_SIGNED_INT32 +from cm95pbf.production import ( + _enumeration_scales, + _fetch_plan_geometries, + _load_policy, + _tiles_intersecting_geom, +) + +from .config import GlobalConfig +from .state import atomic_json, code_fingerprint, default_state, load_json, machine_snapshot, peak_rss_mib, sha256_file, utc_now, write_tag + + +GLOBAL_LEDGER_SCHEMA = "navsea.cm95.global-pbf-ledger/v1" +GLOBAL_RELEASE_SCHEMA = "navsea.cm95.global-release/v1" + + +class GlobalPbfLedger: + def __init__(self, path: Path): + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + self._init() + + def _conn(self): + db = sqlite3.connect(self.path, timeout=60) + db.row_factory = sqlite3.Row + db.execute("PRAGMA journal_mode=WAL") + db.execute("PRAGMA synchronous=NORMAL") + return db + + def _init(self): + with self._conn() as db: + db.executescript( + """ + CREATE TABLE IF NOT EXISTS meta ( + k TEXT PRIMARY KEY, + v TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS zoom_plan ( + z INTEGER PRIMARY KEY, + state TEXT NOT NULL, + tile_count INTEGER NOT NULL, + coverage_pieces INTEGER NOT NULL, + planned_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS tile_job ( + profile TEXT NOT NULL, + z INTEGER NOT NULL, + x INTEGER NOT NULL, + y INTEGER NOT NULL, + state TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + output_feature_count INTEGER, + output_layer_count INTEGER, + written INTEGER, + last_error TEXT, + PRIMARY KEY(profile,z,x,y) + ); + CREATE INDEX IF NOT EXISTS idx_global_tile_state + ON tile_job(state,attempts,z,x,y); + """ + ) + + + def set_meta(self, key: str, value: str) -> None: + with self._conn() as db: + db.execute( + "INSERT INTO meta(k,v) VALUES(?,?) " + "ON CONFLICT(k) DO UPDATE SET v=excluded.v", + (key, str(value)), + ) + + def get_meta(self, key: str) -> str | None: + with self._conn() as db: + row = db.execute("SELECT v FROM meta WHERE k=?", (key,)).fetchone() + return None if row is None else str(row["v"]) + + def zoom_done(self, z: int) -> bool: + with self._conn() as db: + row = db.execute( + "SELECT state FROM zoom_plan WHERE z=?", (int(z),) + ).fetchone() + return bool(row and row["state"] == "PASS") + + def insert_tiles(self, z: int, tiles: set[tuple[int, int, int]]) -> None: + if not tiles: + return + now = utc_now() + rows = (("delivery", int(a), int(b), int(c), now) for a, b, c in tiles) + with self._conn() as db: + db.executemany( + "INSERT OR IGNORE INTO tile_job" + "(profile,z,x,y,state,attempts,updated_at) " + "VALUES(?,?,?,?, 'PENDING',0,?)", + rows, + ) + + def finish_zoom(self, z: int, coverage_pieces: int) -> int: + with self._conn() as db: + n = int( + db.execute( + "SELECT COUNT(*) n FROM tile_job WHERE z=?", (int(z),) + ).fetchone()["n"] + ) + db.execute( + "INSERT INTO zoom_plan(z,state,tile_count,coverage_pieces,planned_at) " + "VALUES(?,?,?,?,?) " + "ON CONFLICT(z) DO UPDATE SET state=excluded.state," + "tile_count=excluded.tile_count,coverage_pieces=excluded.coverage_pieces," + "planned_at=excluded.planned_at", + (int(z), "PASS", n, int(coverage_pieces), utc_now()), + ) + return n + + + def requeue_running(self) -> int: + with self._conn() as db: + cur = db.execute( + "UPDATE tile_job SET state='PENDING',updated_at=? " + "WHERE state='RUNNING'", + (utc_now(),), + ) + return int(cur.rowcount) + + def runnable(self, limit: int, max_attempts: int) -> list[sqlite3.Row]: + with self._conn() as db: + return list( + db.execute( + "SELECT * FROM tile_job WHERE state!='PASS' AND attempts None: + with self._lock, self._conn() as db: + db.execute( + "UPDATE tile_job SET state='RUNNING',attempts=attempts+1," + "updated_at=?,last_error=NULL WHERE profile=? AND z=? AND x=? AND y=?", + (utc_now(), row["profile"], row["z"], row["x"], row["y"]), + ) + + def mark_pass(self, row: sqlite3.Row, result: dict[str, Any]) -> None: + with self._lock, self._conn() as db: + db.execute( + "UPDATE tile_job SET state='PASS',updated_at=?," + "output_feature_count=?,output_layer_count=?,written=?,last_error=NULL " + "WHERE profile=? AND z=? AND x=? AND y=?", + ( + utc_now(), + int(result.get("output_feature_count") or 0), + int(result.get("output_layer_count") or 0), + 1 if result.get("written") else 0, + row["profile"], row["z"], row["x"], row["y"], + ), + ) + + def mark_failed(self, row: sqlite3.Row, error: str) -> None: + with self._lock, self._conn() as db: + db.execute( + "UPDATE tile_job SET state='FAILED',updated_at=?,last_error=? " + "WHERE profile=? AND z=? AND x=? AND y=?", + (utc_now(), error[-12000:], row["profile"], row["z"], row["x"], row["y"]), + ) + + + def summary(self, max_attempts: int) -> dict[str, Any]: + with self._conn() as db: + states = { + str(r["state"]): int(r["n"]) + for r in db.execute( + "SELECT state,COUNT(*) n FROM tile_job GROUP BY state" + ).fetchall() + } + row = db.execute( + "SELECT COUNT(*) jobs,COALESCE(SUM(attempts),0) attempts," + "COALESCE(SUM(output_feature_count),0) output_features," + "COALESCE(SUM(CASE WHEN written=1 THEN 1 ELSE 0 END),0) pbf_files " + "FROM tile_job" + ).fetchone() + exhausted = int( + db.execute( + "SELECT COUNT(*) n FROM tile_job " + "WHERE state='FAILED' AND attempts>=?", + (int(max_attempts),), + ).fetchone()["n"] + ) + return { + "jobs": int(row["jobs"]), + "attempts": int(row["attempts"]), + "output_features": int(row["output_features"]), + "pbf_files": int(row["pbf_files"]), + "states": states, + "exhausted_failed": exhausted, + } + + +def _db_config(global_cfg: GlobalConfig): + return replace(load_config(), database=global_cfg.database) + + +def _subprocess_env(global_cfg: GlobalConfig) -> dict[str, str]: + env = os.environ.copy() + env["CM95_PG_DATABASE"] = global_cfg.database + env["CM95_SOURCE_DIR"] = str(global_cfg.source_root) + env["PYTHONUNBUFFERED"] = "1" + return env + + +def _dataset_row(cfg, source_root: Path) -> dict[str, Any] | None: + try: + with connect(cfg) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT id,parser_version,created_at,updated_at " + "FROM chart_dataset WHERE source_type='CM93' AND source_root=%s " + "ORDER BY id LIMIT 1", + (str(source_root.resolve()),), + ) + row = cur.fetchone() + conn.rollback() + return None if row is None else dict(row) + except Exception: + return None + + +def _database_size_bytes(cfg) -> int | None: + try: + with connect(cfg) as conn: + with conn.cursor() as cur: + cur.execute("SELECT pg_database_size(current_database()) AS n") + row = cur.fetchone() + conn.rollback() + return int(row["n"]) + except Exception: + return None + + +def _sha256_text(obj: Any) -> str: + raw = json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def _release_contract(cfg: GlobalConfig, dataset_id: int, fp: str) -> tuple[str, str]: + immutable = { + "schema": GLOBAL_LEDGER_SCHEMA, + "dataset_id": int(dataset_id), + "dataset_fingerprint": fp, + "builder_contract_version": BUILDER_CONTRACT_VERSION, + "mapping_sha256": sha256_file(cfg.mapping), + "policy_sha256": sha256_file(cfg.policy), + "s52_lookups_sha256": sha256_file(cfg.s52_lookups), + "s52_resource_contract_sha256": sha256_file(cfg.s52_resource_contract), + "zooms": list(cfg.pbf_zooms), + "tile_enumeration_mode": "coverage", + } + contract = _sha256_text(immutable) + release_id = f"cm95-global-2014-d{dataset_id}-{fp[:12]}-{contract[:12]}" + return release_id, contract + + +class GlobalPipeline: + def __init__(self, cfg: GlobalConfig): + self.cfg = cfg + for p in ( + cfg.work_root, + cfg.state_root, + cfg.tag_root, + cfg.log_root, + cfg.report_root, + cfg.manifest_root, + cfg.pbf_root, + ): + p.mkdir(parents=True, exist_ok=True) + self.state_path = cfg.state_root / "pipeline.json" + self.status_path = cfg.state_root / "status.json" + self.state = load_json(self.state_path, default_state()) + if self.state.get("schema") != "navsea.cm95.global-state/v1": + raise RuntimeError("unsupported global pipeline state schema") + self.pg = _db_config(cfg) + + def save_state(self) -> None: + self.state["updated_at"] = utc_now() + atomic_json(self.state_path, self.state) + + def set_status(self, status: str, **extra: Any) -> None: + atomic_json( + self.status_path, + { + "schema": "navsea.cm95.global-status/v1", + "status": status, + "updated_at": utc_now(), + "paused": bool(self.state.get("paused")), + "pause_reason": self.state.get("pause_reason"), + "last_action": self.state.get("last_action"), + "last_result": self.state.get("last_result"), + **extra, + }, + ) + + def tag_passed(self, key: str) -> bool: + return (self.cfg.tag_root / f"{key}.PASS.json").exists() + + def exhausted(self, key: str) -> bool: + return key in set(self.state.get("exhausted") or []) + + def pause(self, reason: str) -> None: + self.state["paused"] = True + self.state["pause_reason"] = str(reason) + self.save_state() + self.set_status("PAUSED", reason=str(reason)) + + + def resume(self, retry_exhausted: bool = False) -> None: + self.state["paused"] = False + self.state["pause_reason"] = None + if retry_exhausted: + self.state["exhausted"] = [] + self.state["attempts"] = {} + self.save_state() + self.set_status("READY") + + def health_gate(self) -> tuple[bool, dict[str, Any], list[str]]: + snap = machine_snapshot(self.cfg.work_root) + reasons: list[str] = [] + if snap["load"][0] > self.cfg.max_load1: + reasons.append( + f"load1 {snap['load'][0]:.2f} > {self.cfg.max_load1:.2f}" + ) + if snap["mem_available_mib"] < self.cfg.min_mem_available_mib: + reasons.append( + f"MemAvailable {snap['mem_available_mib']} MiB < " + f"{self.cfg.min_mem_available_mib} MiB" + ) + if snap["disk_free_gib"] < self.cfg.min_free_gib: + reasons.append( + f"disk free {snap['disk_free_gib']} GiB < {self.cfg.min_free_gib} GiB" + ) + db_bytes = _database_size_bytes(self.pg) + snap["database_size_gib"] = ( + None if db_bytes is None else round(db_bytes / 1024**3, 3) + ) + if db_bytes is not None and db_bytes > self.cfg.max_database_gib * 1024**3: + reasons.append( + f"database {db_bytes / 1024**3:.2f} GiB > " + f"{self.cfg.max_database_gib:.2f} GiB" + ) + return (not reasons), snap, reasons + + def code_gate(self) -> tuple[bool, str]: + current = code_fingerprint(self.cfg.project_root) + baseline = self.state.get("baseline_code_fingerprint") + if not baseline: + return True, current + return current == baseline, current + + + def _run_action(self, key: str, fn, *, exhaustible: bool = False) -> bool: + before = machine_snapshot(self.cfg.work_root) + started = time.monotonic() + attempt = int((self.state.get("attempts") or {}).get(key, 0)) + 1 + self.state["last_action"] = key + self.state["last_result"] = "RUNNING" + self.save_state() + self.set_status("RUNNING", action=key, attempt=attempt, machine=before) + try: + details = fn() or {} + after = machine_snapshot(self.cfg.work_root) + payload = { + "attempt": attempt, + "elapsed_seconds": round(time.monotonic() - started, 3), + "peak_rss_mib": peak_rss_mib(), + "machine_before": before, + "machine_after": after, + "details": details, + } + write_tag(self.cfg.tag_root, key, "PASS", payload) + self.state.setdefault("attempts", {}).pop(key, None) + self.state["last_result"] = "PASS" + self.save_state() + self.set_status("READY", action=key, result="PASS", tag=f"{key}.PASS.json") + return True + except Exception as exc: + after = machine_snapshot(self.cfg.work_root) + error = f"{type(exc).__name__}: {exc}" + payload = { + "attempt": attempt, + "elapsed_seconds": round(time.monotonic() - started, 3), + "peak_rss_mib": peak_rss_mib(), + "machine_before": before, + "machine_after": after, + "error": error, + "traceback": traceback.format_exc(), + } + fail_key = f"{key}.attempt{attempt}" + write_tag(self.cfg.tag_root, fail_key, "FAIL", payload) + self.state.setdefault("attempts", {})[key] = attempt + self.state["last_result"] = "FAIL" + if attempt >= self.cfg.segment_retry_limit: + if exhaustible: + exhausted = set(self.state.get("exhausted") or []) + exhausted.add(key) + self.state["exhausted"] = sorted(exhausted) + else: + self.state["paused"] = True + self.state["pause_reason"] = f"{key} exhausted retries: {error}" + self.save_state() + self.set_status("RETRY" if not self.state.get("paused") else "PAUSED", action=key, error=error, attempt=attempt) + return False + + + def precheck(self) -> dict[str, Any]: + archive = self.cfg.source_archive + if not archive.is_file(): + raise FileNotFoundError(archive) + scale_counts = {k: 0 for k in "ABCDEFGZ"} + uncompressed = 0 + compressed = 0 + files = 0 + with ZipFile(archive) as zf: + bad = zf.testzip() + if bad is not None: + raise RuntimeError(f"ZIP CRC failure: {bad}") + for zi in zf.infolist(): + if zi.is_dir(): + continue + files += 1 + ext = Path(zi.filename).suffix.upper().lstrip(".") + if ext in scale_counts: + scale_counts[ext] += 1 + uncompressed += int(zi.file_size) + compressed += int(zi.compress_size) + cell_count = sum(scale_counts.values()) + if cell_count < 10000: + raise RuntimeError(f"unexpected global cell count: {cell_count}") + archive_sha = sha256_file(archive) + code_sha = code_fingerprint(self.cfg.project_root) + self.state["archive_sha256"] = archive_sha + self.state["baseline_code_fingerprint"] = code_sha + self.state["source_expected_cells"] = cell_count + self.save_state() + return { + "archive": str(archive), + "archive_bytes": archive.stat().st_size, + "archive_sha256": archive_sha, + "zip_files": files, + "cell_count": cell_count, + "scale_counts": scale_counts, + "cell_uncompressed_bytes": uncompressed, + "cell_compressed_bytes": compressed, + "code_fingerprint": code_sha, + } + + def extract(self) -> dict[str, Any]: + marker = self.cfg.work_root / "source" / ".extract_complete.json" + expected_sha = str(self.state.get("archive_sha256") or "") + if marker.is_file(): + obj = json.loads(marker.read_text(encoding="utf-8")) + if obj.get("archive_sha256") == expected_sha and self.cfg.source_root.is_dir(): + return {"status": "REUSED", **obj} + partial = self.cfg.work_root / "source.partial" + final = self.cfg.work_root / "source" + if partial.exists(): + shutil.rmtree(partial) + partial.mkdir(parents=True, exist_ok=True) + with ZipFile(self.cfg.source_archive) as zf: + zf.extractall(partial) + extracted_root = partial / self.cfg.source_dir_name + if not extracted_root.is_dir(): + raise RuntimeError(f"missing extracted root: {extracted_root}") + scan = scan_cm93(extracted_root) + if len(scan.cells) != int(self.state["source_expected_cells"]): + raise RuntimeError( + f"extracted cell count mismatch: {len(scan.cells)} != " + f"{self.state['source_expected_cells']}" + ) + if final.exists(): + shutil.rmtree(final) + os.replace(partial, final) + obj = { + "archive_sha256": expected_sha, + "source_root": str(self.cfg.source_root), + "cell_count": len(scan.cells), + "completed_at": utc_now(), + } + atomic_json(marker, obj) + return obj + + + def inventory(self) -> dict[str, Any]: + scan = scan_cm93(self.cfg.source_root) + if not scan.object_dictionary or not scan.attribute_dictionary: + raise RuntimeError("global source dictionaries not found") + if len(scan.cells) != int(self.state["source_expected_cells"]): + raise RuntimeError("global source cell count changed after extraction") + tmp = self.cfg.source_manifest.with_suffix(".sqlite.tmp") + if tmp.exists(): + tmp.unlink() + db = sqlite3.connect(tmp) + try: + db.executescript( + """ + CREATE TABLE meta(k TEXT PRIMARY KEY,v TEXT NOT NULL); + CREATE TABLE cell( + ordinal INTEGER PRIMARY KEY, + source_relative_path TEXT NOT NULL UNIQUE, + scale_code TEXT NOT NULL, + native_scale INTEGER NOT NULL, + file_size INTEGER NOT NULL, + shard_index INTEGER NOT NULL + ); + CREATE INDEX idx_cell_shard ON cell(shard_index,ordinal); + """ + ) + h = hashlib.sha256() + rows = [] + scale_counts: dict[str, int] = {} + total_bytes = 0 + for i, c in enumerate(scan.cells): + rel = str(c.relative_path).replace(os.sep, "\\") + size = c.path.stat().st_size + shard = i % self.cfg.shard_count + rows.append((i, rel, c.scale_code, c.native_scale, size, shard)) + scale_counts[c.scale_code] = scale_counts.get(c.scale_code, 0) + 1 + total_bytes += size + h.update(f"{i}|{rel}|{c.scale_code}|{c.native_scale}|{size}\n".encode()) + db.executemany( + "INSERT INTO cell(ordinal,source_relative_path,scale_code," + "native_scale,file_size,shard_index) VALUES(?,?,?,?,?,?)", + rows, + ) + meta = { + "schema": "navsea.cm95.global-source-manifest/v1", + "archive_sha256": str(self.state["archive_sha256"]), + "source_fingerprint": h.hexdigest(), + "cell_count": str(len(rows)), + "shard_count": str(self.cfg.shard_count), + "created_at": utc_now(), + } + db.executemany("INSERT INTO meta(k,v) VALUES(?,?)", meta.items()) + db.commit() + finally: + db.close() + os.replace(tmp, self.cfg.source_manifest) + self.state["source_fingerprint"] = h.hexdigest() + self.state["source_total_bytes"] = total_bytes + self.save_state() + return { + "cell_count": len(rows), + "total_bytes": total_bytes, + "scale_counts": scale_counts, + "shard_count": self.cfg.shard_count, + "source_fingerprint": h.hexdigest(), + "object_dictionary": str(scan.object_dictionary), + "attribute_dictionary": str(scan.attribute_dictionary), + } + + def _quarantine_doc(self) -> dict[str, Any]: + obj = json.loads(self.cfg.quarantine.read_text(encoding="utf-8")) + if obj.get("schema") != "navsea.cm95.global-quarantine/v1": + raise RuntimeError("unsupported global quarantine schema") + entries = obj.get("entries") + if not isinstance(entries, list): + raise RuntimeError("global quarantine entries must be a list") + return obj + + def _quarantine_paths(self) -> list[str]: + return [str(x["source_relative_path"]) for x in self._quarantine_doc()["entries"]] + + def _quarantine_cli_args(self) -> list[str]: + out: list[str] = [] + for rel in self._quarantine_paths(): + out.extend(["--exclude-relative-path", rel]) + return out + + def quarantine_validate(self) -> dict[str, Any]: + obj = self._quarantine_doc() + archive_sha = str(self.state.get("archive_sha256") or "") + if obj.get("archive_sha256") != archive_sha: + raise RuntimeError("quarantine archive SHA256 does not match current source archive") + with sqlite3.connect(self.cfg.source_manifest) as db: + rows = db.execute( + "SELECT source_relative_path,file_size,shard_index FROM cell" + ).fetchall() + manifest = {normalize_relative_path(r[0]): (str(r[0]), int(r[1]), int(r[2])) for r in rows} + verified = [] + total_bytes = 0 + seen: set[str] = set() + for entry in obj["entries"]: + rel = str(entry["source_relative_path"]) + norm = normalize_relative_path(rel) + if norm in seen: + raise RuntimeError(f"duplicate quarantine path: {rel}") + seen.add(norm) + row = manifest.get(norm) + if row is None: + raise RuntimeError(f"quarantine path not in source manifest: {rel}") + path = self.cfg.source_root / Path(rel.replace("\\", "/")) + if not path.is_file(): + raise RuntimeError(f"quarantine source missing: {path}") + size = path.stat().st_size + expected_size = int(entry["file_size"]) + if size != expected_size or size != row[1]: + raise RuntimeError(f"quarantine size mismatch: {rel} {size} != {expected_size}") + digest = sha256_file(path) + if digest != str(entry["sha256"]): + raise RuntimeError(f"quarantine SHA256 mismatch: {rel}") + alt_rel = entry.get("alternate_source_relative_path") + alt_verified = None + if alt_rel: + alt = self.cfg.source_root / Path(str(alt_rel).replace("\\", "/")) + if not alt.is_file(): + raise RuntimeError(f"quarantine alternate source missing: {alt_rel}") + alt_digest = sha256_file(alt) + if alt_digest != str(entry.get("alternate_sha256") or ""): + raise RuntimeError(f"quarantine alternate SHA256 mismatch: {alt_rel}") + alt_verified = {"source_relative_path": str(alt_rel), "sha256": alt_digest} + total_bytes += size + verified.append({ + "source_relative_path": rel, + "sha256": digest, + "file_size": size, + "shard_index": row[2], + "reason": entry.get("reason"), + "alternate": alt_verified, + }) + importable_cells = int(self.state["source_expected_cells"]) - len(verified) + importable_bytes = int(self.state.get("source_total_bytes") or 0) - total_bytes + self.state["quarantine"] = verified + self.state["quarantine_count"] = len(verified) + self.state["quarantine_bytes"] = total_bytes + self.state["source_importable_cells"] = importable_cells + self.state["source_importable_bytes"] = importable_bytes + self.save_state() + return { + "quarantine_file": str(self.cfg.quarantine), + "quarantine_count": len(verified), + "quarantine_bytes": total_bytes, + "source_importable_cells": importable_cells, + "source_importable_bytes": importable_bytes, + "entries": verified, + } + + def _expected_shard_cells(self, index: int) -> int: + excluded = {normalize_relative_path(x) for x in self._quarantine_paths()} + with sqlite3.connect(self.cfg.source_manifest) as db: + rows = db.execute( + "SELECT source_relative_path FROM cell WHERE shard_index=?", (int(index),) + ).fetchall() + return sum(1 for row in rows if normalize_relative_path(row[0]) not in excluded) + + + def _run_logged(self, key: str, argv: list[str]) -> dict[str, Any]: + log = self.cfg.log_root / f"{key}.log" + log.parent.mkdir(parents=True, exist_ok=True) + with log.open("w", encoding="utf-8") as out: + out.write("command: " + " ".join(argv) + "\n") + out.flush() + cp = subprocess.run( + argv, + cwd=self.cfg.project_root, + env=_subprocess_env(self.cfg), + stdout=out, + stderr=subprocess.STDOUT, + text=True, + check=False, + ) + if cp.returncode != 0: + tail = "\n".join(log.read_text(encoding="utf-8", errors="replace").splitlines()[-30:]) + raise RuntimeError( + f"command exit {cp.returncode}; log={log}\n{tail}" + ) + return {"log": str(log), "returncode": cp.returncode} + + def source_scan_shard(self, index: int) -> dict[str, Any]: + report = self.cfg.report_root / "source_scan" / f"shard_{index:03d}.json" + report.parent.mkdir(parents=True, exist_ok=True) + argv = [ + str(self.cfg.project_root / ".venv/bin/cm95pg"), + "audit", + str(self.cfg.source_root), + "--source-only", + "--precision-samples", "0", + "--shard-count", str(self.cfg.shard_count), + "--shard-index", str(index), + "--report", str(report), + ] + self._quarantine_cli_args() + result = self._run_logged(f"SOURCE_SCAN_{index:03d}", argv) + obj = json.loads(report.read_text(encoding="utf-8")) + expected = self._expected_shard_cells(index) + if not obj.get("passed"): + raise RuntimeError(f"source scan shard {index} failed audit") + if int(obj.get("source_cells") or 0) != expected: + raise RuntimeError( + f"source scan shard {index} count mismatch " + f"{obj.get('source_cells')} != {expected}" + ) + return {**result, "report": str(report), "expected_cells": expected, "audit": obj} + + + def source_scan_complete(self) -> dict[str, Any]: + totals = { + "source_cells": 0, + "source_features": 0, + "source_attributes": 0, + "source_points": 0, + "source_lines": 0, + "source_areas": 0, + "source_coverages": 0, + "source_sounding_points": 0, + "unknown_objects": 0, + "unknown_attributes": 0, + "source_parse_failed": 0, + } + for i in range(self.cfg.shard_count): + report = self.cfg.report_root / "source_scan" / f"shard_{i:03d}.json" + if not report.is_file(): + raise RuntimeError(f"missing source scan report: {report}") + obj = json.loads(report.read_text(encoding="utf-8")) + if not obj.get("passed"): + raise RuntimeError(f"source scan report not PASS: shard {i}") + for k in totals: + totals[k] += int(obj.get(k) or 0) + expected_cells = int(self.state.get("source_importable_cells") or self.state["source_expected_cells"]) + if totals["source_cells"] != expected_cells: + raise RuntimeError( + f"global source scan cell total mismatch: " + f"{totals['source_cells']} != {expected_cells}" + ) + if totals["source_parse_failed"] or totals["unknown_objects"] or totals["unknown_attributes"]: + raise RuntimeError(f"global source scan gate failed: {totals}") + self.state["source_stats"] = totals + self.save_state() + return totals + + def db_init(self) -> dict[str, Any]: + create_database(self.pg) + schema = self.cfg.project_root / "sql" / "001_schema_postgresql.sql" + apply_schema(self.pg, schema) + with connect(self.pg) as conn: + with conn.cursor() as cur: + cur.execute("SHOW server_version") + pg_version = str(cur.fetchone()["server_version"]) + cur.execute("SELECT PostGIS_Version() AS v") + postgis = str(cur.fetchone()["v"]) + conn.rollback() + return { + "database": self.cfg.database, + "postgresql": pg_version, + "postgis": postgis, + "database_size_bytes": _database_size_bytes(self.pg), + } + + def import_shard(self, index: int) -> dict[str, Any]: + argv = [ + str(self.cfg.project_root / ".venv/bin/cm95pg"), + "import", + str(self.cfg.source_root), + "--shard-count", str(self.cfg.shard_count), + "--shard-index", str(index), + ] + self._quarantine_cli_args() + result = self._run_logged(f"POSTGIS_IMPORT_{index:03d}", argv) + expected = self._expected_shard_cells(index) + return { + **result, + "expected_cells": expected, + "database_size_bytes": _database_size_bytes(self.pg), + } + + + def import_complete(self) -> dict[str, Any]: + row = _dataset_row(self.pg, self.cfg.source_root) + if row is None: + raise RuntimeError("global dataset missing after import") + did = int(row["id"]) + with connect(self.pg) as conn: + with conn.cursor() as cur: + cur.execute( + "UPDATE chart_import_job SET status='CANCELLED',finished_at=NOW()," + "summary_json=COALESCE(summary_json,'{}'::jsonb) || %s::jsonb " + "WHERE dataset_id=%s AND status='RUNNING'", + (json.dumps({"global_pipeline": "stale_run_reconciled"}), did), + ) + reconciled_jobs = int(cur.rowcount) + queries = { + "cells": "SELECT COUNT(*) n FROM chart_cell WHERE dataset_id=%s", + "cells_done": "SELECT COUNT(*) n FROM chart_cell WHERE dataset_id=%s AND import_status='DONE'", + "features": "SELECT COUNT(*) n FROM chart_feature WHERE dataset_id=%s", + "attributes": "SELECT COUNT(*) n FROM chart_feature_attribute a JOIN chart_feature f ON f.id=a.feature_id WHERE f.dataset_id=%s", + "points": "SELECT COUNT(*) n FROM chart_point_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.dataset_id=%s", + "lines": "SELECT COUNT(*) n FROM chart_line_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.dataset_id=%s", + "areas": "SELECT COUNT(*) n FROM chart_area_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.dataset_id=%s", + "coverages": "SELECT COUNT(*) n FROM chart_coverage WHERE dataset_id=%s", + "soundings": "SELECT COUNT(*) n FROM chart_sounding_point s JOIN chart_feature f ON f.id=s.feature_id WHERE f.dataset_id=%s", + "raw_bytes": "SELECT COALESCE(SUM(OCTET_LENGTH(r.raw_blob)),0) n FROM chart_raw_record r JOIN chart_cell c ON c.id=r.cell_id WHERE c.dataset_id=%s", + } + counts = {} + for key, sql in queries.items(): + cur.execute(sql, (did,)) + counts[key] = int(cur.fetchone()["n"]) + conn.commit() + expected = dict(self.state.get("source_stats") or {}) + checks = { + "cells": int(self.state.get("source_importable_cells") or self.state["source_expected_cells"]), + "cells_done": int(self.state.get("source_importable_cells") or self.state["source_expected_cells"]), + "features": int(expected["source_features"]), + "attributes": int(expected["source_attributes"]), + "points": int(expected["source_points"]), + "lines": int(expected["source_lines"]), + "areas": int(expected["source_areas"]), + "coverages": int(expected["source_coverages"]), + "soundings": int(expected["source_sounding_points"]), + } + mismatches = { + k: {"expected": v, "actual": counts.get(k)} + for k, v in checks.items() + if counts.get(k) != v + } + expected_raw_bytes = int(self.state.get("source_importable_bytes") or self.state.get("source_total_bytes") or 0) + if expected_raw_bytes and counts["raw_bytes"] != expected_raw_bytes: + mismatches["raw_bytes"] = {"expected": expected_raw_bytes, "actual": counts["raw_bytes"]} + if mismatches: + raise RuntimeError(f"global import count mismatch: {mismatches}") + fp = None + with connect(self.pg) as conn: + fp = dataset_fingerprint(conn, did) + conn.rollback() + self.state["dataset_id"] = did + self.state["dataset_fingerprint"] = fp + self.save_state() + return { + "dataset_id": did, + "dataset_fingerprint": fp, + "counts": counts, + "reconciled_stale_jobs": reconciled_jobs, + "database_size_bytes": _database_size_bytes(self.pg), + } + + + def db_audit_shard(self, index: int) -> dict[str, Any]: + report = self.cfg.report_root / "db_audit" / f"shard_{index:03d}.json" + report.parent.mkdir(parents=True, exist_ok=True) + argv = [ + str(self.cfg.project_root / ".venv/bin/cm95pg"), + "audit", + str(self.cfg.source_root), + "--precision-samples", "16", + "--shard-count", str(self.cfg.shard_count), + "--shard-index", str(index), + "--report", str(report), + ] + self._quarantine_cli_args() + result = self._run_logged(f"DB_AUDIT_{index:03d}", argv) + obj = json.loads(report.read_text(encoding="utf-8")) + expected = self._expected_shard_cells(index) + if not obj.get("passed"): + raise RuntimeError(f"DB audit shard {index} failed") + if int(obj.get("source_cells") or 0) != expected: + raise RuntimeError( + f"DB audit shard {index} cell count mismatch " + f"{obj.get('source_cells')} != {expected}" + ) + return {**result, "report": str(report), "expected_cells": expected, "audit": obj} + + def db_audit_complete(self) -> dict[str, Any]: + aggregate = { + "source_cells": 0, + "source_parse_failed": 0, + "db_missing_cells": 0, + "db_not_done_cells": 0, + "count_mismatches": 0, + "hash_mismatches": 0, + "raw_preservation_mismatches": 0, + "unknown_objects": 0, + "unknown_attributes": 0, + "precision_checked": 0, + "precision_failed": 0, + } + max_delta = 0.0 + for i in range(self.cfg.shard_count): + report = self.cfg.report_root / "db_audit" / f"shard_{i:03d}.json" + obj = json.loads(report.read_text(encoding="utf-8")) + if not obj.get("passed"): + raise RuntimeError(f"DB audit report not PASS: shard {i}") + for k in aggregate: + aggregate[k] += int(obj.get(k) or 0) + max_delta = max(max_delta, float(obj.get("max_precision_delta_deg") or 0.0)) + aggregate["max_precision_delta_deg"] = max_delta + expected_cells = int(self.state.get("source_importable_cells") or self.state["source_expected_cells"]) + if aggregate["source_cells"] != expected_cells: + raise RuntimeError("DB audit aggregate cell count mismatch") + bad_keys = ( + "source_parse_failed", "db_missing_cells", "db_not_done_cells", + "count_mismatches", "hash_mismatches", "raw_preservation_mismatches", + "unknown_objects", "unknown_attributes", "precision_failed", + ) + if any(int(aggregate[k]) for k in bad_keys): + raise RuntimeError(f"DB audit aggregate failed: {aggregate}") + return aggregate + + + def fid_manifest(self) -> dict[str, Any]: + did = int(self.state["dataset_id"]) + stats = dict(self.state["source_stats"]) + identities = int(stats["source_features"]) + int(stats["source_sounding_points"]) + max_possible_id = identities * ROLE_FACTOR + (ROLE_FACTOR - 1) + if max_possible_id > MAX_SIGNED_INT32: + raise RuntimeError( + f"global deterministic FID capacity exceeded: identities={identities} " + f"max_id={max_possible_id} limit={MAX_SIGNED_INT32}" + ) + argv = [ + str(self.cfg.project_root / ".venv/bin/cm95pbf"), + "build-fid-manifest", + "--dataset-id", str(did), + "--manifest", str(self.cfg.fid_manifest), + "--rebuild-fid-manifest", + ] + result = self._run_logged("FID_MANIFEST", argv) + with connect(self.pg) as conn: + info = validate_manifest(conn, did, self.cfg.fid_manifest) + conn.rollback() + if int(info["identity_count"]) != identities: + raise RuntimeError( + f"FID identity count mismatch: {info['identity_count']} != {identities}" + ) + self.state["fid_identity_count"] = identities + self.save_state() + return { + **result, + "identity_count": identities, + "max_possible_mvt_id": max_possible_id, + "capacity_fraction": max_possible_id / MAX_SIGNED_INT32, + "manifest": str(self.cfg.fid_manifest), + "manifest_bytes": self.cfg.fid_manifest.stat().st_size, + "dataset_fingerprint": info["dataset_fingerprint"], + } + + def _ensure_pbf_contract(self) -> tuple[GlobalPbfLedger, str]: + resource = json.loads(self.cfg.s52_resource_contract.read_text(encoding="utf-8")) + if resource.get("status") != "PASS": + raise RuntimeError("S-52 resource contract is not PASS") + did = int(self.state["dataset_id"]) + fp = str(self.state["dataset_fingerprint"]) + release_id, contract = _release_contract(self.cfg, did, fp) + ledger = GlobalPbfLedger(self.cfg.pbf_ledger) + expected = { + "schema": GLOBAL_LEDGER_SCHEMA, + "dataset_id": str(did), + "dataset_fingerprint": fp, + "release_id": release_id, + "contract_sha256": contract, + "builder_contract_version": BUILDER_CONTRACT_VERSION, + } + for k, v in expected.items(): + old = ledger.get_meta(k) + if old is not None and old != v: + raise RuntimeError(f"global PBF ledger contract mismatch: {k} {old} != {v}") + ledger.set_meta(k, v) + self.state["release_id"] = release_id + self.state["pbf_contract_sha256"] = contract + self.save_state() + return ledger, release_id + + + @staticmethod + def _tile_bbox_estimate(z: int, bounds: tuple[float, float, float, float]) -> int: + half = 20037508.342789244 + world = half * 2.0 + n = 1 << int(z) + minx, miny, maxx, maxy = bounds + + def tx(x: float) -> int: + return max(0, min(n - 1, int(math.floor((x + half) / world * n)))) + + def ty(y: float) -> int: + return max(0, min(n - 1, int(math.floor((half - y) / world * n)))) + + x0, x1 = sorted((tx(minx), tx(maxx))) + y0, y1 = sorted((ty(maxy), ty(miny))) + return (x1 - x0 + 1) * (y1 - y0 + 1) + + def plan_zoom(self, z: int) -> dict[str, Any]: + ledger, release_id = self._ensure_pbf_contract() + if ledger.zoom_done(z): + with ledger._conn() as db: + row = db.execute("SELECT * FROM zoom_plan WHERE z=?", (int(z),)).fetchone() + return {"status": "REUSED", **dict(row), "release_id": release_id} + did = int(self.state["dataset_id"]) + policy = _load_policy(self.cfg.policy) + enumeration = _enumeration_scales(policy, int(z)) + with connect(self.pg) as conn: + fp = dataset_fingerprint(conn, did) + if fp != str(self.state["dataset_fingerprint"]): + raise RuntimeError("dataset fingerprint changed before PBF planning") + validate_manifest(conn, did, self.cfg.fid_manifest) + pieces = _fetch_plan_geometries(conn, did, enumeration) + conn.rollback() + fallback = 0 + max_piece_tiles = 0 + inserted_piece_tiles = 0 + now = utc_now() + with ledger._conn() as ldb: + for piece_index, (_cid, _scale, geom, source) in enumerate(pieces, 1): + if geom is None or geom.is_empty: + continue + estimate = self._tile_bbox_estimate(int(z), geom.bounds) + max_piece_tiles = max(max_piece_tiles, estimate) + if estimate > 500_000: + raise RuntimeError( + f"suspicious coverage span at z{z}: estimated_bbox_tiles={estimate}; " + "possible dateline/wrap geometry; stop before global materialization" + ) + tiles = _tiles_intersecting_geom(int(z), geom) + inserted_piece_tiles += len(tiles) + ldb.executemany( + "INSERT OR IGNORE INTO tile_job" + "(profile,z,x,y,state,attempts,updated_at) " + "VALUES(?,?,?,?, 'PENDING',0,?)", + (("delivery", int(a), int(b), int(c), now) for a,b,c in tiles), + ) + if piece_index % 200 == 0: + ldb.commit() + ldb.commit() + tile_count = ledger.finish_zoom(int(z), len(pieces)) + return { + "release_id": release_id, + "z": int(z), + "enumeration_scales": enumeration, + "coverage_pieces": len(pieces), + "cell_bbox_fallback_pieces": fallback, + "piece_tile_instances": inserted_piece_tiles, + "unique_tile_count": tile_count, + "max_piece_bbox_tile_estimate": max_piece_tiles, + } + + def coverage_plan_complete(self) -> dict[str, Any]: + ledger, release_id = self._ensure_pbf_contract() + rows = [] + with ledger._conn() as db: + rows = [dict(r) for r in db.execute("SELECT * FROM zoom_plan ORDER BY z").fetchall()] + planned = {int(r["z"]) for r in rows if r["state"] == "PASS"} + if planned != set(self.cfg.pbf_zooms): + raise RuntimeError( + f"coverage plan incomplete: planned={sorted(planned)} expected={list(self.cfg.pbf_zooms)}" + ) + summary = ledger.summary(self.cfg.pbf_max_attempts) + self.state["pbf_planned_jobs"] = summary["jobs"] + self.save_state() + return {"release_id": release_id, "zooms": rows, **summary} + + + def pbf_segment(self) -> dict[str, Any]: + ledger, release_id = self._ensure_pbf_contract() + requeued = ledger.requeue_running() + rows = ledger.runnable(self.cfg.pbf_segment_jobs, self.cfg.pbf_max_attempts) + if not rows: + summary = ledger.summary(self.cfg.pbf_max_attempts) + if summary["exhausted_failed"]: + raise RuntimeError( + f"PBF exhausted failures: {summary['exhausted_failed']}" + ) + if summary["states"].get("PENDING", 0) or summary["states"].get("RUNNING", 0): + raise RuntimeError(f"PBF ledger has non-runnable pending work: {summary}") + return {"complete": True, "requeued": requeued, "release_id": release_id, **summary} + + release_dir = self.cfg.pbf_root / release_id + output_root = release_dir / "pbf" / "delivery" + report_root = release_dir / "reports" + output_root.mkdir(parents=True, exist_ok=True) + report_root.mkdir(parents=True, exist_ok=True) + + def work(row): + ledger.mark_running(row) + z, x, y = int(row["z"]), int(row["x"]), int(row["y"]) + try: + builder = Cm95TileBuilder( + db=self.pg, + dataset_id=int(self.state["dataset_id"]), + mapping_path=self.cfg.mapping, + policy_path=self.cfg.policy, + fid_manifest_path=self.cfg.fid_manifest, + output_root=output_root, + report_root=report_root, + profile="delivery", + s52_lookups_path=self.cfg.s52_lookups, + s52_point_table="Paper", + s52_boundary_table="Plain", + s52_mariner_mode="runtime", + ) + result = builder.build_tile(TileJob(z, x, y)) + if result.get("hard_gate_unmapped"): + raise RuntimeError( + f"hard-gate unmapped: {result['hard_gate_unmapped']}" + ) + ledger.mark_pass(row, result) + return True, z, x, y, result + except Exception as exc: + err = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()}" + ledger.mark_failed(row, err) + return False, z, x, y, err + + ok_count = 0 + fail_count = 0 + started = time.monotonic() + with ThreadPoolExecutor( + max_workers=self.cfg.pbf_workers, + thread_name_prefix="cm95global-pbf", + ) as pool: + futures = [pool.submit(work, row) for row in rows] + for future in as_completed(futures): + ok, z, x, y, payload = future.result() + if ok: + ok_count += 1 + print( + f"PASS {z}/{x}/{y} features=" + f"{payload['output_feature_count']} layers={payload['output_layer_count']}" + ) + else: + fail_count += 1 + print(f"FAIL {z}/{x}/{y} {str(payload).splitlines()[0]}") + summary = ledger.summary(self.cfg.pbf_max_attempts) + return { + "complete": False, + "release_id": release_id, + "selected_jobs": len(rows), + "pass_this_segment": ok_count, + "fail_this_segment": fail_count, + "requeued": requeued, + "build_seconds": round(time.monotonic() - started, 3), + **summary, + } + + + def final_audit(self) -> dict[str, Any]: + ledger, release_id = self._ensure_pbf_contract() + summary = ledger.summary(self.cfg.pbf_max_attempts) + if summary["exhausted_failed"]: + raise RuntimeError(f"final audit: exhausted PBF failures {summary['exhausted_failed']}") + non_pass = { + k: v for k, v in summary["states"].items() + if k != "PASS" and int(v) > 0 + } + if non_pass: + raise RuntimeError(f"final audit: non-PASS PBF jobs {non_pass}") + if int(summary["states"].get("PASS", 0)) != int(summary["jobs"]): + raise RuntimeError("final audit: PBF PASS count != planned jobs") + + did = int(self.state["dataset_id"]) + with connect(self.pg) as conn: + current_fp = dataset_fingerprint(conn, did) + validate_manifest(conn, did, self.cfg.fid_manifest) + with conn.cursor() as cur: + cur.execute( + "SELECT COUNT(*) n FROM chart_cell WHERE dataset_id=%s " + "AND import_status='DONE'", + (did,), + ) + done_cells = int(cur.fetchone()["n"]) + conn.rollback() + if current_fp != str(self.state["dataset_fingerprint"]): + raise RuntimeError("final audit: dataset fingerprint changed") + if done_cells != int(self.state["source_expected_cells"]): + raise RuntimeError("final audit: DONE cell count changed") + + release_dir = self.cfg.pbf_root / release_id + actual_files = 0 + actual_bytes = 0 + pbf_dir = release_dir / "pbf" / "delivery" + for root, _dirs, files in os.walk(pbf_dir): + for name in files: + if not name.endswith(".pbf"): + continue + p = Path(root) / name + actual_files += 1 + actual_bytes += p.stat().st_size + if actual_files != int(summary["pbf_files"]): + raise RuntimeError( + f"final audit: PBF file count {actual_files} != ledger {summary['pbf_files']}" + ) + + resource_contract = json.loads( + self.cfg.s52_resource_contract.read_text(encoding="utf-8") + ) + if resource_contract.get("status") != "PASS": + raise RuntimeError("final audit: S-52 resource contract is not PASS") + manifest = { + "schema": GLOBAL_RELEASE_SCHEMA, + "status": "PASS", + "release_id": release_id, + "created_at": utc_now(), + "dataset_id": did, + "dataset_fingerprint": current_fp, + "source_archive_sha256": self.state["archive_sha256"], + "source_fingerprint": self.state["source_fingerprint"], + "source_cells": self.state["source_expected_cells"], + "fid_identity_count": self.state["fid_identity_count"], + "builder_contract_version": BUILDER_CONTRACT_VERSION, + "pbf_contract_sha256": self.state["pbf_contract_sha256"], + "planned_tile_jobs": summary["jobs"], + "pass_tile_jobs": summary["states"].get("PASS", 0), + "pbf_file_count": actual_files, + "pbf_bytes": actual_bytes, + "output_features": summary["output_features"], + "mapping_sha256": sha256_file(self.cfg.mapping), + "policy_sha256": sha256_file(self.cfg.policy), + "s52_lookups_sha256": sha256_file(self.cfg.s52_lookups), + "s52_resource_contract_sha256": sha256_file(self.cfg.s52_resource_contract), + } + out = release_dir / "global_release_manifest_v1.json" + atomic_json(out, manifest) + return {**manifest, "manifest": str(out)} + + + def _stage_exhausted(self, prefix: str) -> list[str]: + return sorted( + x for x in set(self.state.get("exhausted") or []) + if x.startswith(prefix) + ) + + def pbf_complete(self) -> dict[str, Any]: + ledger, release_id = self._ensure_pbf_contract() + ledger.requeue_running() + summary = ledger.summary(self.cfg.pbf_max_attempts) + if summary["exhausted_failed"]: + raise RuntimeError(f"PBF exhausted failures: {summary['exhausted_failed']}") + states = summary["states"] + if int(states.get("PASS", 0)) != int(summary["jobs"]): + raise RuntimeError(f"PBF is not complete: {summary}") + return {"release_id": release_id, **summary} + + def run_next(self) -> dict[str, Any]: + if self.state.get("paused"): + self.set_status("PAUSED") + return {"status": "PAUSED", "reason": self.state.get("pause_reason")} + + code_ok, current_code = self.code_gate() + if not code_ok: + self.pause( + "CODE_VERSION_CHANGED: " + f"{self.state.get('baseline_code_fingerprint')} -> {current_code}" + ) + return {"status": "PAUSED", "reason": self.state.get("pause_reason")} + + healthy, machine, reasons = self.health_gate() + if not healthy: + self.set_status("WAIT_RESOURCE", machine=machine, reasons=reasons) + return {"status": "WAIT_RESOURCE", "reasons": reasons, "machine": machine} + + if not self.tag_passed("PRECHECK"): + self._run_action("PRECHECK", self.precheck) + return {"status": self.state.get("last_result"), "action": "PRECHECK"} + if not self.tag_passed("EXTRACT"): + self._run_action("EXTRACT", self.extract) + return {"status": self.state.get("last_result"), "action": "EXTRACT"} + if not self.tag_passed("INVENTORY"): + self._run_action("INVENTORY", self.inventory) + return {"status": self.state.get("last_result"), "action": "INVENTORY"} + if not self.tag_passed("QUARANTINE_VALIDATE"): + self._run_action("QUARANTINE_VALIDATE", self.quarantine_validate) + return {"status": self.state.get("last_result"), "action": "QUARANTINE_VALIDATE"} + + for i in range(self.cfg.shard_count): + key = f"SOURCE_SCAN_{i:03d}" + if not self.tag_passed(key) and not self.exhausted(key): + self._run_action(key, lambda i=i: self.source_scan_shard(i), exhaustible=True) + return {"status": self.state.get("last_result"), "action": key} + exhausted = self._stage_exhausted("SOURCE_SCAN_") + if exhausted: + self.pause(f"SOURCE_SCAN exhausted segments: {exhausted}") + return {"status": "PAUSED", "reason": self.state.get("pause_reason")} + if not self.tag_passed("SOURCE_SCAN_COMPLETE"): + self._run_action("SOURCE_SCAN_COMPLETE", self.source_scan_complete) + return {"status": self.state.get("last_result"), "action": "SOURCE_SCAN_COMPLETE"} + + + if not self.tag_passed("DB_INIT"): + self._run_action("DB_INIT", self.db_init) + return {"status": self.state.get("last_result"), "action": "DB_INIT"} + + for i in range(self.cfg.shard_count): + key = f"POSTGIS_IMPORT_{i:03d}" + if not self.tag_passed(key) and not self.exhausted(key): + self._run_action(key, lambda i=i: self.import_shard(i), exhaustible=True) + return {"status": self.state.get("last_result"), "action": key} + exhausted = self._stage_exhausted("POSTGIS_IMPORT_") + if exhausted: + self.pause(f"POSTGIS_IMPORT exhausted segments: {exhausted}") + return {"status": "PAUSED", "reason": self.state.get("pause_reason")} + if not self.tag_passed("POSTGIS_IMPORT_COMPLETE"): + self._run_action("POSTGIS_IMPORT_COMPLETE", self.import_complete) + return {"status": self.state.get("last_result"), "action": "POSTGIS_IMPORT_COMPLETE"} + + for i in range(self.cfg.shard_count): + key = f"DB_AUDIT_{i:03d}" + if not self.tag_passed(key) and not self.exhausted(key): + self._run_action(key, lambda i=i: self.db_audit_shard(i), exhaustible=True) + return {"status": self.state.get("last_result"), "action": key} + exhausted = self._stage_exhausted("DB_AUDIT_") + if exhausted: + self.pause(f"DB_AUDIT exhausted segments: {exhausted}") + return {"status": "PAUSED", "reason": self.state.get("pause_reason")} + if not self.tag_passed("DB_AUDIT_COMPLETE"): + self._run_action("DB_AUDIT_COMPLETE", self.db_audit_complete) + return {"status": self.state.get("last_result"), "action": "DB_AUDIT_COMPLETE"} + + if not self.tag_passed("FID_MANIFEST"): + self._run_action("FID_MANIFEST", self.fid_manifest) + return {"status": self.state.get("last_result"), "action": "FID_MANIFEST"} + + for z in self.cfg.pbf_zooms: + key = f"COVERAGE_PLAN_Z{z:02d}" + if not self.tag_passed(key): + self._run_action(key, lambda z=z: self.plan_zoom(z)) + return {"status": self.state.get("last_result"), "action": key} + if not self.tag_passed("COVERAGE_PLAN_COMPLETE"): + self._run_action("COVERAGE_PLAN_COMPLETE", self.coverage_plan_complete) + return {"status": self.state.get("last_result"), "action": "COVERAGE_PLAN_COMPLETE"} + + + if not self.tag_passed("BASE_PBF_COMPLETE"): + ledger, _release_id = self._ensure_pbf_contract() + ledger.requeue_running() + summary = ledger.summary(self.cfg.pbf_max_attempts) + if ( + int(summary["states"].get("PASS", 0)) == int(summary["jobs"]) + and int(summary["jobs"]) > 0 + ): + self._run_action("BASE_PBF_COMPLETE", self.pbf_complete) + return {"status": self.state.get("last_result"), "action": "BASE_PBF_COMPLETE"} + seq = int(self.state.get("pbf_segment_seq") or 0) + 1 + self.state["pbf_segment_seq"] = seq + self.save_state() + key = f"BASE_PBF_SEGMENT_{seq:06d}" + self._run_action(key, self.pbf_segment) + return {"status": self.state.get("last_result"), "action": key} + + if not self.tag_passed("FINAL_AUDIT"): + self._run_action("FINAL_AUDIT", self.final_audit) + return {"status": self.state.get("last_result"), "action": "FINAL_AUDIT"} + + if not self.tag_passed("PIPELINE_DONE"): + self._run_action( + "PIPELINE_DONE", + lambda: { + "release_id": self.state.get("release_id"), + "note": "Global CM95 base PostGIS + sparse PBF pipeline complete; bathymetry is a separate pipeline.", + }, + ) + return {"status": self.state.get("last_result"), "action": "PIPELINE_DONE"} + + self.set_status( + "DONE", + release_id=self.state.get("release_id"), + work_root=str(self.cfg.work_root), + ) + return { + "status": "DONE", + "release_id": self.state.get("release_id"), + "work_root": str(self.cfg.work_root), + } + + def status(self) -> dict[str, Any]: + machine = machine_snapshot(self.cfg.work_root) + obj = { + "schema": "navsea.cm95.global-status/v1", + "paused": bool(self.state.get("paused")), + "pause_reason": self.state.get("pause_reason"), + "last_action": self.state.get("last_action"), + "last_result": self.state.get("last_result"), + "baseline_code_fingerprint": self.state.get("baseline_code_fingerprint"), + "archive_sha256": self.state.get("archive_sha256"), + "source_expected_cells": self.state.get("source_expected_cells"), + "source_fingerprint": self.state.get("source_fingerprint"), + "dataset_id": self.state.get("dataset_id"), + "dataset_fingerprint": self.state.get("dataset_fingerprint"), + "fid_identity_count": self.state.get("fid_identity_count"), + "release_id": self.state.get("release_id"), + "pbf_segment_seq": self.state.get("pbf_segment_seq"), + "exhausted": self.state.get("exhausted") or [], + "machine": machine, + } + if self.cfg.pbf_ledger.exists(): + obj["pbf"] = GlobalPbfLedger(self.cfg.pbf_ledger).summary( + self.cfg.pbf_max_attempts + ) + return obj diff --git a/src/cm95global/state.py b/src/cm95global/state.py new file mode 100644 index 0000000..5b18e7d --- /dev/null +++ b/src/cm95global/state.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +from datetime import datetime, timezone +import fcntl +import hashlib +import json +import os +from pathlib import Path +import resource +import shutil +from typing import Any + + +STATE_SCHEMA = "navsea.cm95.global-state/v1" +TAG_SCHEMA = "navsea.cm95.global-segment-tag/v1" + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def atomic_json(path: Path, obj: dict[str, Any]) -> None: + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(path.name + ".tmp") + tmp.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8") + os.replace(tmp, path) + + +def load_json(path: Path, default: dict[str, Any] | None = None) -> dict[str, Any]: + path = Path(path) + if not path.exists(): + return dict(default or {}) + return json.loads(path.read_text(encoding="utf-8")) + + +def sha256_file(path: Path, block: int = 4 * 1024 * 1024) -> str: + h = hashlib.sha256() + with Path(path).open("rb") as f: + for chunk in iter(lambda: f.read(block), b""): + h.update(chunk) + return h.hexdigest() + + +def code_fingerprint(project_root: Path) -> str: + root = Path(project_root) + items: list[Path] = [] + for rel in ("src/cm93db", "src/cm95pbf", "src/cm95global", "config", "resources/s52", "sql"): + base = root / rel + if base.is_dir(): + items.extend(p for p in base.rglob("*") if p.is_file() and "__pycache__" not in p.parts) + elif base.is_file(): + items.append(base) + for rel in ("pyproject.toml", "requirements.txt"): + p = root / rel + if p.is_file(): + items.append(p) + h = hashlib.sha256() + for p in sorted(set(items), key=lambda x: str(x.relative_to(root))): + rel = str(p.relative_to(root)).replace(os.sep, "/") + h.update(rel.encode()) + h.update(b"\0") + h.update(str(p.stat().st_size).encode()) + h.update(b"\0") + with p.open("rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +def _meminfo_kib() -> dict[str, int]: + out: dict[str, int] = {} + for line in Path("/proc/meminfo").read_text().splitlines(): + key, rest = line.split(":", 1) + value = rest.strip().split()[0] + out[key] = int(value) + return out + + +def machine_snapshot(work_root: Path) -> dict[str, Any]: + mem = _meminfo_kib() + target = Path(work_root) + probe = target if target.exists() else target.parent + probe.mkdir(parents=True, exist_ok=True) + disk = shutil.disk_usage(probe) + load1, load5, load15 = os.getloadavg() + return { + "at": utc_now(), + "load": [load1, load5, load15], + "mem_available_mib": round(mem.get("MemAvailable", 0) / 1024, 1), + "swap_used_mib": round((mem.get("SwapTotal", 0) - mem.get("SwapFree", 0)) / 1024, 1), + "disk_total_gib": round(disk.total / 1024**3, 2), + "disk_used_gib": round(disk.used / 1024**3, 2), + "disk_free_gib": round(disk.free / 1024**3, 2), + } + + +def peak_rss_mib() -> float: + self_kib = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + child_kib = resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss + return round(max(self_kib, child_kib) / 1024, 1) + + +class PipelineLock: + def __init__(self, path: Path): + self.path = Path(path) + self.fd = None + + def __enter__(self): + self.path.parent.mkdir(parents=True, exist_ok=True) + self.fd = self.path.open("w") + try: + fcntl.flock(self.fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + self.fd.close() + raise RuntimeError("global pipeline already running") from exc + self.fd.write(f"{os.getpid()}\n") + self.fd.flush() + return self + + + def __exit__(self, exc_type, exc, tb): + if self.fd is not None: + try: + fcntl.flock(self.fd.fileno(), fcntl.LOCK_UN) + finally: + self.fd.close() + + +def default_state() -> dict[str, Any]: + return { + "schema": STATE_SCHEMA, + "created_at": utc_now(), + "updated_at": utc_now(), + "paused": False, + "pause_reason": None, + "baseline_code_fingerprint": None, + "archive_sha256": None, + "attempts": {}, + "exhausted": [], + "pbf_segment_seq": 0, + "last_action": None, + "last_result": None, + } + + +def write_tag(tag_root: Path, key: str, result: str, payload: dict[str, Any]) -> Path: + safe = key.replace("/", "_").replace(" ", "_") + path = Path(tag_root) / f"{safe}.{result}.json" + atomic_json(path, { + "schema": TAG_SCHEMA, + "key": key, + "result": result, + "written_at": utc_now(), + **payload, + }) + return path diff --git a/src/cm95pbf/production.py b/src/cm95pbf/production.py index bad2594..1a6ba9a 100644 --- a/src/cm95pbf/production.py +++ b/src/cm95pbf/production.py @@ -195,10 +195,9 @@ def _connect(db: DbConfig): def _fetch_plan_geometries(conn, dataset_id: int, scales: list[int]): coverage_wkb_sql = as_wkb_sql("geom") - placeholders = ",".join(["%s"] * len(scales)) with conn.cursor() as cur: cur.execute( - f""" + """ SELECT id, native_scale, @@ -208,10 +207,10 @@ def _fetch_plan_geometries(conn, dataset_id: int, scales: list[int]): bbox_max_lat FROM chart_cell WHERE dataset_id=%s - AND native_scale IN ({placeholders}) + AND native_scale = ANY(%s) ORDER BY native_scale, id """, - (dataset_id, *scales), + (dataset_id, list(scales)), ) cells = list(cur.fetchall()) @@ -219,17 +218,16 @@ def _fetch_plan_geometries(conn, dataset_id: int, scales: list[int]): return [] cell_ids = [int(c["id"]) for c in cells] - ph = ",".join(["%s"] * len(cell_ids)) with conn.cursor() as cur: cur.execute( f""" SELECT cell_id, {coverage_wkb_sql} AS geom_wkb FROM chart_coverage WHERE dataset_id=%s - AND cell_id IN ({ph}) + AND cell_id = ANY(%s) ORDER BY cell_id, id """, - (dataset_id, *cell_ids), + (dataset_id, cell_ids), ) coverage_rows = list(cur.fetchall())