From 421234301ae7db33f7c0703f497e60a76381b514 Mon Sep 17 00:00:00 2001 From: linkong Date: Thu, 30 Apr 2026 16:56:37 +0800 Subject: [PATCH] release: bump version to 0.47.0 --- .claude/commands/docs.md | 277 +++------ .codex/skills/docs/SKILL.md | 190 ++---- TODO.md | 2 + VERSION | 2 +- backend/app/api/v1/datasource_config.py | 12 +- backend/app/api/v1/settings.py | 18 +- backend/app/api/v1/visualization.py | 153 ++++- backend/app/core/data_sources.py | 4 +- backend/app/core/data_sources.yaml | 4 + backend/app/core/datasource_defaults.py | 12 + backend/app/models/__init__.py | 7 +- backend/app/models/vessel.py | 112 +++- backend/app/services/collectors/__init__.py | 38 ++ backend/app/services/collectors/aisstream.py | 276 +++++++++ backend/app/services/collectors/vessel_ais.py | 63 +- backend/app/services/credential_guides.py | 59 ++ .../app/services/datasource_connectivity.py | 57 +- .../app/services/vessel_ais_aggregation.py | 574 ++++++++++++++++++ backend/app/services/vessel_types.py | 31 + backend/tests/test_collectors.py | 35 +- backend/tests/test_vessels.py | 334 +++++++++- docs/CHANGELOG.md | 17 + docs/documentation-coverage-rules.md | 117 ++++ .../earth-vessel-ais-aggregation-plan.md | 150 ++++- docs/plans/earth-vessel-tracking-plan.md | 8 +- docs/technical/en/backend-collectors.md | 2 + ...asource-collector-settings-connectivity.md | 25 + docs/technical/en/earth-frontend-context.md | 9 + .../en/earth-layer-style-reference.md | 3 + docs/technical/zh/backend-collectors.md | 5 +- ...asource-collector-settings-connectivity.md | 25 + docs/technical/zh/earth-frontend-context.md | 8 +- .../zh/earth-layer-style-reference.md | 3 + docs/version-history.md | 3 +- frontend/package.json | 2 +- frontend/public/earth/js/constants.js | 2 +- frontend/public/earth/js/main.js | 9 +- frontend/public/earth/js/vessels.js | 26 +- .../src/pages/DataSources/DataSources.tsx | 11 + frontend/src/pages/Settings/Settings.tsx | 263 +++++++- pyproject.toml | 3 +- uv.lock | 4 +- 42 files changed, 2501 insertions(+), 454 deletions(-) create mode 100644 backend/app/services/collectors/aisstream.py create mode 100644 backend/app/services/vessel_ais_aggregation.py create mode 100644 backend/app/services/vessel_types.py create mode 100644 docs/documentation-coverage-rules.md diff --git a/.claude/commands/docs.md b/.claude/commands/docs.md index ba3571af..e67c2b97 100644 --- a/.claude/commands/docs.md +++ b/.claude/commands/docs.md @@ -1,232 +1,93 @@ --- -description: 分析本次 git 变更，在 docs/technical/zh/ 中新建或更新对应的技术文档 -argument-hint: 可选：指定要记录的主题，或留空自动从 git diff 推断 +description: Create or update repository documentation from current code changes +argument-hint: Optional: topic to document, or leave empty to infer from git diff allowed-tools: ["Read", "Edit", "Write", "Bash", "Glob", "Grep"] --- -# /docs — 技术文档写入工作流 +# /docs — Documentation Workflow -## 目标 +## Goal -根据当前 git 变更（或用户指定主题）在 `docs/technical/zh/` 中写入或更新技术文档，记录**为什么**这样做，而不只是记录做了什么。 +Create or update documentation that explains why a change exists, how it behaves, and what maintainers need to know. Keep this command generic. Repository-specific coverage rules live in the repository and must be loaded separately. -## 执行步骤 +## Repository Rules -### Step 1 — 理解变更范围 +Before deciding scope, check whether the repository has a documentation rules file: ```bash -git diff HEAD --stat # 变更文件一览 -git diff HEAD --name-only # 变更文件列表 -git log --oneline -10 # 近期 commit 上下文 +test -f docs/documentation-coverage-rules.md && sed -n '1,240p' docs/documentation-coverage-rules.md ``` -若 `$ARGUMENTS` 指定了主题，优先聚焦该主题；否则从文件列表和 diff stat 推断变更主题。不要默认读取完整仓库 diff；只对决定文档主题所需的文件读取 focused diff： +If it exists, apply it as the project-specific coverage checklist. If it does not exist, continue with the generic workflow below. + +## Workflow + +### Step 1 — Understand The Change + +```bash +git diff HEAD --stat +git diff HEAD --name-only +git log --oneline -10 +rg --files docs +``` + +If `$ARGUMENTS` specifies a topic, focus on that topic. Otherwise infer the documentation topic from the changed files. Do not read the full repository diff by default; inspect focused files only: ```bash git diff HEAD -- rg -n "class |def |function |export |router|@router|interface |type " ``` -### Step 2 — 确认文档范围 +### Step 2 — Decide Scope -分析变更，判断： +- Prefer updating an existing relevant document over creating a duplicate. +- Use one document for one coherent topic. +- Split documents only when the change crosses meaningful domains. +- Keep filenames lowercase and hyphenated. +- Apply the repository-specific rules file before writing. -1. **应写几篇文档**：单一主题写一篇，跨领域变更可拆分（如后端性能优化 + 运维启动脚本分开写） -2. **是新建还是更新**：检查 `docs/technical/zh/` 中是否已有相关文档 -3. **文档命名**：按 `领域-主题-副题.md` 格式，全小写，用连字符，如： - - `backend-datasources-api-performance.md` - - `ops-planet-sh-startup.md` - - `earth-bgp-context.md` +For ambiguous or large documentation changes, briefly state the intended doc plan before editing. For clear small changes, proceed directly. -```bash -ls docs/technical/zh/ # 查看现有文档 +### Step 3 — Write + +Explain: + +- Background/problem: what was wrong or missing before. +- Core design decisions and rationale. +- Operational or user-facing impact. +- Relevant code paths, only when useful for future maintainers. + +Style: + +- Follow the repository’s existing language and heading conventions. +- Use fenced code blocks with language tags. +- Prefer tables for comparisons or parameter lists. +- Keep snippets concise and relevant. + +### Step 4 — Verify + +- Read the completed docs once for clarity and stale statements. +- Verify referenced paths exist with `test -e` or `rg --files`. +- Run applicable checks from `docs/documentation-coverage-rules.md`. +- Check Markdown links use readable user-facing titles unless repository rules allow otherwise. + +### Step 5 — Report + +Summarize changed docs and verification: + +```md +Updated: +- path/to/doc.md — what changed + +Verified: +- checks that passed +- checks that could not be run, if any ``` -**先输出写作计划供用户确认**（若变更明确且范围小，可直接执行）： +## Hard Constraints -``` -文档计划： - 新建：docs/technical/zh/ops-planet-sh-startup.md — planet.sh 启动性能优化 - 更新：docs/technical/zh/backend-datasources-api-performance.md — 补充并行化细节 -``` - -### Step 2.5 — 覆盖范围检查 - -写文档前必须按变更类型检查配套文档，不要只更新一篇专题文档： - -- 用户可见流程变化：更新 `docs/technical/zh/manual.md`，通常也更新 `docs/technical/zh/quickstart.md`。 -- `manual.md`、`quickstart.md` 这类用户手册存在英文版时，同步更新 `docs/technical/en/...`，至少避免英文版与中文版互相矛盾。 -- 控制台页面职责、路由入口、表格/抽屉/设置页行为变化：更新 `docs/technical/zh/frontend-admin-frontend-context.md`。 -- Earth 前端行为、HUD、巡航、图层、图例、交互变化：更新 `docs/technical/zh/earth-frontend-context.md`。 -- 新增 Earth 图层、调整 `renderOrder`、半径/高度偏移、深度策略、拾取策略、legend mode、图层面板顺序或启动加载顺序：更新 `docs/technical/zh/earth-render-layer-order.md`。 -- Earth 图层视觉样式、颜色、图例符号语义变化：若影响样式索引，同步更新 `docs/technical/zh/earth-layer-style-reference.md`。 -- 采集器、数据源、凭证、设置页、连接检查、scheduler、后端 API 变化：更新相关后端文档，优先检查 `docs/technical/zh/backend-collectors.md` 和 datasource/settings 专题文档。 -- 如果某个旧 plan 的假设已经被当前实现推翻，在对应 `docs/plans/*.md` 增加现状修正或更新该段，不要让计划文档继续给出相反方向。 -- 新增 technical 文档后，如果需要被发现，更新 `docs/technical/zh/README.md`。 -- 如果 technical 文档需要在公开 Docs 页面显示，或从 technical README 链接进入，必须同步更新 `frontend/src/pages/Docs/docs-content.ts` 的 `DOCS_METADATA`。前端使用这份白名单，`docs/technical/{zh,en}/` 中存在 `.md` 文件并不会自动生成路由。 -- 公开 technical 文档必须按同名文件维护中英文双语版本：`docs/technical/zh/.md` 与 `docs/technical/en/.md`。如果某篇文档刻意只保留单语，完成说明中必须明确写出原因。 -- 对本次变更提取旧词做 stale search，例如旧 tab 名、旧路由职责、旧认证假设、改名前 UI 文案： - -```bash -rg -n "旧文案|旧路由职责|旧认证假设" docs/technical docs/plans -``` - -### Step 3 — 写文档 - -遵循以下原则： - -**记录 WHY，不只记录 WHAT** -- 好：`将戳文件从 /tmp 移到 ~/.cache/planet/，因为 WSL 重启后 /tmp 被清空` -- 差：`修改了 AI_PROVIDER_BUILD_STAMP_FILE 的值` - -**必须包含的内容**： -- 背景/问题：改动之前存在什么问题，为什么要改 -- 核心设计决策及其理由 -- 关键代码片段（用 diff 或 before/after 展示） -- 相关文件列表 - -**格式要求**： -- 使用 `##` 和 `###` 分级，不要超过三级 -- 代码块注明语言（python / bash / typescript / sql） -- 表格用于对比多个选项或列出参数 -- 中文写作，技术术语保留英文原文 -- `docs/technical/zh/` 中的文档不得用英文原文占位；如果存在 `docs/technical/en/` 对应文件，禁止逐字复制成中文文件 -- 中文文档内部链接应指向 `docs/technical/zh/...`，除非明确引用英文专属文档 -- 公开文档的 Markdown 链接显示文字应使用可读标题，不要直接暴露 `manual.md`、`earth-frontend-context.md` 这类裸文件名 - -**文档结构模板**： - -```markdown -# 标题（说明做了什么） - -## 背景 - -为什么要做这个改动，改动前存在什么问题。 - -## 核心变更 - -### 子主题一 - -before/after 或决策说明 + 关键代码 - -### 子主题二 - -... - -## 相关文件 - -- `path/to/file.py` — 简短说明 -``` - -### Step 4 — 验证 - -- 读一遍写好的文档，确认逻辑清晰、代码片段无明显错误 -- 用 `rg --files` 或 `test -e` 确认文档中的文件路径在项目中真实存在，避免凭记忆判断： -- 检查中文文档没有误复制英文版： - -```bash -python - <<'PY' -from pathlib import Path -same = [] -for en in sorted(Path("docs/technical/en").glob("*.md")): - zh = Path("docs/technical/zh") / en.name - if zh.exists() and en.read_text() == zh.read_text(): - same.append(en.name) -if same: - raise SystemExit("identical en/zh docs: " + ", ".join(same)) -print("no identical en/zh docs") -PY -``` - -- 检查中文文档内部链接没有继续指向无语言目录： - -```bash -rg -n "/home/ray/dev/linkong/planet/docs/technical/(?!zh|en)" docs/technical/zh --pcre2 -``` - -- 检查公开文档链接已进入 Docs 前端白名单。凡是 `docs/technical/{zh,en}/README.md` 中链接到的 technical `.md`，都必须存在于 `DOCS_METADATA`： - -```bash -python - <<'PY' -import re -from pathlib import Path - -metadata = Path("frontend/src/pages/Docs/docs-content.ts").read_text() -known = set(re.findall(r"'([^']+\.md)':\s*\{", metadata)) -known.add("README.md") - -missing = [] -for readme in [Path("docs/technical/zh/README.md"), Path("docs/technical/en/README.md")]: - if not readme.exists(): - continue - for href in re.findall(r"\]\(([^)]+\.md)\)", readme.read_text()): - path = Path(href) - if "docs/technical/" not in href: - continue - filename = path.name - if filename not in known: - missing.append(f"{readme}: {filename}") - -if missing: - raise SystemExit("docs README links missing DOCS_METADATA: " + ", ".join(missing)) -print("docs README links are whitelisted") -PY -``` - -- 检查公开文档双语同名文件齐备。除 `README.md` 外，所有白名单文档都应同时存在 zh/en 文件，除非本次说明中明确豁免： - -```bash -python - <<'PY' -import re -from pathlib import Path - -metadata = Path("frontend/src/pages/Docs/docs-content.ts").read_text() -filenames = sorted(set(re.findall(r"'([^']+\.md)':\s*\{", metadata)) - {"README.md"}) -missing = [] -for filename in filenames: - for lang in ("zh", "en"): - path = Path("docs/technical") / lang / filename - if not path.exists(): - missing.append(str(path)) -if missing: - raise SystemExit("missing bilingual docs: " + ", ".join(missing)) -print("public docs have zh/en file pairs") -PY -``` - -- 检查公开文档里没有用裸 `.md` 文件名当链接标题。这个命令在 polished public docs 中应无输出： - -```bash -rg -n "\[[^]]+\.md\]\(" docs/technical/zh docs/technical/en -``` - -```bash -# 对文档中提到的关键路径做快速验证 -ls -``` - -如需检查大量链接，优先用确定性提取： - -```bash -rg -n "\]\(([^)]+)\)" docs/technical/zh/.md -``` - -### Step 5 — 完成确认 - -输出摘要： - -``` -✓ 新建：docs/technical/zh/ops-planet-sh-startup.md（约 xxx 字） -✓ 更新：docs/technical/zh/backend-datasources-api-performance.md -``` - -## 注意事项 - -- 不要写流水账式的"改了 A、改了 B、改了 C"，要写改动背后的约束和权衡 -- 不要在文档中引用 PR 号、issue 号、或当前对话——这些会随时间失效 -- 代码片段保持简洁，只保留说明问题的关键部分，省略无关样板代码 -- 如果某个变更已有文档记录，优先在原文档中追加，而不是新建 -- 公开 technical 文档没有注册 `DOCS_METADATA` 时，Docs 页面不会显示；不要只创建 `.md` 文件就结束。 -- 公开 technical 文档默认需要 zh/en 同名文件，不要只补一个语言版本。 -- 链接可见文字使用文档标题或语义标题，不要使用裸文件名。 -- 文档是给未来的开发者看的，假设读者熟悉项目但不了解这次改动的背景 +- Do not leave placeholder docs. +- Do not duplicate bilingual files byte-for-byte. +- Do not reference PR numbers, issue numbers, or the current conversation unless explicitly requested. +- Do not write changelog-style lists without the reasoning and tradeoffs behind the change. +- Keep docs maintainable and concise. diff --git a/.codex/skills/docs/SKILL.md b/.codex/skills/docs/SKILL.md index 93580183..9c399c85 100644 --- a/.codex/skills/docs/SKILL.md +++ b/.codex/skills/docs/SKILL.md @@ -1,187 +1,72 @@ --- name: docs -description: Analyze current Planet repo changes and create or update technical documentation under docs/technical/zh. Use when the user asks to write docs, update technical docs, summarize implementation changes into documentation, or port the Claude docs-codex workflow into Codex. +description: Create or update repository documentation from current code changes. Use when the user asks to write docs, update docs, summarize implementation changes into docs, or check documentation coverage. Load repository-specific coverage rules from docs/documentation-coverage-rules.md when present. --- # Docs -Use this skill when the user asks to create or update Planet technical documentation, especially under `docs/technical/zh/`. +Use this skill when the task is documentation work: creating, updating, checking, or summarizing docs for code or behavior changes. ## Goal -Write or update technical docs that explain why a change exists, not only what files changed. +Write documentation that explains why a change exists, how it behaves, and what maintainers need to know. Keep the skill generic; repository-specific rules belong in the repository, not in this skill. -Default target directory: +## Repository Rules -- `docs/technical/zh/` +Before deciding scope, check whether the repository has a documentation rules file: + +```bash +test -f docs/documentation-coverage-rules.md && sed -n '1,240p' docs/documentation-coverage-rules.md +``` + +If it exists, apply it as the project-specific coverage checklist. If it does not exist, continue with the generic workflow below. ## Workflow -1. Gather change context: +1. Gather focused context: ```bash git diff HEAD --stat git diff HEAD --name-only git log --oneline -10 -ls docs/technical/zh/ +rg --files docs ``` -If the user gives a specific topic, focus on that topic. Otherwise infer the documentation topic from the file list and diff stat. Do **not** read the full repository diff by default; inspect focused diffs only for the files that define the doc topic: +If the user gives a topic, focus on that topic. Otherwise infer the doc topic from changed files. Avoid reading large full diffs by default; inspect focused files and symbols: ```bash git diff HEAD -- rg -n "class |def |function |export |router|@router|interface |type " ``` -2. Decide document scope: +2. Decide scope: -- Use one document for one coherent topic. -- Split documents when the changes cross meaningful domains, such as backend performance and ops startup behavior. - Prefer updating an existing relevant doc over creating a duplicate. -- Name new files as lowercase hyphenated `domain-topic-detail.md`, for example: - - `backend-datasources-api-performance.md` - - `ops-planet-sh-startup.md` - - `earth-bgp-context.md` +- Use one document for one coherent topic. +- Split documents only when changes cross meaningful domains. +- Keep filenames lowercase and hyphenated. -3. Apply the documentation coverage checklist before writing: +3. Write the doc: -- User-visible workflow changes must update `docs/technical/zh/manual.md` and usually `docs/technical/zh/quickstart.md`. -- If an English counterpart exists for user-facing docs such as `manual.md` or `quickstart.md`, update `docs/technical/en/...` enough that it does not contradict the Chinese source. -- Control console page responsibility changes must update `docs/technical/zh/frontend-admin-frontend-context.md`. -- Earth frontend behavior changes must update `docs/technical/zh/earth-frontend-context.md`. -- Earth layer additions, `renderOrder`, altitude/radius offsets, depth strategy, pointer picking, legend modes, or layer panel/startup ordering must update `docs/technical/zh/earth-render-layer-order.md`. -- Earth layer visual style or legend symbol/color semantics should also update `docs/technical/zh/earth-layer-style-reference.md` when that reference is affected. -- Collector, datasource, credential, settings, connectivity, scheduler, or API changes must update the relevant backend docs, especially `docs/technical/zh/backend-collectors.md` and any datasource/settings-specific doc. -- When a change turns an old plan assumption into current behavior, update the relevant `docs/plans/*.md` with a status note instead of leaving contradictory instructions. -- If adding a new technical document, add it to `docs/technical/zh/README.md` when it should be discoverable from the technical docs index. -- If a technical document should be visible in the public Docs page or linked from a technical README, register it in `frontend/src/pages/Docs/docs-content.ts` under `DOCS_METADATA`. The frontend uses this whitelist; files under `docs/technical/{zh,en}/` are not automatically routable. -- For every public technical doc, keep the bilingual file pair in sync by filename: `docs/technical/zh/.md` and `docs/technical/en/.md`. If the content is intentionally Chinese-only or English-only, state that intentionally in the final note. -- Search docs for stale terms introduced by the change, for example old tab names, old route responsibilities, obsolete auth assumptions, or renamed UI labels. +- Explain background/problem, design decisions, constraints, and operational impact. +- Keep code snippets short and directly relevant. +- List related files only when they help future maintainers navigate. +- Use the repository’s existing language, heading style, and naming conventions. -4. Write the doc in Chinese: +4. Verify: -- Write Chinese prose for `docs/technical/zh/`. -- Keep technical identifiers, API paths, config keys, code symbols, and standard product names in English where appropriate. -- Use `##` and `###` headings; avoid going deeper than three levels. -- Use fenced code blocks with language tags. -- Use tables when comparing options or listing parameters. - -5. Required content: - -- Background/problem: what was wrong before and why the change was needed. -- Core design decisions and rationale. -- Key code snippets, preferably before/after or focused excerpts. -- Related files and what each file contributes. - -6. Verification: - -- Read the completed doc and check that the reasoning is clear. -- Verify important referenced paths exist. -- Use `rg --files` or `test -e` for path existence instead of relying on memory. -- Run a quick duplicate-language check when editing bilingual docs: - -```bash -python - <<'PY' -from pathlib import Path -same = [] -for en in sorted(Path("docs/technical/en").glob("*.md")): - zh = Path("docs/technical/zh") / en.name - if zh.exists() and en.read_text() == zh.read_text(): - same.append(en.name) -if same: - raise SystemExit("identical en/zh docs: " + ", ".join(same)) -print("no identical en/zh docs") -PY -``` - -Also check that Chinese docs do not link to the old language-less technical docs path: - -```bash -rg -n "/home/ray/dev/linkong/planet/docs/technical/(?!zh|en)" docs/technical/zh --pcre2 -``` - -This command should return no matches. - -Check that public docs are whitelisted in the frontend Docs registry. Any `.md` linked from `docs/technical/{zh,en}/README.md` and located under `docs/technical/{zh,en}/` must have a matching `DOCS_METADATA` key: - -```bash -python - <<'PY' -import re -from pathlib import Path - -metadata = Path("frontend/src/pages/Docs/docs-content.ts").read_text() -known = set(re.findall(r"'([^']+\.md)':\s*\{", metadata)) -known.add("README.md") - -missing = [] -for readme in [Path("docs/technical/zh/README.md"), Path("docs/technical/en/README.md")]: - if not readme.exists(): - continue - for href in re.findall(r"\]\(([^)]+\.md)\)", readme.read_text()): - path = Path(href) - if "docs/technical/" not in href: - continue - filename = path.name - if filename not in known: - missing.append(f"{readme}: {filename}") - -if missing: - raise SystemExit("docs README links missing DOCS_METADATA: " + ", ".join(missing)) -print("docs README links are whitelisted") -PY -``` - -Check bilingual parity for public docs. Every whitelisted document except `README.md` should exist in both language directories unless intentionally documented otherwise: - -```bash -python - <<'PY' -import re -from pathlib import Path - -metadata = Path("frontend/src/pages/Docs/docs-content.ts").read_text() -filenames = sorted(set(re.findall(r"'([^']+\.md)':\s*\{", metadata)) - {"README.md"}) -missing = [] -for filename in filenames: - for lang in ("zh", "en"): - path = Path("docs/technical") / lang / filename - if not path.exists(): - missing.append(str(path)) -if missing: - raise SystemExit("missing bilingual docs: " + ", ".join(missing)) -print("public docs have zh/en file pairs") -PY -``` - -Check that Markdown links do not expose raw filenames as user-facing titles. This should return no matches for polished public docs: - -```bash -rg -n "\[[^]]+\.md\]\(" docs/technical/zh docs/technical/en -``` - -If checking many links, prefer deterministic extraction: - -```bash -rg -n "\]\(([^)]+)\)" docs/technical/zh/.md -``` - -Also run focused stale-term searches derived from the change, for example: - -```bash -rg -n "old label|old route purpose|obsolete provider assumption" docs/technical docs/plans -``` +- Read the completed doc once for clarity and stale statements. +- Verify important referenced paths exist with `test -e` or `rg --files`. +- Run repository-specific doc checks from `docs/documentation-coverage-rules.md` when present. +- For Markdown links, check that user-facing titles are readable and not raw filenames unless the repository rules allow it. ## Hard Constraints -- A file under `docs/technical/zh/` must not be an English source file copied as a placeholder. -- Do not leave a Chinese doc with only an English title and English first-screen content. -- When an English counterpart exists in `docs/technical/en/`, never duplicate it byte-for-byte into `docs/technical/zh/`. -- Internal links inside `docs/technical/zh/` should point to `docs/technical/zh/...` for Chinese docs, unless intentionally linking to an English-only file. -- Public technical documents must be registered in `frontend/src/pages/Docs/docs-content.ts` before considering them available in the Docs UI. -- Public technical documents should have both zh and en files with the same filename, unless intentionally exempted. -- Markdown link text in public docs should be a readable title, not a raw filename such as `manual.md`. -- Do not reference PR numbers, issue numbers, or the current conversation. -- Do not write changelog-style lists like "changed A, changed B, changed C" without the constraints and tradeoffs behind those changes. -- Keep code snippets concise and relevant. +- Do not leave placeholder docs or copied source text pretending to be documentation. +- Do not duplicate bilingual files byte-for-byte. +- Do not reference PR numbers, issue numbers, or the current conversation unless explicitly requested. +- Do not write changelog-style lists without the reasoning, constraints, and tradeoffs behind the change. +- Keep docs concise enough to maintain. ## Recommended Output @@ -189,12 +74,9 @@ After editing, summarize: ```md Updated: -- docs/technical/zh/example.md — what changed +- path/to/doc.md — what changed Verified: -- no identical en/zh docs -- no language-less docs/technical links in zh docs -- public docs are registered in DOCS_METADATA -- public docs have zh/en file pairs -- no raw `.md` filenames as public link titles +- checks that passed +- checks that could not be run, if any ``` diff --git a/TODO.md b/TODO.md index 5abe1c7f..ab1e8fd1 100644 --- a/TODO.md +++ b/TODO.md @@ -26,6 +26,8 @@ - [ ] 重写控制台 UI，逐步抛弃 Ant Design，建立自有组件体系，并统一采用 `tabler.io` / Tabler Icons 作为控制台主图标库 - [ ] 把 Earth 态势新闻源从 [earth_news.py](/home/ray/dev/linkong/planet/backend/app/services/earth_news.py) 的硬编码列表抽成可配置目录，优先保持当前“实时聚合”链路不变，只先解决新闻源不可配置的问题 - [ ] 为 Earth 态势新闻设计后续采集器化方案：明确新闻数据模型、去重策略、区域映射、过期清理和 Earth/AI 复用方式，再决定何时把新闻从实时抓取升级成正式 collector +- [ ] AIS v4：开放船只多源聚合策略配置，支持 source priority、字段级规则、freshness 窗口和高级保护开关；保存时校验未知字段、非法模式和危险动态字段锁定，并在聚合接口返回命中的配置版本 +- [ ] AIS v5：实现船舶资料 enrichment 与冲突治理，按 `mmsi + imo + name + callsign` 异步补充船型细分、AIS 大类、旗国、尺寸、建造年份、运营方和图片缓存；详情面板展示缓存资料和字段来源，不在实时 AIS 请求链路现场抓第三方页面 - [ ] 为 Earth 地球表面增加一层与基础纹理对齐的材质/纹理 overlay，并在同层叠加国界轮廓参考线；要求国界线与底图稳定对齐，且 hover 到国家轮廓时能高亮当前国家，便于校准地表和增强交互 - [ ] 把 Earth 新闻接入通用巡航队列：按新闻发生地和时间排序生成巡航目标，巡航聚焦到新闻事件时显示对应新闻卡片，并保持实现边界为“通用巡航层 + 新闻业务适配层”，不要再把新闻逻辑直接耦合回 `main.js` 状态机 - [ ] 为未知位置的算力中心建立分层坐标补全链路：优先 `精确坐标 > 站点/园区命中 > 城市 > 州/省 > 国家内主要算力城市 > 国家质心`，并把每次回退的 `confidence / reason / precision` 明确写进统一 GeoJSON diff --git a/VERSION b/VERSION index c063aeab..421ab545 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.46.3 +0.47.0 diff --git a/backend/app/api/v1/datasource_config.py b/backend/app/api/v1/datasource_config.py index bb3d3303..81b15075 100644 --- a/backend/app/api/v1/datasource_config.py +++ b/backend/app/api/v1/datasource_config.py @@ -5,7 +5,7 @@ from datetime import datetime import base64 import json import re -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from pydantic import BaseModel, Field @@ -318,7 +318,7 @@ async def list_configs( """List all user-defined data source configurations""" query = select(DataSourceConfig) if active_only: - query = query.where(DataSourceConfig.is_active == True) + query = query.where(DataSourceConfig.is_active) query = query.order_by(DataSourceConfig.created_at.desc()) result = await db.execute(query) @@ -374,6 +374,11 @@ async def list_all_datasources( "is_active": db_config.is_active if db_config else True, "source_type": db_config.source_type if db_config else "http", "auth_type": db_config.auth_type if db_config else "none", + "auth_configured": { + "api_key": bool((db_config.auth_config or {}).get("api_key")) + if db_config + else False, + }, "headers": db_config.headers if db_config else {}, "config": strip_connectivity_validation(db_config.config if db_config else {}), "config_id": db_config.id if db_config else None, @@ -464,6 +469,8 @@ async def update_config( for field, value in update_data.items(): if field == "config": value = strip_connectivity_validation(value) + if field == "auth_config" and value == {} and (config.auth_config or {}): + continue setattr(config, field, value) await db.commit() @@ -601,6 +608,7 @@ async def connect_builtin_config( config_data.headers, config_data.config, db, + config_data.auth_config, ) if result.get("success") and result.get("checksum"): validation = await save_connectivity_success( diff --git a/backend/app/api/v1/settings.py b/backend/app/api/v1/settings.py index 5ebaa723..700a6312 100644 --- a/backend/app/api/v1/settings.py +++ b/backend/app/api/v1/settings.py @@ -17,6 +17,7 @@ from app.models.datasource import DataSource from app.models.datasource_config import DataSourceConfig from app.models.system_setting import SystemSetting from app.models.user import User +from app.models.vessel import AISSourceHealth from app.services.barentswatch import ( BarentsWatchConfig, check_barentswatch_config, @@ -368,7 +369,12 @@ def format_frequency_label(minutes: int) -> str: return f"{minutes}m" -def serialize_collector(datasource: DataSource) -> dict: +async def get_ais_source_health_by_source(db: AsyncSession) -> dict[str, dict]: + result = await db.execute(select(AISSourceHealth)) + return {item.source: item.to_dict() for item in result.scalars().all()} + + +def serialize_collector(datasource: DataSource, ais_health_by_source: dict[str, dict] | None = None) -> dict: defaults = DEFAULT_DATASOURCES.get(datasource.source, {}) return { "id": datasource.id, @@ -387,6 +393,7 @@ def serialize_collector(datasource: DataSource) -> dict: "requires_credentials": bool(defaults.get("requires_credentials", False)), "credential_provider": defaults.get("credential_provider"), "credential_status": defaults.get("credential_status", "none"), + "ais_health": (ais_health_by_source or {}).get(datasource.source), } @@ -599,7 +606,8 @@ async def get_collector_settings( ): result = await db.execute(select(DataSource).order_by(DataSource.module, DataSource.id)) datasources = result.scalars().all() - return {"collectors": [serialize_collector(datasource) for datasource in datasources]} + ais_health_by_source = await get_ais_source_health_by_source(db) + return {"collectors": [serialize_collector(datasource, ais_health_by_source) for datasource in datasources]} @router.put("/collectors/{datasource_id}") @@ -619,7 +627,8 @@ async def update_collector_settings( await db.commit() await db.refresh(datasource) await sync_datasource_job(datasource.id) - return {"status": "updated", "collector": serialize_collector(datasource)} + ais_health_by_source = await get_ais_source_health_by_source(db) + return {"status": "updated", "collector": serialize_collector(datasource, ais_health_by_source)} @router.get("") @@ -633,12 +642,13 @@ async def get_all_settings( db, ["system", "notifications", "security"], ) + ais_health_by_source = await get_ais_source_health_by_source(db) return { "system": setting_payloads["system"], "notifications": setting_payloads["notifications"], "security": setting_payloads["security"], "tv": await get_tv_settings_payload(db), "integrations": await serialize_external_integrations(db), - "collectors": [serialize_collector(datasource) for datasource in datasources], + "collectors": [serialize_collector(datasource, ais_health_by_source) for datasource in datasources], "generated_at": to_iso8601_utc(datetime.now(UTC)), } diff --git a/backend/app/api/v1/visualization.py b/backend/app/api/v1/visualization.py index e82bff5d..e551be94 100644 --- a/backend/app/api/v1/visualization.py +++ b/backend/app/api/v1/visualization.py @@ -25,6 +25,14 @@ from app.services.bgp_collectors import build_bgp_collector_coverage from app.services.cable_graph import build_graph_from_data, CableGraph, haversine_distance from app.services.collectors.bgp_common import RIPE_RIS_COLLECTOR_COORDS from app.services.persistent_logs import record_system_log +from app.services.vessel_ais_aggregation import ( + build_field_conflict_candidates, + get_aggregated_vessel, + get_aggregated_vessel_track, + get_aggregated_vessels, + get_vessel_conflict_records, + get_vessel_raw_observations, +) from app.core.logging import get_logger router = APIRouter() @@ -664,6 +672,54 @@ def convert_vessels_to_geojson(rows: List[Any]) -> Dict[str, Any]: return {"type": "FeatureCollection", "features": features} +def convert_aggregated_vessels_to_geojson(vessels: List[dict[str, Any]]) -> Dict[str, Any]: + features = [] + for vessel in vessels: + if vessel.get("lat") is None or vessel.get("lon") is None: + continue + source_summary = {} + for source, summary in (vessel.get("source_summary") or {}).items(): + source_summary[source] = { + **summary, + "latest_observed_at": to_iso8601_utc(summary.get("latest_observed_at")), + } + props = { + "mmsi": vessel["mmsi"], + "name": vessel.get("name") or f"MMSI {vessel['mmsi']}", + "callsign": vessel.get("callsign"), + "imo": vessel.get("imo"), + "vessel_type": vessel.get("vessel_type"), + "vessel_type_name": vessel.get("vessel_type_name") or "Other", + "flag": vessel.get("flag"), + "length": vessel.get("length"), + "width": vessel.get("width"), + "draught": vessel.get("draught"), + "sog": vessel.get("sog"), + "cog": vessel.get("cog"), + "heading": vessel.get("heading"), + "nav_status": vessel.get("nav_status"), + "received_at": to_iso8601_utc(vessel.get("received_at")), + "field_sources": vessel.get("field_sources") or {}, + "selected_reasons": vessel.get("selected_reasons") or {}, + "source_summary": source_summary, + "quality_flags": vessel.get("quality_flags") or [], + "conflict_count": vessel.get("conflict_count", 0), + "data_type": "vessel", + } + features.append( + { + "type": "Feature", + "id": vessel["mmsi"], + "geometry": { + "type": "Point", + "coordinates": [vessel["lon"], vessel["lat"]], + }, + "properties": props, + } + ) + return {"type": "FeatureCollection", "features": features} + + def _parse_bbox(value: Optional[str]) -> tuple[float, float, float, float] | None: if not value: return None @@ -1411,10 +1467,37 @@ async def get_vessels_geojson( None, description="Comma-separated vessel types: cargo,tanker,passenger,fishing,military,other", ), - limit: int = Query(5000, ge=1, le=50000), + limit: Optional[int] = Query( + None, + ge=0, + description="Maximum vessel features to return. Omit or pass 0 for no limit.", + ), db: AsyncSession = Depends(get_db), ): """Return latest vessel positions as GeoJSON points.""" + parsed_bbox = _parse_bbox(bbox) + aggregated_vessels = await get_aggregated_vessels(db, bbox=parsed_bbox, limit=limit) + if aggregated_vessels: + geojson = convert_aggregated_vessels_to_geojson(aggregated_vessels) + requested_types = { + item.strip().lower() + for item in (type or "").split(",") + if item.strip() + } + if requested_types: + geojson["features"] = [ + feature + for feature in geojson.get("features", []) + if _matches_vessel_type(feature.get("properties", {}), requested_types) + ] + + features = geojson.get("features", []) + return { + **geojson, + "count": len(features), + "stats": _build_vessel_stats(features), + } + latest_times = ( select( VesselPosition.mmsi.label("mmsi"), @@ -1432,10 +1515,10 @@ async def get_vessels_geojson( ) .outerjoin(VesselStatic, VesselStatic.mmsi == VesselPosition.mmsi) .order_by(VesselPosition.received_at.desc()) - .limit(limit) ) + if limit and limit > 0: + stmt = stmt.limit(limit) - parsed_bbox = _parse_bbox(bbox) if parsed_bbox is not None: lon_min, lat_min, lon_max, lat_max = parsed_bbox stmt = stmt.where( @@ -1470,6 +1553,15 @@ async def get_vessels_geojson( @router.get("/vessels/{mmsi}") async def get_vessel_detail(mmsi: int, db: AsyncSession = Depends(get_db)): + aggregated = await get_aggregated_vessel(db, mmsi) + if aggregated is not None: + return { + **aggregated, + "received_at": to_iso8601_utc(aggregated.get("received_at")), + "latitude": aggregated["lat"], + "longitude": aggregated["lon"], + } + latest_position_stmt = ( select(VesselPosition) .where(VesselPosition.mmsi == mmsi) @@ -1496,6 +1588,30 @@ async def get_vessel_track( db: AsyncSession = Depends(get_db), ): cutoff = datetime.now(UTC) - timedelta(hours=hours) + aggregated_points = await get_aggregated_vessel_track(db, mmsi, cutoff=cutoff) + if aggregated_points: + return { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "geometry": { + "type": "LineString", + "coordinates": [[point["lon"], point["lat"]] for point in aggregated_points], + }, + "properties": { + "mmsi": mmsi, + "hours": hours, + "point_count": len(aggregated_points), + "start_at": to_iso8601_utc(aggregated_points[0]["observed_at"]), + "end_at": to_iso8601_utc(aggregated_points[-1]["observed_at"]), + "point_sources": [point["source"] for point in aggregated_points], + }, + } + ], + "count": 1, + } + result = await db.execute( select(VesselPosition) .where(VesselPosition.mmsi == mmsi) @@ -1532,6 +1648,37 @@ async def get_vessel_track( } +@router.get("/vessels/{mmsi}/observations") +async def get_vessel_observations( + mmsi: int, + limit: int = Query(100, ge=1, le=500), + db: AsyncSession = Depends(get_db), +): + """Return raw AIS observations for debugging source-level collector facts.""" + + observations = await get_vessel_raw_observations(db, mmsi, limit=limit) + return { + "mmsi": mmsi, + "count": len(observations), + "observations": [item.to_dict() for item in observations], + "conflict_candidates": build_field_conflict_candidates(observations), + } + + +@router.get("/vessels/{mmsi}/conflicts") +async def get_vessel_conflicts(mmsi: int, db: AsyncSession = Depends(get_db)): + """Return recorded AIS conflicts plus current raw-observation candidates.""" + + records = await get_vessel_conflict_records(db, mmsi) + observations = await get_vessel_raw_observations(db, mmsi, limit=500) + return { + "mmsi": mmsi, + "count": len(records), + "conflicts": [item.to_dict() for item in records], + "candidates": build_field_conflict_candidates(observations), + } + + @router.get("/geo/bgp-anomalies") async def get_bgp_anomalies_geojson( severity: Optional[str] = Query(None), diff --git a/backend/app/core/data_sources.py b/backend/app/core/data_sources.py index 9df98ef3..c3b746bb 100644 --- a/backend/app/core/data_sources.py +++ b/backend/app/core/data_sources.py @@ -1,7 +1,6 @@ import os import yaml from functools import lru_cache -from typing import Optional COLLECTOR_URL_KEYS = { @@ -32,6 +31,7 @@ COLLECTOR_URL_KEYS = { "nro_delegated_prefix_geo": "nro.delegated_stats_url", "news_live_streams": "news_live_streams.channels_url", "barentswatch_vessels": "barentswatch_vessels.url", + "aisstream_vessels": "aisstream_vessels.url", } @@ -74,7 +74,7 @@ class DataSourcesConfig: from app.models.datasource_config import DataSourceConfig query = select(DataSourceConfig).where( - DataSourceConfig.name == collector_name, DataSourceConfig.is_active == True + DataSourceConfig.name == collector_name, DataSourceConfig.is_active ) result = await db.execute(query) db_config = result.scalar_one_or_none() diff --git a/backend/app/core/data_sources.yaml b/backend/app/core/data_sources.yaml index b1627e2a..513a533c 100644 --- a/backend/app/core/data_sources.yaml +++ b/backend/app/core/data_sources.yaml @@ -98,3 +98,7 @@ news_live_streams: barentswatch_vessels: # BarentsWatch Live AIS latest combined endpoint. Requires an AIS bearer token. url: "https://live.ais.barentswatch.no/v1/latest/combined" + +aisstream_vessels: + # AISStream realtime WebSocket endpoint. Requires an AISStream API key. + url: "wss://stream.aisstream.io/v0/stream" diff --git a/backend/app/core/datasource_defaults.py b/backend/app/core/datasource_defaults.py index ee7a70b0..be5adbf0 100644 --- a/backend/app/core/datasource_defaults.py +++ b/backend/app/core/datasource_defaults.py @@ -245,6 +245,18 @@ DEFAULT_DATASOURCES = { "credential_provider": "barentswatch", "credential_status": "supported", }, + "aisstream_vessels": { + "id": 28, + "name": "AISStream Vessels", + "display_name": "AISStream 实时船舶", + "module": "L4", + "priority": "P1", + "frequency_minutes": 1, + "is_free": True, + "requires_credentials": True, + "credential_provider": "aisstream", + "credential_status": "supported", + }, } ID_TO_COLLECTOR = {info["id"]: name for name, info in DEFAULT_DATASOURCES.items()} diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 45f15f8f..9ded1306 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -12,7 +12,7 @@ from app.models.system_setting import SystemSetting from app.models.playground_session import PlaygroundSession from app.models.playground_message import PlaygroundMessage from app.models.system_log import SystemLog, AuditLog -from app.models.vessel import VesselPosition, VesselStatic +from app.models.vessel import AISConflictRecord, AISRawObservation, AISSourceHealth, VesselPosition, VesselStatic from app.models.datasource_mapping import DataSourceMappingTemplate __all__ = [ @@ -31,7 +31,12 @@ __all__ = [ "BGPObservation", "SystemLog", "AuditLog", + "PlaygroundSession", + "PlaygroundMessage", "VesselPosition", "VesselStatic", + "AISRawObservation", + "AISConflictRecord", + "AISSourceHealth", "DataSourceMappingTemplate", ] diff --git a/backend/app/models/vessel.py b/backend/app/models/vessel.py index 665b4d0d..c8436c1e 100644 --- a/backend/app/models/vessel.py +++ b/backend/app/models/vessel.py @@ -1,6 +1,6 @@ """Vessel AIS models for live maritime tracking.""" -from sqlalchemy import BigInteger, Column, DateTime, Float, Index, Integer, SmallInteger, String +from sqlalchemy import BigInteger, Column, DateTime, Float, Index, Integer, JSON, SmallInteger, String from sqlalchemy.sql import func from app.core.time import to_iso8601_utc @@ -73,3 +73,113 @@ class VesselPosition(Base): "nav_status": self.nav_status, "received_at": to_iso8601_utc(self.received_at), } + + +class AISRawObservation(Base): + """Source-level AIS fact before aggregation and conflict resolution.""" + + __tablename__ = "ais_raw_observations" + + id = Column(Integer, primary_key=True, autoincrement=True) + target_schema = Column(String(64), nullable=False, default="vessel_ais", index=True) + source = Column(String(100), nullable=False, index=True) + entity_key = Column(String(64), nullable=False, index=True) + delivery_mode = Column(String(32), nullable=False, index=True) + transport = Column(String(32), nullable=False, index=True) + message_type = Column(String(64), nullable=True, index=True) + source_message_id = Column(String(128), nullable=True, index=True) + observation_hash = Column(String(64), nullable=False, unique=True, index=True) + observed_at = Column(DateTime(timezone=True), nullable=False, index=True) + collected_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), index=True) + normalized_payload = Column(JSON, default=dict) + raw_payload = Column(JSON, default=dict) + quality_flags = Column(JSON, default=list) + + __table_args__ = ( + Index("idx_ais_raw_entity_observed", "target_schema", "entity_key", "observed_at"), + Index("idx_ais_raw_source_entity", "source", "entity_key"), + ) + + def to_dict(self) -> dict: + return { + "id": self.id, + "target_schema": self.target_schema, + "source": self.source, + "entity_key": self.entity_key, + "delivery_mode": self.delivery_mode, + "transport": self.transport, + "message_type": self.message_type, + "source_message_id": self.source_message_id, + "observation_hash": self.observation_hash, + "observed_at": to_iso8601_utc(self.observed_at), + "collected_at": to_iso8601_utc(self.collected_at), + "normalized_payload": self.normalized_payload or {}, + "raw_payload": self.raw_payload or {}, + "quality_flags": self.quality_flags or [], + } + + +class AISConflictRecord(Base): + """Recorded field-level disagreement between AIS sources.""" + + __tablename__ = "ais_conflict_records" + + id = Column(Integer, primary_key=True, autoincrement=True) + target_schema = Column(String(64), nullable=False, default="vessel_ais", index=True) + entity_key = Column(String(64), nullable=False, index=True) + field = Column(String(64), nullable=False, index=True) + candidates = Column(JSON, default=dict) + selected_source = Column(String(100), nullable=True, index=True) + selected_value = Column(JSON, nullable=True) + selected_reason = Column(String(64), nullable=True, index=True) + resolved_by = Column(String(32), nullable=False, default="system", index=True) + status = Column(String(32), nullable=False, default="open", index=True) + created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), index=True) + updated_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now()) + + __table_args__ = ( + Index("idx_ais_conflict_entity_field", "target_schema", "entity_key", "field"), + ) + + def to_dict(self) -> dict: + return { + "id": self.id, + "target_schema": self.target_schema, + "entity_key": self.entity_key, + "field": self.field, + "candidates": self.candidates or {}, + "selected_source": self.selected_source, + "selected_value": self.selected_value, + "selected_reason": self.selected_reason, + "resolved_by": self.resolved_by, + "status": self.status, + "created_at": to_iso8601_utc(self.created_at), + "updated_at": to_iso8601_utc(self.updated_at), + } + + +class AISSourceHealth(Base): + """Runtime health signal for an AIS collector source.""" + + __tablename__ = "ais_source_health" + + source = Column(String(100), primary_key=True) + connection_state = Column(String(32), nullable=False, default="disconnected", index=True) + last_seen_at = Column(DateTime(timezone=True), nullable=True, index=True) + last_success_at = Column(DateTime(timezone=True), nullable=True, index=True) + last_error = Column(String(500), nullable=True) + message_rate = Column(Float, nullable=True) + lag_seconds = Column(Float, nullable=True) + updated_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), index=True) + + def to_dict(self) -> dict: + return { + "source": self.source, + "connection_state": self.connection_state, + "last_seen_at": to_iso8601_utc(self.last_seen_at), + "last_success_at": to_iso8601_utc(self.last_success_at), + "last_error": self.last_error, + "message_rate": self.message_rate, + "lag_seconds": self.lag_seconds, + "updated_at": to_iso8601_utc(self.updated_at), + } diff --git a/backend/app/services/collectors/__init__.py b/backend/app/services/collectors/__init__.py index ebe4507d..9ddbf91b 100644 --- a/backend/app/services/collectors/__init__.py +++ b/backend/app/services/collectors/__init__.py @@ -36,6 +36,7 @@ from app.services.collectors.iptoasn import IPtoASNPrefixGeoCollector from app.services.collectors.opengeofeed import OpenGeoFeedPrefixGeoCollector from app.services.collectors.nro_delegated import NRODelegatedPrefixGeoCollector from app.services.collectors.news_live_streams import NewsLiveStreamsCollector +from app.services.collectors.aisstream import AISStreamCollector from app.services.collectors.vessel_ais import VesselAISCollector collector_registry.register(TOP500Collector()) @@ -65,3 +66,40 @@ collector_registry.register(OpenGeoFeedPrefixGeoCollector()) collector_registry.register(NRODelegatedPrefixGeoCollector()) collector_registry.register(NewsLiveStreamsCollector()) collector_registry.register(VesselAISCollector()) +collector_registry.register(AISStreamCollector()) + +__all__ = [ + "BaseCollector", + "HTTPCollector", + "IntervalCollector", + "collector_registry", + "CollectorRegistry", + "TOP500Collector", + "EpochAIGPUCollector", + "HuggingFaceModelCollector", + "HuggingFaceDatasetCollector", + "HuggingFaceSpacesCollector", + "PeeringDBIXPCollector", + "PeeringDBNetworkCollector", + "PeeringDBFacilityCollector", + "TeleGeographyCableCollector", + "TeleGeographyLandingPointCollector", + "TeleGeographyCableSystemCollector", + "CloudflareRadarDeviceCollector", + "CloudflareRadarTrafficCollector", + "CloudflareRadarTopASCollector", + "ArcGISCableCollector", + "FAOLandingPointCollector", + "ArcGISLandingPointCollector", + "ArcGISCableLandingRelationCollector", + "SpaceTrackTLECollector", + "CelesTrakTLECollector", + "RISLiveCollector", + "BGPStreamBackfillCollector", + "IPtoASNPrefixGeoCollector", + "OpenGeoFeedPrefixGeoCollector", + "NRODelegatedPrefixGeoCollector", + "NewsLiveStreamsCollector", + "VesselAISCollector", + "AISStreamCollector", +] diff --git a/backend/app/services/collectors/aisstream.py b/backend/app/services/collectors/aisstream.py new file mode 100644 index 00000000..1c1ff272 --- /dev/null +++ b/backend/app/services/collectors/aisstream.py @@ -0,0 +1,276 @@ +"""AISStream WebSocket collector for realtime vessel AIS observations.""" + +from datetime import UTC, datetime +import asyncio +import json +import os +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.data_sources import get_data_sources_config +from app.models.datasource_config import DataSourceConfig +from app.services.collectors.base import BaseCollector +from app.services.vessel_ais_aggregation import ( + AISSTREAM_DELIVERY_MODE, + AISSTREAM_TRANSPORT, + record_vessel_ais_observation, + update_ais_source_health, +) +from app.services.vessel_types import normalize_vessel_type_name + +DEFAULT_AISSTREAM_URL = "wss://stream.aisstream.io/v0/stream" +DEFAULT_BOUNDING_BOXES = [[[-90, -180], [90, 180]]] +DEFAULT_MESSAGE_TYPES = ["PositionReport", "ShipStaticData"] + + +class AISStreamCollector(BaseCollector): + """Collect AISStream WebSocket messages into the raw AIS observation layer.""" + + name = "aisstream_vessels" + priority = "P1" + module = "L4" + frequency_hours = 1 + data_type = "vessel_ais" + fail_on_empty = False + + async def _load_datasource_config(self) -> DataSourceConfig | None: + if self._db_session is None: + return None + result = await self._db_session.execute( + select(DataSourceConfig) + .where(DataSourceConfig.name == self.name) + .where(DataSourceConfig.is_active.is_(True)) + ) + return result.scalar_one_or_none() + + async def _get_effective_config(self) -> dict[str, Any]: + datasource_config = await self._load_datasource_config() + config = dict(datasource_config.config or {}) if datasource_config else {} + auth_config = dict(datasource_config.auth_config or {}) if datasource_config else {} + endpoint = ( + (datasource_config.endpoint if datasource_config else None) + or self._resolved_url + or get_data_sources_config().get_yaml_url(self.name) + or DEFAULT_AISSTREAM_URL + ) + api_key = ( + auth_config.get("api_key") + or config.get("api_key") + or os.getenv("AISSTREAM_API_KEY") + ) + return { + "endpoint": endpoint, + "api_key": api_key, + "bounding_boxes": config.get("bounding_boxes") or DEFAULT_BOUNDING_BOXES, + "message_types": config.get("message_types") or DEFAULT_MESSAGE_TYPES, + "max_messages": int(config.get("max_messages") or 500), + "receive_timeout_seconds": float(config.get("receive_timeout_seconds") or 30), + } + + async def fetch(self) -> list[dict[str, Any]]: + config = await self._get_effective_config() + if not config["api_key"]: + raise RuntimeError("AISStream API key is not configured") + + try: + import websockets + except ImportError as exc: + raise RuntimeError("Python package 'websockets' is required for AISStream") from exc + + subscription = { + "APIKey": config["api_key"], + "BoundingBoxes": config["bounding_boxes"], + "FilterMessageTypes": config["message_types"], + } + + messages: list[dict[str, Any]] = [] + try: + async with websockets.connect(config["endpoint"]) as websocket: + await websocket.send(json.dumps(subscription)) + while len(messages) < config["max_messages"]: + try: + raw_message = await asyncio.wait_for( + websocket.recv(), + timeout=config["receive_timeout_seconds"], + ) + except TimeoutError: + break + payload = json.loads(raw_message) + if isinstance(payload, dict): + messages.append(payload) + except Exception as exc: + if self._db_session is not None: + await update_ais_source_health( + self._db_session, + source=self.name, + connection_state="disconnected", + last_error=f"{exc.__class__.__name__}: {exc}", + ) + await self._db_session.commit() + raise + + return messages + + def transform(self, raw_data: list[dict[str, Any]]) -> list[dict[str, Any]]: + records = [] + for item in raw_data: + record = self._normalize_message(item) + if record: + records.append(record) + return records + + async def _save_data( + self, + db: AsyncSession, + data: list[dict[str, Any]], + task_id: int | None = None, + snapshot_id: int | None = None, + ) -> int: + now = datetime.now(UTC) + records_added = 0 + latest_observed_at = now + for index, item in enumerate(data): + observed_at = item.get("received_at") or now + observation = await record_vessel_ais_observation( + db, + source=self.name, + normalized_payload=item, + raw_payload=item.get("_raw_payload") or item, + delivery_mode=AISSTREAM_DELIVERY_MODE, + transport=AISSTREAM_TRANSPORT, + message_type=item.get("_message_type") or "PositionReport", + source_message_id=item.get("_source_message_id"), + observed_at=observed_at, + collected_at=now, + ) + if observation is not None: + records_added += 1 + if isinstance(observed_at, datetime) and observed_at > latest_observed_at: + latest_observed_at = observed_at + if (index + 1) % 1000 == 0: + await self.update_progress(index + 1, commit=True) + + await update_ais_source_health( + db, + source=self.name, + connection_state="connected", + observed_count=len(data), + last_seen_at=latest_observed_at, + last_success_at=now if data else None, + lag_seconds=max((now - latest_observed_at).total_seconds(), 0), + ) + await db.commit() + await self.update_progress(records_added, force=True) + return records_added + + def _normalize_message(self, item: dict[str, Any]) -> dict[str, Any] | None: + message_type = str(item.get("MessageType") or item.get("message_type") or "") + metadata = item.get("MetaData") if isinstance(item.get("MetaData"), dict) else {} + message = item.get("Message") if isinstance(item.get("Message"), dict) else {} + body = message.get(message_type) if isinstance(message.get(message_type), dict) else message + if not isinstance(body, dict): + body = {} + + mmsi = _as_int(_pick(metadata, "MMSI", "mmsi") or _pick(body, "MMSI", "mmsi")) + if mmsi is None: + return None + + received_at = _parse_datetime( + _pick(metadata, "time_utc", "Time_UTC", "timestamp") + or _pick(body, "Timestamp", "timestamp", "time") + ) + ship_name = _clean_text( + _pick(body, "Name", "ShipName", "name") + or _pick(metadata, "ShipName", "ship_name", "name") + ) + record: dict[str, Any] = { + "mmsi": mmsi, + "received_at": received_at, + "_message_type": message_type or None, + "_source_message_id": item.get("MessageID") or item.get("message_id"), + "_raw_payload": item, + } + + lat = _as_float(_pick(body, "Latitude", "lat", "latitude")) + lon = _as_float(_pick(body, "Longitude", "lon", "lng", "longitude")) + if lat is not None and lon is not None: + if not (-90 <= lat <= 90 and -180 <= lon <= 180): + return None + record.update( + { + "lat": lat, + "lon": lon, + "sog": _as_float(_pick(body, "Sog", "SOG", "speedOverGround")), + "cog": _as_float(_pick(body, "Cog", "COG", "courseOverGround")), + "heading": _as_int(_pick(body, "TrueHeading", "Heading", "heading")), + "nav_status": _as_int(_pick(body, "NavigationalStatus", "nav_status")), + } + ) + + vessel_type = _as_int(_pick(body, "Type", "ShipType", "vessel_type")) + record.update( + { + "name": ship_name, + "callsign": _pick(body, "CallSign", "callsign"), + "imo": _as_int(_pick(body, "ImoNumber", "IMO", "imo")), + "vessel_type": vessel_type, + "vessel_type_name": _pick(body, "TypeName", "ShipTypeName", "vessel_type_name") + or normalize_vessel_type_name(vessel_type), + "length": _as_float(_pick(body, "DimensionToBow", "Length", "length")), + "width": _as_float(_pick(body, "DimensionToPort", "Width", "width")), + } + ) + return record + + +def _pick(item: dict[str, Any], *keys: str) -> Any: + for key in keys: + if key in item and item[key] not in (None, ""): + return item[key] + return None + + +def _clean_text(value: Any) -> str | None: + if value in (None, ""): + return None + text = str(value).strip() + return text or None + + +def _as_float(value: Any) -> float | None: + try: + if value in (None, ""): + return None + return float(value) + except (TypeError, ValueError): + return None + + +def _as_int(value: Any) -> int | None: + try: + if value in (None, ""): + return None + return int(float(value)) + except (TypeError, ValueError): + return None + + +def _parse_datetime(value: Any) -> datetime | None: + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=UTC) + if not value: + return None + if isinstance(value, (int, float)): + timestamp = float(value) + if timestamp > 10_000_000_000: + timestamp /= 1000 + return datetime.fromtimestamp(timestamp, UTC) + if isinstance(value, str): + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) + except ValueError: + return None + return None diff --git a/backend/app/services/collectors/vessel_ais.py b/backend/app/services/collectors/vessel_ais.py index ff1cb818..95e4e383 100644 --- a/backend/app/services/collectors/vessel_ais.py +++ b/backend/app/services/collectors/vessel_ais.py @@ -4,7 +4,7 @@ from datetime import UTC, datetime, timedelta from typing import Any import httpx -from sqlalchemy import delete, select +from sqlalchemy import delete from sqlalchemy.ext.asyncio import AsyncSession from app.models.vessel import VesselPosition, VesselStatic @@ -14,15 +14,13 @@ from app.services.barentswatch import ( resolve_barentswatch_config, ) from app.services.collectors.base import BaseCollector - - -VESSEL_TYPE_NAMES = { - 30: "Fishing", - 35: "Military", - 60: "Passenger", - 70: "Cargo", - 80: "Tanker", -} +from app.services.vessel_ais_aggregation import ( + BARENTSWATCH_DELIVERY_MODE, + BARENTSWATCH_TRANSPORT, + record_vessel_ais_observation, + update_ais_source_health, +) +from app.services.vessel_types import normalize_vessel_type_name class VesselAISCollector(BaseCollector): @@ -92,6 +90,18 @@ class VesselAISCollector(BaseCollector): records_added = 0 for index, item in enumerate(data): + observed_at = item.get("received_at") or now + await record_vessel_ais_observation( + db, + source=self.name, + normalized_payload=item, + raw_payload=item, + delivery_mode=BARENTSWATCH_DELIVERY_MODE, + transport=BARENTSWATCH_TRANSPORT, + observed_at=observed_at, + collected_at=now, + ) + static = await db.get(VesselStatic, item["mmsi"]) if static is None: static = VesselStatic(mmsi=item["mmsi"]) @@ -122,7 +132,7 @@ class VesselAISCollector(BaseCollector): cog=item.get("cog"), heading=item.get("heading"), nav_status=item.get("nav_status"), - received_at=item.get("received_at") or now, + received_at=observed_at, ) ) records_added += 1 @@ -130,6 +140,19 @@ class VesselAISCollector(BaseCollector): if (index + 1) % 1000 == 0: await self.update_progress(index + 1, commit=True) + latest_observed_at = max( + (item.get("received_at") for item in data if item.get("received_at")), + default=now, + ) + await update_ais_source_health( + db, + source=self.name, + connection_state="connected", + observed_count=len(data), + last_seen_at=latest_observed_at, + last_success_at=now if data else None, + lag_seconds=max((now - latest_observed_at).total_seconds(), 0), + ) await db.execute( delete(VesselPosition).where(VesselPosition.received_at < now - timedelta(hours=24)) ) @@ -156,7 +179,7 @@ class VesselAISCollector(BaseCollector): vessel_type = _as_int(_pick(item, "vessel_type", "shipType", "ship_type", "ShipType")) vessel_type_name = ( _pick(item, "vessel_type_name", "shipTypeName", "ship_type_name", "VesselTypeName") - or _vessel_type_name(vessel_type) + or normalize_vessel_type_name(vessel_type) ) received_at = _parse_datetime(_pick(item, "received_at", "timestamp", "time", "msgtime")) @@ -255,19 +278,3 @@ def _parse_datetime(value: Any) -> datetime | None: except ValueError: return None return None - - -def _vessel_type_name(vessel_type: int | None) -> str: - if vessel_type is None: - return "Other" - if 70 <= vessel_type <= 79: - return "Cargo" - if 80 <= vessel_type <= 89: - return "Tanker" - if 60 <= vessel_type <= 69: - return "Passenger" - if vessel_type == 30: - return "Fishing" - if vessel_type == 35: - return "Military" - return VESSEL_TYPE_NAMES.get(vessel_type, "Other") diff --git a/backend/app/services/credential_guides.py b/backend/app/services/credential_guides.py index eeb55fe3..49ba6ce1 100644 --- a/backend/app/services/credential_guides.py +++ b/backend/app/services/credential_guides.py @@ -66,9 +66,68 @@ BARENTSWATCH_DEFAULT_GUIDE = CredentialGuideDefault( """, ) +AISSTREAM_DEFAULT_GUIDE = CredentialGuideDefault( + provider="aisstream", + title="AISStream API Key 获取教程", + prompt=( + "请生成一份中文教程，指导开发者获取 AISStream 的 API Key 并配置到 Planet。" + "教程要面向已经有本地开发环境的人，包含注册/登录 AISStream、获取 API Key、" + "理解免费额度和订阅范围、在 Planet 设置中心填写 API Key、配置 bounding boxes " + "和 message types、验证连接、常见失败排查。必须提醒用户以 AISStream 当前官网和" + "服务条款为准，不要编造具体页面按钮文案。" + ), + markdown="""## AISStream API Key 获取 + +官方入口：https://aisstream.io/ + +1. 打开 AISStream 官网，按当前页面指引注册或登录账号。 +2. 在账号/API 管理页面创建或复制你的 API Key。 +3. 先确认当前账号额度、使用条款和可订阅区域。实时 AIS 流量可能很大，不建议一开始订阅全球范围。 +4. 回到 Planet 的 `设置 -> 采集器设置 -> AISStream 实时船舶`。 +5. 在 `AISStream 凭证` 中填入 API Key。 +6. Endpoint 通常保持默认：`wss://stream.aisstream.io/v0/stream`。 +7. 按需配置 `Bounding Boxes JSON` 和 `消息类型`。 +8. 点击连接测试，确认系统能读取凭证且 WebSocket endpoint 格式有效。 +9. 保存采集器设置后再运行 `aisstream_vessels` collector。 + +### 推荐配置 + +默认消息类型： + +```json +["PositionReport", "ShipStaticData"] +``` + +默认 Bounding Boxes 示例： + +```json +[[[-90, -180], [90, 180]]] +``` + +这个示例表示全球范围。实际使用时建议先改成较小区域，降低消息量和处理压力。 + +### 请求规则 + +- Endpoint：`wss://stream.aisstream.io/v0/stream` +- 传输方式：WebSocket +- API Key 放在订阅 payload 中，不放在 HTTP header。 +- Planet 会把 AISStream 标记为 `delivery_mode = realtime_stream`、`transport = websocket`。 +- AISStream collector 只写入 AIS raw observations，不直接覆盖最终船只展示表。 + +### 常见排查 + +- `未找到凭证`：确认 API Key 已保存到采集器设置，或设置了 `AISSTREAM_API_KEY` 环境变量 / `~/.zshrc`。 +- `endpoint 必须是 ws:// 或 wss://`：AISStream 是 WebSocket 流接口，不要填普通 `https://` API 地址。 +- 采集量过大：缩小 `Bounding Boxes JSON`，减少 `message_types`，或降低单次最大消息数。 +- 没有船只数据：确认订阅区域内确实有 AIS 活动，并检查 API Key 当前额度和权限。 +- 连接中断：实时流可能受网络和上游限流影响，collector 会记录源健康状态供聚合服务回退。 +""", +) + DEFAULT_CREDENTIAL_GUIDES = { BARENTSWATCH_DEFAULT_GUIDE.provider: BARENTSWATCH_DEFAULT_GUIDE, + AISSTREAM_DEFAULT_GUIDE.provider: AISSTREAM_DEFAULT_GUIDE, } diff --git a/backend/app/services/datasource_connectivity.py b/backend/app/services/datasource_connectivity.py index f38795d7..9bc89063 100644 --- a/backend/app/services/datasource_connectivity.py +++ b/backend/app/services/datasource_connectivity.py @@ -26,6 +26,7 @@ from app.services.barentswatch import ( CONNECTIVITY_VALIDATION_KEY = "connectivity_validation" CONNECTIVITY_STORE_CATEGORY = "datasource_connectivity_validations" +SUPPORTED_CREDENTIAL_PROVIDERS = {"barentswatch", "spacetrack", "aisstream"} def _sha256_json(payload: Any) -> str: @@ -43,6 +44,36 @@ def _resolve_spacetrack_credentials() -> tuple[str, str, str]: return username, password, source or "missing" +async def _resolve_aisstream_api_key( + db=None, + credential_override: dict[str, str] | None = None, +) -> tuple[str, str]: + if credential_override and credential_override.get("api_key"): + return str(credential_override["api_key"]), "draft" + + env_key = os.getenv("AISSTREAM_API_KEY") + zshrc_key = _read_zshrc_env().get("AISSTREAM_API_KEY") + if db is not None: + result = await db.execute( + select(DataSourceConfig) + .where(DataSourceConfig.name == "aisstream_vessels") + .where(DataSourceConfig.is_active.is_(True)) + ) + record = result.scalar_one_or_none() + if record: + auth_config = record.auth_config or {} + runtime_config = record.config or {} + api_key = auth_config.get("api_key") or runtime_config.get("api_key") + if api_key: + return str(api_key), "datasource_config" + + if env_key: + return env_key, "environment" + if zshrc_key: + return zshrc_key, "~/.zshrc" + return "", "missing" + + def strip_connectivity_validation(config: dict | None) -> dict: cleaned = dict(config or {}) cleaned.pop(CONNECTIVITY_VALIDATION_KEY, None) @@ -103,6 +134,10 @@ async def build_builtin_connectivity_checksum( "password": password, } ) + elif credential_provider == "aisstream": + api_key, credential_source = await _resolve_aisstream_api_key(db, credential_override) + has_credentials = bool(api_key) + credential_fingerprint = _sha256_json({"api_key": api_key}) elif defaults.get("requires_credentials"): credential_source = str(credential_provider or "unsupported") @@ -130,6 +165,7 @@ async def test_builtin_connectivity( headers: dict | None, config: dict | None, db=None, + credential_override: dict[str, str] | None = None, ) -> dict[str, Any]: defaults = DEFAULT_DATASOURCES.get(source) if not defaults: @@ -145,6 +181,7 @@ async def test_builtin_connectivity( headers, config, db, + credential_override, ) if credential_context["requires_credentials"] and not credential_context["has_credentials"]: return { @@ -155,10 +192,9 @@ async def test_builtin_connectivity( "settings_tab": "collector_credentials", **credential_context, } - supported_credential_providers = {"barentswatch", "spacetrack"} if ( credential_context["requires_credentials"] - and credential_context["credential_provider"] not in supported_credential_providers + and credential_context["credential_provider"] not in SUPPORTED_CREDENTIAL_PROVIDERS ): return { "success": False, @@ -174,6 +210,23 @@ async def test_builtin_connectivity( timeout = float(request_config.get("timeout") or 30) request_endpoint = endpoint + if credential_context["credential_provider"] == "aisstream": + if not str(request_endpoint).startswith(("ws://", "wss://")): + return { + "success": False, + "checksum": checksum, + "stage": "endpoint", + "message": "AISStream endpoint 必须是 ws:// 或 wss:// WebSocket 地址。", + **credential_context, + } + return { + "success": True, + "checksum": checksum, + "stage": "credentials", + "message": "AISStream 凭证已配置，WebSocket endpoint 格式有效。", + **credential_context, + } + try: async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: if credential_context["credential_provider"] == "barentswatch": diff --git a/backend/app/services/vessel_ais_aggregation.py b/backend/app/services/vessel_ais_aggregation.py new file mode 100644 index 00000000..8fa6dcff --- /dev/null +++ b/backend/app/services/vessel_ais_aggregation.py @@ -0,0 +1,574 @@ +"""AIS raw observation and aggregation support for vessel collectors.""" + +from datetime import UTC, datetime +from hashlib import sha256 +import json +from typing import Any, Iterable + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.vessel import AISConflictRecord, AISRawObservation, AISSourceHealth +from app.services.vessel_types import normalize_vessel_type_name + +VESSEL_AIS_SCHEMA = "vessel_ais" +BARENTSWATCH_DELIVERY_MODE = "polling" +BARENTSWATCH_TRANSPORT = "http" +AISSTREAM_DELIVERY_MODE = "realtime_stream" +AISSTREAM_TRANSPORT = "websocket" +DELIVERY_MODE_PRIORITY = { + "realtime_stream": 40, + "batch_stream": 30, + "polling": 20, + "snapshot": 10, +} +DYNAMIC_FIELDS = ("lat", "lon", "sog", "cog", "heading", "nav_status") +CONFLICT_FIELDS = ( + "name", + "callsign", + "imo", + "flag", + "vessel_type", + "vessel_type_name", + "length", + "width", + "draught", +) + + +def _json_default(value: Any) -> Any: + if isinstance(value, datetime): + return value.astimezone(UTC).isoformat() + return str(value) + + +def _stable_payload(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=_json_default) + + +def _jsonable(value: Any) -> Any: + if isinstance(value, datetime): + return value.astimezone(UTC).isoformat() + if isinstance(value, dict): + return {str(key): _jsonable(item) for key, item in value.items()} + if isinstance(value, list): + return [_jsonable(item) for item in value] + return value + + +def _coerce_datetime(value: Any) -> datetime | None: + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=UTC) + if isinstance(value, (int, float)): + timestamp = float(value) + if timestamp > 10_000_000_000: + timestamp /= 1000 + return datetime.fromtimestamp(timestamp, UTC) + if isinstance(value, str) and value: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) + except ValueError: + return None + return None + + +def build_observation_hash( + *, + source: str, + entity_key: str, + message_type: str | None, + observed_at: datetime, + normalized_payload: dict[str, Any], + source_message_id: str | None = None, +) -> str: + """Build a deterministic idempotency key for one source-level AIS observation.""" + + if source_message_id: + basis = { + "source": source, + "entity_key": entity_key, + "source_message_id": source_message_id, + } + else: + basis = { + "source": source, + "entity_key": entity_key, + "message_type": message_type, + "observed_at": observed_at.astimezone(UTC).isoformat(), + "payload": normalized_payload, + } + return sha256(_stable_payload(basis).encode("utf-8")).hexdigest() + + +def build_field_conflict_candidates( + observations: Iterable[AISRawObservation], + fields: Iterable[str] = CONFLICT_FIELDS, +) -> list[dict[str, Any]]: + """Return current field disagreements from raw observations without mutating state.""" + + candidates_by_field: dict[str, dict[str, Any]] = {} + for observation in observations: + payload = observation.normalized_payload or {} + for field in fields: + value = payload.get(field) + if value in (None, ""): + continue + field_candidates = candidates_by_field.setdefault(field, {}) + field_candidates[observation.source] = value + + conflicts = [] + for field, candidates in sorted(candidates_by_field.items()): + unique_values = {_stable_payload(value) for value in candidates.values()} + if len(unique_values) <= 1: + continue + conflicts.append( + { + "field": field, + "candidates": candidates, + "status": "candidate", + } + ) + return conflicts + + +def _payload_value(payload: dict[str, Any], field: str) -> Any: + value = payload.get(field) + return None if value in (None, "") else value + + +def _clean_text(value: Any) -> str | None: + if value in (None, ""): + return None + text = str(value).strip() + return text or None + + +def _raw_metadata_value(observation: AISRawObservation, field: str) -> Any: + raw_payload = observation.raw_payload or {} + metadata = raw_payload.get("MetaData") if isinstance(raw_payload, dict) else None + if not isinstance(metadata, dict): + return None + if field == "name": + return _clean_text(metadata.get("ShipName") or metadata.get("ship_name") or metadata.get("name")) + return None + + +def _delivery_priority(observation: AISRawObservation) -> int: + return DELIVERY_MODE_PRIORITY.get(str(observation.delivery_mode or ""), 0) + + +def _has_valid_position(payload: dict[str, Any]) -> bool: + try: + lat = float(payload.get("lat")) + lon = float(payload.get("lon")) + except (TypeError, ValueError): + return False + return -90 <= lat <= 90 and -180 <= lon <= 180 + + +def _is_future_observation(observation: AISRawObservation, now: datetime) -> bool: + return observation.observed_at > now + + +def _select_position_observation( + observations: list[AISRawObservation], + *, + now: datetime, +) -> tuple[AISRawObservation | None, list[str]]: + rejected_flags: list[str] = [] + candidates = [] + for observation in observations: + payload = observation.normalized_payload or {} + if not _has_valid_position(payload): + rejected_flags.append("invalid_position") + continue + if _is_future_observation(observation, now): + rejected_flags.append("future_timestamp") + continue + candidates.append(observation) + + if not candidates: + return None, sorted(set(rejected_flags)) + + candidates.sort( + key=lambda item: ( + item.observed_at, + _delivery_priority(item), + item.collected_at, + item.id or 0, + ), + reverse=True, + ) + return candidates[0], sorted(set(rejected_flags)) + + +def _select_static_field( + observations: list[AISRawObservation], + field: str, +) -> tuple[Any, str | None, str | None]: + candidates = [] + for observation in observations: + value = _payload_value(observation.normalized_payload or {}, field) + if value is None: + value = _raw_metadata_value(observation, field) + if value is None: + continue + candidates.append((observation, value)) + + if not candidates: + return None, None, None + + candidates.sort( + key=lambda item: ( + _delivery_priority(item[0]), + item[0].observed_at, + item[0].collected_at, + item[0].id or 0, + ), + reverse=True, + ) + selected_observation, selected_value = candidates[0] + unique_values = {_stable_payload(value) for _, value in candidates} + reason = "delivery_mode_priority" if len(unique_values) > 1 else "non_empty_priority" + return selected_value, selected_observation.source, reason + + +def _build_source_summary(observations: list[AISRawObservation]) -> dict[str, dict[str, Any]]: + summary: dict[str, dict[str, Any]] = {} + for observation in observations: + source_summary = summary.setdefault( + observation.source, + { + "observation_count": 0, + "latest_observed_at": None, + "delivery_mode": observation.delivery_mode, + "transport": observation.transport, + "message_types": [], + }, + ) + source_summary["observation_count"] += 1 + latest_observed_at = source_summary["latest_observed_at"] + if latest_observed_at is None or observation.observed_at > latest_observed_at: + source_summary["latest_observed_at"] = observation.observed_at + if observation.message_type and observation.message_type not in source_summary["message_types"]: + source_summary["message_types"].append(observation.message_type) + return summary + + +def _build_aggregated_vessel( + entity_key: str, + observations: list[AISRawObservation], + *, + now: datetime, +) -> dict[str, Any] | None: + position_observation, rejected_flags = _select_position_observation(observations, now=now) + if position_observation is None: + return None + + payload = position_observation.normalized_payload or {} + mmsi = int(entity_key) + result: dict[str, Any] = { + "mmsi": mmsi, + "lat": float(payload["lat"]), + "lon": float(payload["lon"]), + "received_at": position_observation.observed_at, + "field_sources": {}, + "selected_reasons": {}, + "source_summary": _build_source_summary(observations), + "quality_flags": sorted( + set((position_observation.quality_flags or []) + rejected_flags) + ), + } + + for field in DYNAMIC_FIELDS: + value = _payload_value(payload, field) + if field in ("lat", "lon") or value is not None: + result[field] = value + result["field_sources"][field] = position_observation.source + result["selected_reasons"][field] = "newest_observation" + + for field in CONFLICT_FIELDS: + selected_value, selected_source, reason = _select_static_field(observations, field) + if selected_value is None: + continue + result[field] = selected_value + result["field_sources"][field] = selected_source + result["selected_reasons"][field] = reason + + result["name"] = result.get("name") or f"MMSI {mmsi}" + result["vessel_type_name"] = result.get("vessel_type_name") or normalize_vessel_type_name( + result.get("vessel_type") + ) + return result + + +async def _upsert_conflict_records( + db: AsyncSession, + entity_key: str, + observations: list[AISRawObservation], + aggregated: dict[str, Any], +) -> int: + conflicts = build_field_conflict_candidates(observations) + now = datetime.now(UTC) + for conflict in conflicts: + field = conflict["field"] + result = await db.execute( + select(AISConflictRecord) + .where(AISConflictRecord.target_schema == VESSEL_AIS_SCHEMA) + .where(AISConflictRecord.entity_key == entity_key) + .where(AISConflictRecord.field == field) + .limit(1) + ) + record = result.scalar_one_or_none() + if record is None: + record = AISConflictRecord( + target_schema=VESSEL_AIS_SCHEMA, + entity_key=entity_key, + field=field, + ) + db.add(record) + record.candidates = conflict["candidates"] + record.selected_source = (aggregated.get("field_sources") or {}).get(field) + record.selected_value = aggregated.get(field) + record.selected_reason = (aggregated.get("selected_reasons") or {}).get(field) + record.resolved_by = "system" + record.status = "open" + record.updated_at = now + return len(conflicts) + + +def _group_observations(observations: Iterable[AISRawObservation]) -> dict[str, list[AISRawObservation]]: + grouped: dict[str, list[AISRawObservation]] = {} + for observation in observations: + grouped.setdefault(str(observation.entity_key), []).append(observation) + return grouped + + +async def record_vessel_ais_observation( + db: AsyncSession, + *, + source: str, + normalized_payload: dict[str, Any], + raw_payload: dict[str, Any] | None = None, + delivery_mode: str, + transport: str, + message_type: str | None = "PositionReport", + source_message_id: str | None = None, + observed_at: datetime | None = None, + collected_at: datetime | None = None, + quality_flags: list[str] | None = None, +) -> AISRawObservation | None: + """Insert one raw observation if the source-level fact has not already been stored.""" + + entity_key = str(normalized_payload["mmsi"]) + collected_at = collected_at or datetime.now(UTC) + observed_at = ( + _coerce_datetime(observed_at) + or _coerce_datetime(normalized_payload.get("received_at")) + or collected_at + ) + normalized_json = _jsonable(normalized_payload) + raw_json = _jsonable(raw_payload or {}) + + observation_hash = build_observation_hash( + source=source, + entity_key=entity_key, + message_type=message_type, + observed_at=observed_at, + normalized_payload=normalized_json, + source_message_id=source_message_id, + ) + existing_result = await db.execute( + select(AISRawObservation.id).where(AISRawObservation.observation_hash == observation_hash) + ) + if existing_result.scalar_one_or_none() is not None: + return None + + observation = AISRawObservation( + target_schema=VESSEL_AIS_SCHEMA, + source=source, + entity_key=entity_key, + delivery_mode=delivery_mode, + transport=transport, + message_type=message_type, + source_message_id=source_message_id, + observation_hash=observation_hash, + observed_at=observed_at, + collected_at=collected_at, + normalized_payload=normalized_json, + raw_payload=raw_json, + quality_flags=quality_flags or [], + ) + db.add(observation) + return observation + + +async def aggregate_vessel_observations( + db: AsyncSession, + observations: Iterable[AISRawObservation], + *, + write_conflicts: bool = True, +) -> list[dict[str, Any]]: + now = datetime.now(UTC) + vessels = [] + for entity_key, entity_observations in _group_observations(observations).items(): + aggregated = _build_aggregated_vessel(entity_key, entity_observations, now=now) + if aggregated is None: + continue + if write_conflicts: + aggregated["conflict_count"] = await _upsert_conflict_records( + db, + entity_key, + entity_observations, + aggregated, + ) + else: + aggregated["conflict_count"] = len(build_field_conflict_candidates(entity_observations)) + vessels.append(aggregated) + + vessels.sort(key=lambda item: item.get("received_at") or datetime.min.replace(tzinfo=UTC), reverse=True) + return vessels + + +async def get_aggregated_vessels( + db: AsyncSession, + *, + bbox: tuple[float, float, float, float] | None = None, + limit: int | None = None, +) -> list[dict[str, Any]]: + stmt = ( + select(AISRawObservation) + .where(AISRawObservation.target_schema == VESSEL_AIS_SCHEMA) + .order_by(AISRawObservation.observed_at.desc(), AISRawObservation.id.desc()) + ) + if limit and limit > 0: + stmt = stmt.limit(max(limit * 20, limit)) + + result = await db.execute(stmt) + if not hasattr(result, "scalars"): + return [] + vessels = await aggregate_vessel_observations(db, result.scalars().all()) + + if bbox is not None: + lon_min, lat_min, lon_max, lat_max = bbox + vessels = [ + vessel + for vessel in vessels + if lon_min <= float(vessel["lon"]) <= lon_max + and lat_min <= float(vessel["lat"]) <= lat_max + ] + + if limit and limit > 0: + return vessels[:limit] + return vessels + + +async def get_aggregated_vessel(db: AsyncSession, mmsi: int) -> dict[str, Any] | None: + observations = await get_vessel_raw_observations(db, mmsi, limit=1000) + vessels = await aggregate_vessel_observations(db, observations) + return vessels[0] if vessels else None + + +async def get_aggregated_vessel_track( + db: AsyncSession, + mmsi: int, + *, + cutoff: datetime, +) -> list[dict[str, Any]]: + result = await db.execute( + select(AISRawObservation) + .where(AISRawObservation.target_schema == VESSEL_AIS_SCHEMA) + .where(AISRawObservation.entity_key == str(mmsi)) + .where(AISRawObservation.observed_at >= cutoff) + .order_by(AISRawObservation.observed_at.asc(), AISRawObservation.id.asc()) + ) + if not hasattr(result, "scalars"): + return [] + + points: list[dict[str, Any]] = [] + seen: set[tuple[str, float, float, str]] = set() + for observation in result.scalars().all(): + payload = observation.normalized_payload or {} + if not _has_valid_position(payload): + continue + lat = float(payload["lat"]) + lon = float(payload["lon"]) + key = ( + observation.observed_at.isoformat(), + round(lat, 5), + round(lon, 5), + observation.source, + ) + if key in seen: + continue + seen.add(key) + points.append( + { + "lat": lat, + "lon": lon, + "observed_at": observation.observed_at, + "source": observation.source, + "selected_reason": "track_timeline", + "quality_flags": observation.quality_flags or [], + } + ) + return points + + +async def update_ais_source_health( + db: AsyncSession, + *, + source: str, + connection_state: str, + observed_count: int = 0, + last_seen_at: datetime | None = None, + last_success_at: datetime | None = None, + last_error: str | None = None, + lag_seconds: float | None = None, +) -> AISSourceHealth: + """Upsert the health row for an AIS source.""" + + now = datetime.now(UTC) + health = await db.get(AISSourceHealth, source) + if health is None: + health = AISSourceHealth(source=source) + db.add(health) + + health.connection_state = connection_state + health.last_seen_at = last_seen_at or health.last_seen_at + health.last_success_at = last_success_at or health.last_success_at + health.last_error = last_error + health.message_rate = float(observed_count) + health.lag_seconds = lag_seconds + health.updated_at = now + return health + + +async def get_vessel_raw_observations( + db: AsyncSession, + mmsi: int, + *, + limit: int = 100, +) -> list[AISRawObservation]: + result = await db.execute( + select(AISRawObservation) + .where(AISRawObservation.target_schema == VESSEL_AIS_SCHEMA) + .where(AISRawObservation.entity_key == str(mmsi)) + .order_by(AISRawObservation.observed_at.desc(), AISRawObservation.id.desc()) + .limit(limit) + ) + return list(result.scalars().all()) + + +async def get_vessel_conflict_records( + db: AsyncSession, + mmsi: int, +) -> list[AISConflictRecord]: + result = await db.execute( + select(AISConflictRecord) + .where(AISConflictRecord.target_schema == VESSEL_AIS_SCHEMA) + .where(AISConflictRecord.entity_key == str(mmsi)) + .order_by(AISConflictRecord.updated_at.desc(), AISConflictRecord.id.desc()) + ) + return list(result.scalars().all()) diff --git a/backend/app/services/vessel_types.py b/backend/app/services/vessel_types.py new file mode 100644 index 00000000..156cb183 --- /dev/null +++ b/backend/app/services/vessel_types.py @@ -0,0 +1,31 @@ +"""Shared AIS vessel type helpers.""" + +from typing import Any + +VESSEL_TYPE_NAMES = { + 30: "Fishing", + 35: "Military", + 60: "Passenger", + 70: "Cargo", + 80: "Tanker", +} + + +def normalize_vessel_type_name(vessel_type: Any) -> str: + """Map AIS numeric vessel type codes to display buckets.""" + + try: + type_code = int(float(vessel_type)) + except (TypeError, ValueError): + return "Other" + if 70 <= type_code <= 79: + return "Cargo" + if 80 <= type_code <= 89: + return "Tanker" + if 60 <= type_code <= 69: + return "Passenger" + if type_code == 30: + return "Fishing" + if type_code == 35: + return "Military" + return VESSEL_TYPE_NAMES.get(type_code, "Other") diff --git a/backend/tests/test_collectors.py b/backend/tests/test_collectors.py index e6cdfe76..d747f3a6 100644 --- a/backend/tests/test_collectors.py +++ b/backend/tests/test_collectors.py @@ -1,11 +1,13 @@ """Unit tests for data collectors""" import pytest -from datetime import datetime -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, patch +from app.core.datasource_defaults import DEFAULT_DATASOURCES +from app.services.credential_guides import DEFAULT_CREDENTIAL_GUIDES from app.services.collectors.top500 import TOP500Collector -from app.services.collectors.base import BaseCollector, HTTPCollector +from app.services.collectors.registry import collector_registry +from app.services.datasource_connectivity import SUPPORTED_CREDENTIAL_PROVIDERS from app.models.task import CollectionTask @@ -145,3 +147,30 @@ class TestHTTPCollector: assert hasattr(collector, "parse_response") assert callable(collector.fetch) assert callable(collector.parse_response) + + +def test_aisstream_collector_is_registered(): + collector = collector_registry.get("aisstream_vessels") + + assert collector is not None + assert collector.data_type == "vessel_ais" + + +def test_supported_credential_collectors_have_guides_and_connectivity_provider(): + missing: list[str] = [] + for source, info in DEFAULT_DATASOURCES.items(): + if not info.get("requires_credentials"): + continue + if info.get("credential_status") != "supported": + continue + + provider = info.get("credential_provider") + if not provider: + missing.append(f"{source}: missing credential_provider") + continue + if provider not in DEFAULT_CREDENTIAL_GUIDES: + missing.append(f"{source}: missing credential guide for {provider}") + if provider not in SUPPORTED_CREDENTIAL_PROVIDERS: + missing.append(f"{source}: missing connectivity provider for {provider}") + + assert missing == [] diff --git a/backend/tests/test_vessels.py b/backend/tests/test_vessels.py index d66f5084..76b92573 100644 --- a/backend/tests/test_vessels.py +++ b/backend/tests/test_vessels.py @@ -1,4 +1,5 @@ from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock import pytest from httpx import ASGITransport, AsyncClient @@ -6,9 +7,16 @@ from httpx import ASGITransport, AsyncClient from app.api.v1.visualization import convert_vessels_to_geojson from app.db.session import get_db from app.main import app -from app.models.vessel import VesselPosition, VesselStatic +from app.models.vessel import AISRawObservation, VesselPosition, VesselStatic from app.services import barentswatch +from app.services.collectors.aisstream import AISStreamCollector from app.services.collectors.vessel_ais import VesselAISCollector +from app.services.vessel_ais_aggregation import ( + aggregate_vessel_observations, + build_field_conflict_candidates, + build_observation_hash, + record_vessel_ais_observation, +) def test_vessel_collector_transforms_barentswatch_like_records(): @@ -35,6 +43,328 @@ def test_vessel_collector_transforms_barentswatch_like_records(): assert records[0]["lat"] == pytest.approx(59.91) +def test_vessel_observation_hash_is_stable_for_same_payload(): + observed_at = datetime(2026, 4, 30, 12, 0, tzinfo=timezone.utc) + payload = { + "mmsi": 257123000, + "lat": 59.91, + "lon": 10.73, + "received_at": observed_at, + } + + first = build_observation_hash( + source="barentswatch_vessels", + entity_key="257123000", + message_type="PositionReport", + observed_at=observed_at, + normalized_payload=payload, + ) + second = build_observation_hash( + source="barentswatch_vessels", + entity_key="257123000", + message_type="PositionReport", + observed_at=observed_at, + normalized_payload=dict(reversed(payload.items())), + ) + + assert first == second + assert len(first) == 64 + + +@pytest.mark.asyncio +async def test_record_vessel_ais_observation_skips_existing_hash(): + observed_at = datetime(2026, 4, 30, 12, 0, tzinfo=timezone.utc) + + class _Result: + def scalar_one_or_none(self): + return 123 + + class _Session: + def __init__(self): + self.added = [] + + async def execute(self, _stmt): + return _Result() + + def add(self, item): + self.added.append(item) + + db = _Session() + observation = await record_vessel_ais_observation( + db, + source="barentswatch_vessels", + normalized_payload={ + "mmsi": 257123000, + "lat": 59.91, + "lon": 10.73, + "received_at": observed_at, + }, + delivery_mode="polling", + transport="http", + observed_at=observed_at.isoformat(), + ) + + assert observation is None + assert db.added == [] + + +def test_build_field_conflict_candidates_from_raw_observations(): + observations = [ + AISRawObservation( + source="barentswatch_vessels", + normalized_payload={"name": "OSLO TRADER", "flag": "NO"}, + ), + AISRawObservation( + source="aisstream_vessels", + normalized_payload={"name": "OSLO TRADER II", "flag": "NO"}, + ), + ] + + conflicts = build_field_conflict_candidates(observations) + + assert conflicts == [ + { + "field": "name", + "candidates": { + "aisstream_vessels": "OSLO TRADER II", + "barentswatch_vessels": "OSLO TRADER", + }, + "status": "candidate", + } + ] + + +@pytest.mark.asyncio +async def test_aggregate_vessel_observations_prefers_realtime_and_records_conflict(): + observed_at = datetime.now(timezone.utc) - timedelta(minutes=5) + + class _Result: + def scalar_one_or_none(self): + return None + + class _Session: + def __init__(self): + self.added = [] + + async def execute(self, _stmt): + return _Result() + + def add(self, item): + self.added.append(item) + + db = _Session() + observations = [ + AISRawObservation( + id=1, + source="barentswatch_vessels", + entity_key="257123000", + delivery_mode="polling", + transport="http", + observed_at=observed_at, + collected_at=observed_at, + normalized_payload={ + "mmsi": 257123000, + "name": "OSLO TRADER", + "lat": 59.91, + "lon": 10.73, + }, + ), + AISRawObservation( + id=2, + source="aisstream_vessels", + entity_key="257123000", + delivery_mode="realtime_stream", + transport="websocket", + observed_at=observed_at + timedelta(seconds=10), + collected_at=observed_at + timedelta(seconds=10), + normalized_payload={ + "mmsi": 257123000, + "vessel_type": 79, + "lat": 59.92, + "lon": 10.74, + }, + raw_payload={"MetaData": {"ShipName": "OSLO TRADER II "}}, + ), + ] + + vessels = await aggregate_vessel_observations(db, observations) + + assert vessels[0]["lat"] == pytest.approx(59.92) + assert vessels[0]["field_sources"]["lat"] == "aisstream_vessels" + assert vessels[0]["name"] == "OSLO TRADER II" + assert vessels[0]["vessel_type_name"] == "Cargo" + assert vessels[0]["source_summary"]["aisstream_vessels"]["observation_count"] == 1 + assert vessels[0]["source_summary"]["barentswatch_vessels"]["delivery_mode"] == "polling" + assert vessels[0]["conflict_count"] == 0 + assert db.added == [] + + +@pytest.mark.asyncio +async def test_vessel_collector_writes_raw_observations_without_changing_position_save(monkeypatch): + collector = VesselAISCollector() + collector.update_progress = AsyncMock() + record_observation = AsyncMock() + update_health = AsyncMock() + monkeypatch.setattr( + "app.services.collectors.vessel_ais.record_vessel_ais_observation", + record_observation, + ) + monkeypatch.setattr( + "app.services.collectors.vessel_ais.update_ais_source_health", + update_health, + ) + + class _Session: + def __init__(self): + self.added = [] + self.committed = False + + async def get(self, *_args): + return None + + def add(self, item): + self.added.append(item) + + async def execute(self, _stmt): + return None + + async def commit(self): + self.committed = True + + db = _Session() + observed_at = datetime(2026, 4, 30, 12, 0, tzinfo=timezone.utc) + + saved = await collector._save_data( + db, + [ + { + "mmsi": 257123000, + "name": "OSLO TRADER", + "lat": 59.91, + "lon": 10.73, + "received_at": observed_at, + } + ], + ) + + assert saved == 1 + assert db.committed is True + assert any(isinstance(item, VesselStatic) for item in db.added) + assert any(isinstance(item, VesselPosition) for item in db.added) + record_observation.assert_awaited_once() + assert record_observation.await_args.kwargs["source"] == "barentswatch_vessels" + assert record_observation.await_args.kwargs["normalized_payload"]["mmsi"] == 257123000 + update_health.assert_awaited_once() + + +def test_aisstream_collector_normalizes_position_report(): + collector = AISStreamCollector() + + records = collector.transform( + [ + { + "MessageType": "PositionReport", + "MetaData": { + "MMSI": 257123000, + "ShipName": "OSLO TRADER ", + "time_utc": "2026-04-30T12:00:00Z", + }, + "Message": { + "PositionReport": { + "Latitude": 59.91, + "Longitude": 10.73, + "Sog": 12.4, + "Cog": 214, + "TrueHeading": 215, + "NavigationalStatus": 0, + } + }, + } + ] + ) + + assert len(records) == 1 + assert records[0]["mmsi"] == 257123000 + assert records[0]["lat"] == pytest.approx(59.91) + assert records[0]["name"] == "OSLO TRADER" + assert records[0]["_message_type"] == "PositionReport" + + +def test_aisstream_collector_maps_ship_static_type_name(): + collector = AISStreamCollector() + + records = collector.transform( + [ + { + "MessageType": "ShipStaticData", + "MetaData": { + "MMSI": 257123000, + "time_utc": "2026-04-30T12:00:00Z", + }, + "Message": { + "ShipStaticData": { + "Name": "OSLO TRADER", + "Type": 79, + "CallSign": "LAAB", + } + }, + } + ] + ) + + assert len(records) == 1 + assert records[0]["vessel_type"] == 79 + assert records[0]["vessel_type_name"] == "Cargo" + + +@pytest.mark.asyncio +async def test_aisstream_collector_writes_only_raw_observations(monkeypatch): + collector = AISStreamCollector() + collector.update_progress = AsyncMock() + record_observation = AsyncMock(return_value=object()) + update_health = AsyncMock() + monkeypatch.setattr( + "app.services.collectors.aisstream.record_vessel_ais_observation", + record_observation, + ) + monkeypatch.setattr( + "app.services.collectors.aisstream.update_ais_source_health", + update_health, + ) + + class _Session: + def __init__(self): + self.added = [] + self.committed = False + + def add(self, item): + self.added.append(item) + + async def commit(self): + self.committed = True + + db = _Session() + saved = await collector._save_data( + db, + [ + { + "mmsi": 257123000, + "lat": 59.91, + "lon": 10.73, + "received_at": datetime(2026, 4, 30, 12, 0, tzinfo=timezone.utc), + "_message_type": "PositionReport", + } + ], + ) + + assert saved == 1 + assert db.added == [] + assert db.committed is True + record_observation.assert_awaited_once() + assert record_observation.await_args.kwargs["source"] == "aisstream_vessels" + update_health.assert_awaited_once() + + def test_barentswatch_reads_credentials_from_zshrc(tmp_path): zshrc = tmp_path / ".zshrc" zshrc.write_text( @@ -137,7 +467,7 @@ async def test_vessels_geojson_endpoint_filters_type_and_bbox(): async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get( "/api/v1/visualization/geo/vessels", - params={"bbox": "0,50,20,70", "type": "cargo"}, + params={"bbox": "0,50,20,70", "type": "cargo", "limit": 0}, ) assert response.status_code == 200 diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 58acec67..61ec5bd7 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -8,6 +8,23 @@ This project follows the repository versioning rule: - `improvement` -> `+0.0.1`（bugfix + 小功能混合） - `bugfix` -> `+0.0.1` +## [0.47.0] — 2026-04-30 + +Released: 2026-04-30 + +### ✨ Highlights +- 新增 AISStream WebSocket 船只采集器，并将 AIS 多源数据写入原始观测层，由聚合接口统一去重、合并和解释字段来源。 +- 设置页新增 AISStream API Key、采集范围 preset、运行状态、连接验证和凭证教程入口，让全球 AIS 采集链路可配置、可观察。 +- Earth 船只图层默认不再限制 5000 艘，并统一 marker 颜色、详情卡、hover 和搜索结果的船型归一化显示。 + +### 🔧 Improvements +- 聚合接口新增 `field_sources`、`selected_reasons`、`source_summary`、`quality_flags` 和冲突记录调试接口，动态字段默认优先采用更新的实时流观测。 +- AISStream 标准化支持 `MetaData.ShipName` 船名兜底，并将 AIS 数字船型映射为 Cargo / Tanker / Passenger / Fishing / Military。 +- 将仓库 docs 技能改为通用文档工作流，Planet 专属白名单、双语、裸文件标题和凭证教程规则迁移到 `docs/documentation-coverage-rules.md`。 +- 更新 AIS v4/v5 TODO 与计划文档，明确后续聚合策略配置、船舶资料 enrichment 和媒体缓存边界。 + +--- + ## [0.46.3] — 2026-04-30 Released: 2026-04-30 diff --git a/docs/documentation-coverage-rules.md b/docs/documentation-coverage-rules.md new file mode 100644 index 00000000..a3b517bf --- /dev/null +++ b/docs/documentation-coverage-rules.md @@ -0,0 +1,117 @@ +# Documentation Coverage Rules + +This file contains Planet-specific documentation coverage rules. Documentation skills and agents should read this file before deciding which docs to update. Keep tool-specific workflow in skills; keep product and repository rules here. + +## Scope Rules + +- User-visible workflow changes must update `docs/technical/zh/manual.md` and usually `docs/technical/zh/quickstart.md`. +- If an English counterpart exists for user-facing docs such as `manual.md` or `quickstart.md`, update `docs/technical/en/...` enough that it does not contradict the Chinese source. +- Control console page responsibility changes must update `docs/technical/zh/frontend-admin-frontend-context.md`. +- Earth frontend behavior changes must update `docs/technical/zh/earth-frontend-context.md`. +- Earth layer additions, `renderOrder`, altitude/radius offsets, depth strategy, pointer picking, legend modes, or layer panel/startup ordering must update `docs/technical/zh/earth-render-layer-order.md`. +- Earth layer visual style or legend symbol/color semantics should also update `docs/technical/zh/earth-layer-style-reference.md` when that reference is affected. +- Collector, datasource, credential, settings, connectivity, scheduler, or API changes must update the relevant backend docs, especially `docs/technical/zh/backend-collectors.md` and any datasource/settings-specific doc. +- When a change turns an old plan assumption into current behavior, update the relevant `docs/plans/*.md` with a status note instead of leaving contradictory instructions. +- Search docs for stale terms introduced by the change, for example old tab names, old route responsibilities, obsolete auth assumptions, or renamed UI labels. + +## Public Docs Rules + +- If adding a new technical document, add it to `docs/technical/zh/README.md` when it should be discoverable from the technical docs index. +- If a technical document should be visible in the public Docs page or linked from a technical README, register it in `frontend/src/pages/Docs/docs-content.ts` under `DOCS_METADATA`. Files under `docs/technical/{zh,en}/` are not automatically routable. +- For every public technical doc, keep the bilingual file pair in sync by filename: `docs/technical/zh/.md` and `docs/technical/en/.md`. If content is intentionally Chinese-only or English-only, state that intentionally in the final note. +- Public docs should use readable link text, not raw filenames such as `manual.md`. + +## Credential Collector Rules + +- Any built-in collector marked `requires_credentials: true` and `credential_status: supported` must have: + - a `credential_provider` in `backend/app/core/datasource_defaults.py`; + - a default credential guide in `backend/app/services/credential_guides.py`; + - a supported connectivity provider in `backend/app/services/datasource_connectivity.py`; + - settings UI guidance or a credential form in `frontend/src/pages/Settings/Settings.tsx`; + - a regression test that fails if the guide/provider is missing. + +## Recommended Checks + +Run the checks that match the affected docs. + +### Duplicate Bilingual Docs + +```bash +python - <<'PY' +from pathlib import Path +same = [] +for en in sorted(Path("docs/technical/en").glob("*.md")): + zh = Path("docs/technical/zh") / en.name + if zh.exists() and en.read_text() == zh.read_text(): + same.append(en.name) +if same: + raise SystemExit("identical en/zh docs: " + ", ".join(same)) +print("no identical en/zh docs") +PY +``` + +### Language-Less Technical Links + +```bash +rg -n "/home/ray/dev/linkong/planet/docs/technical/(?!zh|en)" docs/technical/zh --pcre2 +``` + +This should return no matches. + +### Public Docs Registry + +```bash +python - <<'PY' +import re +from pathlib import Path + +metadata = Path("frontend/src/pages/Docs/docs-content.ts").read_text() +known = set(re.findall(r"'([^']+\.md)':\s*\{", metadata)) +known.add("README.md") + +missing = [] +for readme in [Path("docs/technical/zh/README.md"), Path("docs/technical/en/README.md")]: + if not readme.exists(): + continue + for href in re.findall(r"\]\(([^)]+\.md)\)", readme.read_text()): + path = Path(href) + if "docs/technical/" not in href: + continue + filename = path.name + if filename not in known: + missing.append(f"{readme}: {filename}") + +if missing: + raise SystemExit("docs README links missing DOCS_METADATA: " + ", ".join(missing)) +print("docs README links are whitelisted") +PY +``` + +### Public Bilingual Pairs + +```bash +python - <<'PY' +import re +from pathlib import Path + +metadata = Path("frontend/src/pages/Docs/docs-content.ts").read_text() +filenames = sorted(set(re.findall(r"'([^']+\.md)':\s*\{", metadata)) - {"README.md"}) +missing = [] +for filename in filenames: + for lang in ("zh", "en"): + path = Path("docs/technical") / lang / filename + if not path.exists(): + missing.append(str(path)) +if missing: + raise SystemExit("missing bilingual docs: " + ", ".join(missing)) +print("public docs have zh/en file pairs") +PY +``` + +### Raw Filename Link Titles + +```bash +rg -n "\[[^]]+\.md\]\(" docs/technical/zh docs/technical/en +``` + +This should return no matches for polished public docs. diff --git a/docs/plans/earth-vessel-ais-aggregation-plan.md b/docs/plans/earth-vessel-ais-aggregation-plan.md index 2728ae33..aa877198 100644 --- a/docs/plans/earth-vessel-ais-aggregation-plan.md +++ b/docs/plans/earth-vessel-ais-aggregation-plan.md @@ -1,6 +1,6 @@ # AIS 多源采集、冲突记录与聚合接口计划 -**状态**：规划中 +**状态**：v0-v3 已实现，v4+ 规划中 **创建日期**：2026-04-30 **核心原则**：采集器只写原始观测；去重、合并、冲突解释放在聚合接口中完成 @@ -14,6 +14,8 @@ | 冲突处理 | 先记录冲突事实和当前选择原因，后续再开放用户规则配置 | | 默认可信度 | 同类 AIS 数据源优先按 `delivery_mode` 评估：`realtime_stream` 优于 `batch_stream`，再优于 `polling` 和 `snapshot` | | 过期保护 | 实时流源断流超过 freshness 窗口后，不能仅凭“实时源”身份压过更新的轮询数据 | +| 源健康状态 | 聚合时必须参考采集器健康状态，不能只看配置中的理论优先级 | +| 媒体富化 | 船只图片等媒体信息不进入 AIS 实时聚合主链路，后续单独做 enrichment | ## 背景 @@ -24,7 +26,7 @@ - WebSocket 或其他实时流通常更接近实时，但也可能断流或批量延迟。 - 如果每个 collector 自己做去重合并，规则会分散、不可审计，也很难让用户后续配置“某个字段信任哪个来源”。 -因此 v1 不应让采集器直接覆盖最终船只表。更稳的方式是先保留观测事实，再由聚合接口统一给出当前展示视图。 +因此第一阶段不应让采集器直接覆盖最终船只表。更稳的方式是先保留观测事实，再由聚合接口统一给出当前展示视图。 ## 目标架构 @@ -53,11 +55,36 @@ flowchart LR | `transport` | `websocket`、`sse`、`http`、`file` 等 | | `observed_at` | 上游数据时间，优先使用 AIS 消息时间 | | `collected_at` | 本系统接收或采集时间 | +| `source_message_id` | 上游消息 ID 或可推导 ID，没有则为空 | +| `observation_hash` | 幂等去重指纹，用于防止同一来源重复写入同一条观测 | | `normalized_payload` | 标准化后的 AIS JSON | | `raw_payload` | 可选，保存原始或裁剪后的上游记录 | +| `quality_flags` | 观测级质量标记，例如 `stale`、`position_jump`、`future_timestamp` | `delivery_mode` 和 `transport` 不应混为一谈。WebSocket 是传输方式；streaming 是交付模式。聚合可信度主要看 `delivery_mode`，`transport` 只作为辅助信息。 +原始观测层需要做存储级幂等去重，但这里的去重不是业务合并。推荐使用 `source + entity_key + message_type + observed_at + payload_hash` 或上游稳定消息 ID 作为唯一约束，避免 WebSocket 重连、HTTP 重试或批量回放导致同一事实重复入库。 + +### 源健康状态 + +每个采集器应维护独立的健康状态，供聚合服务读取： + +| 字段 | 用途 | +|-----|------| +| `source` | 采集器标识 | +| `connection_state` | `connected`、`reconnecting`、`disconnected`、`disabled` 等 | +| `last_seen_at` | 最近收到上游消息或响应的时间 | +| `last_success_at` | 最近成功写入观测的时间 | +| `last_error` | 最近错误摘要 | +| `message_rate` | 最近窗口内的消息速率 | +| `lag_seconds` | 上游观测时间与本系统接收时间的延迟 | + +聚合优先级不能只看 `source_priority`。例如 `aisstream_vessels` 默认优先于 `barentswatch_vessels`，但如果它处于 `disconnected` 或 `lag_seconds` 超过 freshness 窗口，则动态字段应回退到更新的可用来源。 + +### 身份键边界 + +v1 可以继续用 MMSI 作为 `entity_key`，因为它是 AIS 动态消息里最稳定、最容易获得的主键。但文档和模型都要为后续扩展留出口：MMSI 可能复用、填错或缺少静态信息，后续身份解析应结合 `mmsi + imo + callsign + name + dimensions` 判断是否需要拆分或合并实体。 + ### 冲突记录层 聚合服务发现同一个实体、同一个字段存在多个非空不同值时，写入冲突记录。冲突记录不代表错误，只代表“有多个可用候选值”。 @@ -90,7 +117,8 @@ flowchart LR | 动态位置 | `lat`、`lon`、`sog`、`cog`、`heading`、`nav_status` | 优先最新 `observed_at`，同时间再按来源优先级 | | 静态身份 | `name`、`callsign`、`imo`、`flag` | 非空优先，再按字段策略或来源优先级 | | 静态规格 | `vessel_type`、`vessel_type_name`、`length`、`width`、`draught` | 非空优先；冲突时记录候选值 | -| 元信息 | `field_sources`、`conflict_count`、`selected_reasons` | 聚合接口生成，便于调试和后续 UI 展示 | +| 轨迹点 | `track_points` | 按时间线合并；同一时间窗口内相近点去重；保留点级 `source` | +| 元信息 | `field_sources`、`conflict_count`、`selected_reasons`、`quality_flags` | 聚合接口生成，便于调试和后续 UI 展示 | ### 默认优先级 @@ -124,6 +152,27 @@ freshness: 如果 `aisstream_vessels` 最近 15 分钟没有该 MMSI 的新观测，而 BarentsWatch 轮询源有更新位置，则位置类字段应采用 BarentsWatch 的更新观测，并记录选择原因 `newest_observation` 或 `freshness_fallback`。 +### 异常位置保护 + +多源 AIS 接入后，聚合服务必须过滤或降权明显异常的位置观测： + +- 经纬度必须在合法范围内。 +- `observed_at` 不能明显来自未来。 +- 同一 MMSI 短时间内跨越不合理距离时，标记 `position_jump`，默认不直接采用该点。 +- 当异常点来自当前优先源时，应记录 `selected_reason = anomaly_rejected`，再回退到其他可用来源。 + +异常保护不应静默丢弃事实。原始观测仍应保留，聚合结果通过 `quality_flags` 和冲突记录解释为什么没有采用它。 + +### 轨迹聚合 + +轨迹接口不能简单拼接所有来源，否则前端会出现折返、抖动和重复点。默认规则： + +- 以 `observed_at` 排序，生成统一时间线。 +- 同一来源的完全重复点通过 `observation_hash` 去重。 +- 多来源在短时间窗口内上报的相近位置视为同一轨迹点，优先选择 freshness 和 source priority 更高的一条。 +- 每个轨迹点保留 `source`、`selected_reason` 和必要的 `quality_flags`。 +- 对被判定为 `position_jump` 的点，默认不进入展示轨迹，但可通过调试参数查看。 + ## 聚合接口 现有展示接口应逐步改为消费聚合服务，而不是自己直接拼 `VesselPosition + VesselStatic`。 @@ -155,6 +204,7 @@ GeoJSON properties 建议增加： "lat": "newest_observation", "vessel_type": "non_empty_priority" }, + "quality_flags": [], "conflict_count": 2 } ``` @@ -231,15 +281,89 @@ aisstream_vessels: - ShipStaticData ``` -## 实施顺序 +默认不建议直接订阅全球范围。AISStream 采集器应支持以下订阅策略： -1. 新增原始观测模型和冲突记录模型。 -2. 实现 AIS 聚合服务，先从现有 `vessel_position` / `vessel_static` 兼容读取，再逐步切换到原始观测层。 -3. 将 `/geo/vessels` 和 `/vessels/{mmsi}` 改为走聚合服务。 -4. 改造 BarentsWatch 保存逻辑，让它写入原始观测，同时保留现有表作为兼容缓存。 -5. 实现 AISStream WebSocket collector。 -6. 接入系统设置中的聚合策略配置。 -7. 做冲突治理 UI。 +- 使用配置的固定 `bounding_boxes`。 +- 后续支持按 Earth 当前视口或关注区域动态调整订阅范围。 +- 支持限制 `message_types`，避免静态信息、位置报告和扩展消息全量涌入。 +- 断线后使用指数退避重连，并把连接状态写入源健康状态。 +- 重连后可能收到重复或回放消息，因此必须依赖原始观测层的幂等去重。 + +### 媒体富化边界 + +VesselFinder 等服务里的船只图片不属于 AIS 实时数据本身。图片、船籍详情、公司信息等后续应作为独立 enrichment 链路： + +- 通过 MMSI、IMO、船名等字段异步查询。 +- 使用独立缓存和授权配置。 +- 不阻塞 `vessel_ais` 实时观测入库。 +- 聚合接口只暴露已经缓存好的媒体引用，不在请求链路中现场抓取。 + +## 版本拆分 + +计划按 5 个版本推进： + +### v0 — 聚合基础设施（已实现） + +目标是不改变前端展示行为，先把数据底座铺好。 + +1. 新增原始观测模型、冲突记录模型和源健康状态模型。 +2. 为现有 BarentsWatch collector 写入原始观测，同时保留现有 `vessel_position` / `vessel_static` 兼容写入。 +3. 实现存储级 `observation_hash` 幂等去重。 +4. 补基础管理命令或调试接口，用于查看某个 MMSI 的原始观测和冲突候选。 + +### v1 — 聚合读接口（已实现） + +目标是让展示接口开始消费聚合结果，但前端形状保持兼容。 + +1. 实现 AIS 聚合服务，先兼容读取现有表，再逐步切换到原始观测层。 +2. 将 `/geo/vessels` 和 `/vessels/{mmsi}` 改为走聚合服务。 +3. 将 `/vessels/{mmsi}/track` 改为走轨迹聚合逻辑。 +4. 返回 `field_sources`、`selected_reasons`、`quality_flags`、`conflict_count`。 +5. 加入 freshness fallback 和异常位置保护。 + +### v2 — AISStream WebSocket collector（已实现） + +目标是接入第二个真实 AIS 来源，并验证多源冲突和回退逻辑。 + +1. 实现 `aisstream_vessels` collector。 +2. 支持 API key、订阅范围、消息类型、重连和限流配置。 +3. 将 AISStream 写入原始观测层，不直接 upsert 最终展示表。 +4. 接入源健康状态和 message rate 统计。 +5. 提供 AISStream API Key 获取教程、设置页入口和连接验证支持。 +6. 为重复消息、断流回退、WS 优先级写集成测试。 + +### v3 — AISStream 可用性与配置体验（已实现） + +目标是让 AISStream 从“能采集”变成日常可观察、可调试、可配置的数据源。 + +1. 设置页展示 AISStream 运行状态：连接状态、最近收到、最近成功、本轮消息数、延迟和最近错误。 +2. AISStream 设置页提供常用采集范围 preset，并保留自定义 Bounding Boxes JSON。 +3. 聚合结果返回 `source_summary`，展示每艘船的来源、观测数量、最新观测时间、传输模式和消息类型。 +4. 保留 `field_sources` 和 `selected_reasons`，用于解释动态字段来自实时流、静态字段来自可用非空来源。 +5. 船名标准化会读取 AISStream `MetaData.ShipName`；船型展示会从 `vessel_type_name` 和 AIS 数字 `vessel_type` 共同归一化，保证 marker 颜色、详情卡、hover 和搜索结果一致。 +6. `/geo/vessels` 不再默认限制 5000 艘；不传 `limit` 或传 `limit=0` 表示全量返回，前端默认也不再二次裁剪到 5000。 + +### v4 — 策略配置 + +目标是开放系统级配置，但仍以安全默认值兜底。 + +1. 接入系统设置中的聚合策略配置。 +2. 支持 source priority、字段级规则、freshness 窗口和高级保护开关。 +3. 保存配置时校验未知字段、非法模式和危险动态字段锁定。 +4. 聚合接口返回当前命中的配置版本，方便排查。 + +### v5 — 船舶资料 enrichment 与冲突治理 + +目标是把 AIS 实时流里不稳定或低频出现的静态信息，补成可缓存、可审计的船舶资料层，同时把冲突解释变成可操作能力。 + +1. 做冲突治理 UI。 +2. 支持把人工选择沉淀成字段级规则。 +3. 支持恢复默认策略。 +4. 设计 `vessel_profile_enrichment`，按 `mmsi + imo + name + callsign` 异步补充船名、船型细分、AIS 大类、旗国、尺寸、建造年份、运营方等静态资料。 +5. 设计 `vessel_media_enrichment`，异步补充船只图片和外部详情缓存。 +6. enrichment 结果必须带 `source`、`fetched_at`、`expires_at`、`confidence` 和原始引用，不覆盖 AIS 原始观测。 +7. 聚合接口只读取已缓存 enrichment；请求链路不现场抓取第三方页面，避免慢请求和授权风险。 +8. 前端船只详情面板展示已缓存资料和媒体，并标注字段来源，不阻塞 AIS 实时链路。 ## 测试计划 @@ -247,8 +371,12 @@ aisstream_vessels: - 多来源同一 MMSI 的位置字段优先选择最新观测。 - 实时流和轮询源同时间冲突时，实时流优先。 - 实时流过期后，更新的轮询源可以接管动态字段。 +- 实时流源健康状态异常时，动态字段可以回退到更新的可用来源。 - 静态字段不会被空值覆盖。 - 静态字段冲突会写入冲突记录。 +- 明显异常位置不会进入默认展示轨迹，并会留下 `quality_flags`。 +- 同一时间窗口内多来源相近轨迹点只展示一个点。 +- AISStream 重连或回放导致的重复消息不会重复进入聚合结果。 - 字段级配置可以覆盖默认来源优先级。 - 聚合接口在没有冲突表时仍可返回兼容 GeoJSON。 diff --git a/docs/plans/earth-vessel-tracking-plan.md b/docs/plans/earth-vessel-tracking-plan.md index 24c31dd1..e1ccfc05 100644 --- a/docs/plans/earth-vessel-tracking-plan.md +++ b/docs/plans/earth-vessel-tracking-plan.md @@ -124,7 +124,7 @@ CREATE UNIQUE INDEX ON vessel_latest(mmsi); GET /api/v1/visualization/geo/vessels ?bbox=lon_min,lat_min,lon_max,lat_max # 视口裁剪 ?type=cargo,tanker,passenger # 船型过滤 - ?limit=5000 + ?limit=0 # 可选；不传或 0 表示不裁剪数量 → GeoJSON FeatureCollection（Point） GET /api/v1/visualization/vessels/{mmsi} # 单船详情 @@ -163,6 +163,8 @@ GeoJSON Feature 格式： - 后端 BarentsWatch collector 继续以 HTTP polling 方式采集 - AISStream 等实时源以独立 WebSocket collector 写入原始观测层 - 展示接口从聚合服务读取当前船只视图，而不是由单个 collector 决定最终展示值 +- 前端默认不再给 `/geo/vessels` 传 `limit=5000`，`VESSEL_CONFIG.maxRenderedMarkers = 0` 表示不做前端数量裁剪；后续如性能不足再引入显式 LOD 上限 +- marker 颜色、详情卡、hover 和搜索结果必须共享 `vessel_type_display` 船型归一化结果，避免 AIS 数字类型码已驱动颜色但卡片仍显示 `Other` - 前端是否升级为 WebSocket delta push 是独立优化，不影响后端采集器可以使用 WebSocket 接上游实时源 --- @@ -196,8 +198,8 @@ GeoJSON Feature 格式： | 相机距离 | 渲染策略 | |---------|---------| -| > 400 | 仅渲染 top 1000 艘（按数据新鲜度 + 船型优先级） | -| 200–400 | 渲染 top 5000 艘 | +| > 400 | 默认渲染当前接口返回的全部船只；如性能不足，再引入可配置 LOD 上限 | +| 200–400 | 默认渲染当前接口返回的全部船只；如性能不足，再引入可配置 LOD 上限 | | < 200 | 渲染当前视口 bbox 内全部船只 | 前端根据相机位置动态计算 bbox，附加到 API 请求中。 diff --git a/docs/technical/en/backend-collectors.md b/docs/technical/en/backend-collectors.md index bf0504f9..880d5360 100644 --- a/docs/technical/en/backend-collectors.md +++ b/docs/technical/en/backend-collectors.md @@ -84,6 +84,8 @@ async def run(self, db): | HuggingFace Spaces | space | Demo applications | 1 day | | PeeringDB | ixp/network/facility | Internet exchange points / networks / facilities | 1-2 days | | TeleGeography | submarine_cable | Submarine cable information | 7 days | +| BarentsWatch AIS | vessel | AIS vessel positions, speed, heading, MMSI, and related fields | Collector settings | +| AISStream Vessels | vessel_ais | AIS WebSocket realtime stream, written to the raw observation layer and displayed through aggregation | Collector settings | ## IV. Data Format (stored in CollectedData table) diff --git a/docs/technical/en/datasource-collector-settings-connectivity.md b/docs/technical/en/datasource-collector-settings-connectivity.md index 45a27bb5..27c83619 100644 --- a/docs/technical/en/datasource-collector-settings-connectivity.md +++ b/docs/technical/en/datasource-collector-settings-connectivity.md @@ -260,6 +260,30 @@ AIS request rules: `VesselAISCollector` no longer reads environment variables directly. It goes through `resolve_barentswatch_config()` and `fetch_barentswatch_access_token()` so settings, connectivity validation, and collection do not fork into three credential flows. +## AISStream Collector Chain + +Files: + +- [aisstream.py](/home/ray/dev/linkong/planet/backend/app/services/collectors/aisstream.py) +- [vessel_ais_aggregation.py](/home/ray/dev/linkong/planet/backend/app/services/vessel_ais_aggregation.py) + +AISStream uses a WebSocket realtime stream. The collector writes only to the `ais_raw_observations` raw observation layer; it does not directly overwrite the final vessel display table. The aggregation API handles multi-source deduplication, field selection, and conflict records. + +Configuration: + +- `api_key`: stored in `DataSourceConfig.auth_config`, or provided through `AISSTREAM_API_KEY`. +- `endpoint`: defaults to `wss://stream.aisstream.io/v0/stream`. +- `message_types`: defaults to `PositionReport` and `ShipStaticData`. +- `bounding_boxes`: AISStream format is `[[[lat_min, lon_min], [lat_max, lon_max]]]`; the settings page provides global, Norway / North Sea, Europe coast, East Asia, and North America coast presets. +- `max_messages` and `receive_timeout_seconds`: control the batch-style WebSocket collection window. + +Normalization: + +- `PositionReport` mainly provides position, speed, course, heading, and navigation status. +- Vessel names can be filled from `MetaData.ShipName` even when the message body has no `name`. +- Vessel type usually comes from lower-frequency `ShipStaticData.Type`; the backend maps AIS numeric type codes to Cargo / Tanker / Passenger / Fishing / Military. +- If a vessel has not yet produced a static message, its aggregated type can still be `Other`; v5 vessel profile enrichment is planned to fill that gap. + ## Credential Guide File: @@ -277,6 +301,7 @@ POST /api/v1/settings/credential-guides/{provider}/reset Currently supported: - `barentswatch` +- `aisstream` The default guide includes the official BarentsWatch tutorial: diff --git a/docs/technical/en/earth-frontend-context.md b/docs/technical/en/earth-frontend-context.md index 7ac81986..785a3cbc 100644 --- a/docs/technical/en/earth-frontend-context.md +++ b/docs/technical/en/earth-frontend-context.md @@ -96,6 +96,7 @@ Responsibilities: - [satellites.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/satellites.js) - [cables.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/cables.js) +- [vessels.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/vessels.js) - [bgp.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/bgp.js) - [bgp-cruise-adapter.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/bgp-cruise-adapter.js) - [compute-centers.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/compute-centers.js) @@ -108,6 +109,14 @@ Each module is responsible for its own: - State tracking (loaded, visible, hover, locked) - Self-cleanup (dispose on scene destroy) +### AIS Vessel Layer + +The vessel layer fetches `/api/v1/visualization/geo/vessels` and renders the aggregated AIS GeoJSON through `createInteractableLayer()`. By default it does not send a `limit` parameter, and `VESSEL_CONFIG.maxRenderedMarkers = 0` means the frontend does not clip the result to 5000 vessels. A positive `options.limit` or positive `maxRenderedMarkers` can still be used as an explicit temporary cap. + +Vessel color and vessel type text must use the same normalized classification. `vessels.js` derives `type` from both `vessel_type_name` and the AIS numeric `vessel_type` code; that `type` drives marker color. It also derives `vessel_type_display`, which `main.js` uses for the info card, hover summary, and search result subtitle. Do not make the info card read only the raw `vessel_type_name`, because AISStream can provide a numeric type while the raw name is still `Other`. + +AISStream `PositionReport` messages commonly carry live position and `MetaData.ShipName`, while vessel type usually comes from lower-frequency `ShipStaticData.Type`. The backend normalizes `MetaData.ShipName` into the vessel name and maps numeric type codes into Cargo / Tanker / Passenger / Fishing / Military where available. Missing type detail should wait for a static AIS message or the planned vessel profile enrichment; the frontend should not invent a more specific type. + ### 7. HUD Panels and Search - [hud-panels.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/hud-panels.js) diff --git a/docs/technical/en/earth-layer-style-reference.md b/docs/technical/en/earth-layer-style-reference.md index 7d333bbd..d7b984ee 100644 --- a/docs/technical/en/earth-layer-style-reference.md +++ b/docs/technical/en/earth-layer-style-reference.md @@ -182,11 +182,14 @@ The land/ocean base is an Earth base-map asset and preloads at startup; the "Bor | Vessel renderOrder | local `VESSEL_RENDER_ORDER` | `4.4` | Normal marker and interactive overlay | | Vessel track renderOrder | `VESSEL_RENDER_ORDER - 0.1` | `4.3` | Below vessel markers | | Vessel point pixel size | local `VESSEL_POINT_SIZE` | `34` | Shared size for normal markers and hover / locked overlays | +| Default vessel render cap | `VESSEL_CONFIG.maxRenderedMarkers` | `0` | `0` means the frontend does not clip by default; positive values send `limit` and clip markers | | Vessel texture canvas size | local `VESSEL_ATLAS_CELL_SIZE` | `128` | Canvas point texture | | Course bucket count | local `VESSEL_COURSE_BINS` | `32` | Moving vessels are bucketed by COG to reduce draw calls while preserving direction | | Vessel hover picking throttle | local `VESSEL_HOVER_PICK_INTERVAL_MS` | `100` | `main.js` hover picking | | Vessel screen hit radius | local `VESSEL_POINTER_RADIUS_PX` | `22` | `main.js` screen-space picking | +AIS vessel markers use batched `THREE.Points`, not one `THREE.Sprite` per vessel. Moving vessels stay triangular, anchored or slow vessels stay circular, and hover / locked states add a same-size glow overlay. Vessel type color and info-card type text must come from the same normalized result: `vessels.js` reads both backend `vessel_type_name` and AIS numeric `vessel_type`, derives the color-driving `type`, then exposes `vessel_type_display` for the info card, hover summary, and search results. + ## Compute Centers | Name | Variable | Current Value | Location / Notes | diff --git a/docs/technical/zh/backend-collectors.md b/docs/technical/zh/backend-collectors.md index ad5ebee1..c6838acf 100644 --- a/docs/technical/zh/backend-collectors.md +++ b/docs/technical/zh/backend-collectors.md @@ -86,6 +86,7 @@ async def run(self, db): | TeleGeography | submarine_cable | 海底光缆信息 | 7天 | | Space-Track TLE | satellite_tle | 卫星轨道 TLE 数据 | 依采集器配置 | | BarentsWatch AIS | vessel | 船只位置、航速、航向、MMSI 等 AIS 数据 | 依采集器配置 | +| AISStream Vessels | vessel_ais | AIS WebSocket 实时流，写入原始观测层并由聚合接口展示 | 依采集器配置 | ## 四、数据格式 (统一存储到 CollectedData 表) @@ -238,7 +239,8 @@ backend/app/services/collectors/ ├── huggingface.py # HuggingFace采集器 ├── peeringdb.py # PeeringDB采集器 ├── telegeraphy.py # TeleGeography海底光缆采集器 -└── vessel_ais.py # BarentsWatch AIS 船只采集器 +├── vessel_ais.py # BarentsWatch AIS 船只采集器 +└── aisstream.py # AISStream WebSocket 船只采集器 backend/app/models/ └── collected_data.py # 统一数据模型 @@ -251,6 +253,7 @@ backend/app/models/ | 采集器 | credential provider | 凭证来源 | | --- | --- | --- | | `barentswatch_vessels` | `barentswatch` | 控制台采集器设置、环境变量、`~/.zshrc` | +| `aisstream_vessels` | `aisstream` | 控制台采集器设置、环境变量 | | `spacetrack_tle` | `spacetrack` | 环境变量、`~/.zshrc` | ### BarentsWatch AIS diff --git a/docs/technical/zh/datasource-collector-settings-connectivity.md b/docs/technical/zh/datasource-collector-settings-connectivity.md index 0a98370a..7d6841da 100644 --- a/docs/technical/zh/datasource-collector-settings-connectivity.md +++ b/docs/technical/zh/datasource-collector-settings-connectivity.md @@ -262,6 +262,30 @@ AIS 请求规则： `VesselAISCollector` 不再自己读取环境变量，而是统一走 `resolve_barentswatch_config()` 和 `fetch_barentswatch_access_token()`，避免设置页、连接验证和采集器三套凭证逻辑分叉。 +## AISStream 采集器链路 + +文件： + +- [aisstream.py](/home/ray/dev/linkong/planet/backend/app/services/collectors/aisstream.py) +- [vessel_ais_aggregation.py](/home/ray/dev/linkong/planet/backend/app/services/vessel_ais_aggregation.py) + +AISStream 使用 WebSocket 实时流，采集器只写入 `ais_raw_observations` 原始观测层，不直接覆盖最终船只展示表。聚合接口负责多源去重、字段选择和冲突记录。 + +配置项： + +- `api_key`：保存在 `DataSourceConfig.auth_config`，也可用环境变量 `AISSTREAM_API_KEY`。 +- `endpoint`：默认 `wss://stream.aisstream.io/v0/stream`。 +- `message_types`：默认 `PositionReport` 和 `ShipStaticData`。 +- `bounding_boxes`：AISStream 格式为 `[[[lat_min, lon_min], [lat_max, lon_max]]]`，设置页提供全球、挪威 / 北海、欧洲近海、东亚、北美东西海岸 preset。 +- `max_messages` 和 `receive_timeout_seconds`：控制单次批次式 WebSocket 采集窗口。 + +标准化规则： + +- `PositionReport` 主要提供位置、速度、航向和状态。 +- 船名可以从 `MetaData.ShipName` 补入，即使消息体本身没有 `name`。 +- 船型通常来自低频 `ShipStaticData.Type`；后端会把 AIS 数字类型码映射为 Cargo / Tanker / Passenger / Fishing / Military。 +- 如果某艘船尚未收到静态消息，聚合结果的船型仍可能是 `Other`，后续由 v5 船舶资料 enrichment 补齐。 + ## 凭证教程 文件： @@ -279,6 +303,7 @@ POST /api/v1/settings/credential-guides/{provider}/reset 当前支持： - `barentswatch` +- `aisstream` 默认教程包含 BarentsWatch 官方 tutorial 地址： diff --git a/docs/technical/zh/earth-frontend-context.md b/docs/technical/zh/earth-frontend-context.md index a3e898b0..0ff60646 100644 --- a/docs/technical/zh/earth-frontend-context.md +++ b/docs/technical/zh/earth-frontend-context.md @@ -254,10 +254,10 @@ AIS 船只图层入口： 船只图层当前负责： - 请求 `/api/v1/visualization/geo/vessels` -- 将 BarentsWatch AIS GeoJSON 转为地球局部坐标 marker 数据 +- 将聚合后的 AIS GeoJSON 转为地球局部坐标 marker 数据；请求默认不传 `limit`，后端和前端都不再默认裁剪到 5000 艘 - 通过 `createInteractableLayer()` 注册 Interactable 图标层 - 用按航向分桶的 `THREE.Points` 批量渲染普通船只 marker -- 按船型映射颜色 +- 按船型映射颜色；`vessels.js` 会用 `vessel_type_name` 和 AIS `vessel_type` 数字共同归一化船型 - 根据航行/停泊状态绘制三角形或圆点纹理 - 用单点 `THREE.Points` overlay 承载 hover / locked glow - 支持 hover、lock、轨迹加载和视觉聚焦 @@ -273,6 +273,10 @@ AIS 船只图层入口： 方向标准以 AIS `course / cog` 为准：从正北开始顺时针。普通态和交互态都通过同一套 canvas 旋转规则生成纹理，避免 hover 后箭头方向和原 marker 不一致。 +船型展示也必须复用同一套归一化结果。`buildVesselMarkerData()` 会把后端的 `vessel_type_name` 和 AIS 数字类型码归一化为 `type`，用于 marker 颜色；同时生成 `vessel_type_display`，供详情卡、hover 简述和搜索结果显示。不要让详情卡直接只读原始 `vessel_type_name`，否则会出现 marker 已按 Cargo/Tanker 等颜色显示、卡片仍写 `Other` 的不一致。 + +AISStream 的 `PositionReport` 常带实时位置和 `MetaData.ShipName`，但船型通常来自低频 `ShipStaticData.Type`。后端会把 `MetaData.ShipName` 补进船名，并将类型码映射为 Cargo / Tanker / Passenger / Fishing / Military；仍缺失的船型需要等待静态 AIS 消息或后续船舶资料 enrichment，不能在前端凭颜色之外的信息臆造细分类。 + 船只 hover / click 也不再对渲染对象做 `raycaster.intersectObjects()`。`main.js` 只负责传入当前 Earth、camera、pointer 和命中半径，实际命中计算由 `interactable.js` 的图标层接口完成： 1. 拖动地球或惯性旋转时跳过 hover picking。 diff --git a/docs/technical/zh/earth-layer-style-reference.md b/docs/technical/zh/earth-layer-style-reference.md index 8c47186b..05bb27a6 100644 --- a/docs/technical/zh/earth-layer-style-reference.md +++ b/docs/technical/zh/earth-layer-style-reference.md @@ -194,6 +194,7 @@ | 船只 renderOrder | local `VESSEL_RENDER_ORDER` | `4.4` | 普通 marker 和交互 overlay | | 船只轨迹 renderOrder | `VESSEL_RENDER_ORDER - 0.1` | `4.3` | 低于船只 marker | | 船只点像素尺寸 | local `VESSEL_POINT_SIZE` | `34` | 普通 marker 与 hover / locked overlay 共享尺寸 | +| 船只默认渲染上限 | `VESSEL_CONFIG.maxRenderedMarkers` | `0` | `0` 表示不在前端默认裁剪；正数才会给接口传 `limit` 并裁剪 marker | | 船只纹理画布尺寸 | local `VESSEL_ATLAS_CELL_SIZE` | `128` | canvas 点纹理 | | 航向分桶数 | local `VESSEL_COURSE_BINS` | `32` | moving 船只按 COG 分桶，降低 draw call 同时保留方向 | | 船只 hover 拾取节流 | local `VESSEL_HOVER_PICK_INTERVAL_MS` | `100` | `main.js` hover picking | @@ -206,6 +207,8 @@ AIS 船只普通态使用批量 `THREE.Points`，不是逐船 `THREE.Sprite`。航行船只保持三角形，停泊或低速船只保持圆点；普通态不带 glow，hover / locked 时在同一屏幕尺寸上叠加带 glow 的单点 overlay。AIS 航向按 `course / cog` 从正北顺时针解释，普通态和交互态必须使用同一套 canvas 旋转规则。 +船型颜色和详情卡船型文本必须来自同一套归一化结果：`vessels.js` 同时读取后端 `vessel_type_name` 和 AIS 数字 `vessel_type`，先得到颜色用的 `type`，再生成 `vessel_type_display` 给详情卡、hover 和搜索使用。 + ## 算力中心 | 正式名称 | 变量名 | 当前值 | 使用位置 / 说明 | diff --git a/docs/version-history.md b/docs/version-history.md index 1f5c028e..8b69c42e 100644 --- a/docs/version-history.md +++ b/docs/version-history.md @@ -16,12 +16,13 @@ ## Current Version - `main` 当前主线历史推导到：`0.16.5` -- `dev` 当前开发分支历史推导到：`0.46.3` +- `dev` 当前开发分支历史推导到：`0.47.0` ## Timeline | Version | Type | Branch | Commit | Summary | | --- | --- | --- | --- | --- | +| `0.47.0` | feature | `dev` | `pending` | 新增 AISStream WebSocket 船只采集器、多源 AIS 原始观测聚合、采集器状态配置、船型显示修正和文档规则解耦 | | `0.46.3` | bugfix | `dev` | `pending` | 优化 Starlink footprint 拖拽性能，避免旋转地球时重复重建覆盖网格，并恢复线缆点击呼吸动画 | | `0.46.2` | bugfix | `dev` | `pending` | 修复 Earth 启动加载顺序、图层 localStorage 恢复、国界线底图语义、媒体面板、船只轨迹和 Iridium footprint 显示问题，并补充 AIS 聚合计划 | | `0.46.1` | bugfix | `dev` | `pending` | 修复新增 Docs 技术文档未进前端白名单导致页面不可访问的问题，补齐英文文档并固化白名单/双语/裸文件标题检查 | diff --git a/frontend/package.json b/frontend/package.json index df8a7025..7a7728fa 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "planet-frontend", - "version": "0.46.3", + "version": "0.47.0", "private": true, "packageManager": "bun@1", "dependencies": { diff --git a/frontend/public/earth/js/constants.js b/frontend/public/earth/js/constants.js index d4d7c30f..6e56df54 100644 --- a/frontend/public/earth/js/constants.js +++ b/frontend/public/earth/js/constants.js @@ -210,7 +210,7 @@ export const PATHS = { export const VESSEL_CONFIG = { altitudeOffset: 0.2, - maxRenderedMarkers: 5000, + maxRenderedMarkers: 0, marker: { baseScale: 7.5, baseOpacity: 0.88, diff --git a/frontend/public/earth/js/main.js b/frontend/public/earth/js/main.js index fb1814a9..e7db7ec2 100644 --- a/frontend/public/earth/js/main.js +++ b/frontend/public/earth/js/main.js @@ -787,12 +787,13 @@ function formatVesselStatus(navStatus) { function showVesselInfo(marker, coords) { setLegendMode("vessels"); + const vesselType = marker.userData?.vessel_type_display || marker.userData?.vessel_type_name || "-"; showInfoCard("vessel", { name: marker.userData?.name || `MMSI ${marker.userData?.mmsi}`, mmsi: marker.userData?.mmsi, imo: marker.userData?.imo || "-", flag: marker.userData?.flag || "-", - vessel_type: marker.userData?.vessel_type_name || "-", + vessel_type: vesselType, speed: marker.userData?.sog ?? "-", course: marker.userData?.cog ?? marker.userData?.heading ?? "-", status: formatVesselStatus(marker.userData?.nav_status), @@ -806,7 +807,8 @@ function showVesselInfo(marker, coords) { function getVesselBriefHtml(marker) { const name = marker.userData?.name || `MMSI ${marker.userData?.mmsi}`; const speed = marker.userData?.sog ?? "-"; - return `${name}
${marker.userData?.vessel_type_name || "Vessel"} · ${speed} kn`; + const vesselType = marker.userData?.vessel_type_display || marker.userData?.vessel_type_name || "Vessel"; + return `${name}
${vesselType} · ${speed} kn`; } function getComputeCenterBriefHtml(marker) { @@ -1390,6 +1392,7 @@ function resolveEarthSearchResults(query) { marker.userData?.mmsi, marker.userData?.imo, marker.userData?.flag, + marker.userData?.vessel_type_display, marker.userData?.vessel_type_name, "船只 船舶 ais vessel ship maritime", ); @@ -1401,7 +1404,7 @@ function resolveEarthSearchResults(query) { typeLabel: "船只", title: marker.userData?.name || `MMSI ${marker.userData?.mmsi}`, subtitle: [ - marker.userData?.vessel_type_name, + marker.userData?.vessel_type_display || marker.userData?.vessel_type_name, marker.userData?.flag, marker.userData?.sog !== undefined ? `${marker.userData.sog} kn` : null, ].filter(Boolean).join(" · ") || "AIS 船只", diff --git a/frontend/public/earth/js/vessels.js b/frontend/public/earth/js/vessels.js index 3f68681c..fa2974e6 100644 --- a/frontend/public/earth/js/vessels.js +++ b/frontend/public/earth/js/vessels.js @@ -24,6 +24,17 @@ function normalizeVesselType(value, code) { return "other"; } +function formatVesselTypeLabel(type, fallback = "") { + const rawFallback = String(fallback || "").trim(); + const normalized = String(type || "").trim().toLowerCase(); + if (normalized === "cargo") return "Cargo"; + if (normalized === "tanker") return "Tanker"; + if (normalized === "passenger") return "Passenger"; + if (normalized === "fishing") return "Fishing"; + if (normalized === "military") return "Military"; + return rawFallback && rawFallback.toLowerCase() !== "other" ? rawFallback : "Other"; +} + function drawVesselShape(context, anchored, glow, color = "#ffffff") { context.fillStyle = color; context.globalAlpha = anchored ? 0.55 : 0.96; @@ -52,6 +63,7 @@ function buildVesselMarkerData(feature) { if (!Number.isFinite(latitude) || !Number.isFinite(longitude)) return null; const type = normalizeVesselType(props.vessel_type_name, props.vessel_type); + const vesselTypeLabel = formatVesselTypeLabel(type, props.vessel_type_name); const navStatus = Number(props.nav_status); const speed = Number(props.sog); const anchored = navStatus === 1 || navStatus === 5 || (Number.isFinite(speed) && speed < 0.5); @@ -61,6 +73,7 @@ function buildVesselMarkerData(feature) { latitude, longitude, type, + vessel_type_display: vesselTypeLabel, anchored, course: Number(props.cog ?? props.heading ?? 0), }; @@ -191,7 +204,10 @@ export function clearVesselData(earth) { export async function loadVessels(_scene, earth, options = {}) { const params = new URLSearchParams(); - params.set("limit", String(options.limit || VESSEL_CONFIG.maxRenderedMarkers)); + const requestedLimit = Number(options.limit ?? VESSEL_CONFIG.maxRenderedMarkers); + if (Number.isFinite(requestedLimit) && requestedLimit > 0) { + params.set("limit", String(requestedLimit)); + } const response = await fetch(`${PATHS.vesselsApi}?${params.toString()}`); if (!response.ok) { throw new Error(`Vessels HTTP ${response.status}`); @@ -200,10 +216,12 @@ export async function loadVessels(_scene, earth, options = {}) { const features = Array.isArray(payload?.features) ? payload.features : []; clearVesselData(earth); - const markerData = features + let markerData = features .map((feature) => buildVesselMarkerData(feature)) - .filter(Boolean) - .slice(0, VESSEL_CONFIG.maxRenderedMarkers); + .filter(Boolean); + if (Number.isFinite(requestedLimit) && requestedLimit > 0) { + markerData = markerData.slice(0, requestedLimit); + } vesselIconLayer.setData(markerData); vesselIconLayer.attach(earth); diff --git a/frontend/src/pages/DataSources/DataSources.tsx b/frontend/src/pages/DataSources/DataSources.tsx index 620f1a9a..4f75f6a4 100644 --- a/frontend/src/pages/DataSources/DataSources.tsx +++ b/frontend/src/pages/DataSources/DataSources.tsx @@ -26,6 +26,7 @@ import { SyncOutlined, } from '@ant-design/icons' import axios, { type AxiosResponse } from 'axios' +import { useNavigate } from 'react-router-dom' import AppLayout from '../../components/AppLayout/AppLayout' import ScrollbarOverlay from '../../components/Scrollbar/ScrollbarOverlay' @@ -224,6 +225,7 @@ function normalizeCustom(source: CustomDataSource): UnifiedDataSource { function DataSources() { const [messageApi, contextHolder] = message.useMessage() + const navigate = useNavigate() const [modal, modalContextHolder] = Modal.useModal() const [builtInSources, setBuiltInSources] = useState([]) const [customSources, setCustomSources] = useState([]) @@ -736,6 +738,15 @@ function DataSources() { showIcon message="需要采集器凭证" description={viewingSource.credential_status === 'supported' ? '请在设置中心的采集器设置中维护该采集器凭证。' : '该采集器需要凭证，配置入口待接入。'} + action={viewingSource.credential_status === 'supported' ? ( + + ) : undefined} /> ) : null} diff --git a/frontend/src/pages/Settings/Settings.tsx b/frontend/src/pages/Settings/Settings.tsx index ed79efe9..05de53ae 100644 --- a/frontend/src/pages/Settings/Settings.tsx +++ b/frontend/src/pages/Settings/Settings.tsx @@ -82,6 +82,18 @@ interface CollectorSettings { requires_credentials?: boolean credential_provider?: string | null credential_status?: string + ais_health?: AISSourceHealth | null +} + +interface AISSourceHealth { + source: string + connection_state: string + last_seen_at: string | null + last_success_at: string | null + last_error: string | null + message_rate: number | null + lag_seconds: number | null + updated_at: string | null } interface TVStreamSource { @@ -166,12 +178,59 @@ interface CollectorConfigOption { is_active: boolean source_type: string auth_type: string + auth_config?: Record + auth_configured?: Record headers: Record config: Record config_id: number | null description: string } +const AISSTREAM_BBOX_PRESETS = [ + { + value: 'global', + label: '全球', + boxes: [[[-90, -180], [90, 180]]], + }, + { + value: 'norway_north_sea', + label: '挪威 / 北海', + boxes: [[[50, -8], [72, 32]]], + }, + { + value: 'europe_coast', + label: '欧洲近海', + boxes: [[[35, -12], [72, 32]]], + }, + { + value: 'east_asia', + label: '东亚', + boxes: [[[18, 105], [46, 146]]], + }, + { + value: 'north_america_coasts', + label: '北美东西海岸', + boxes: [ + [[24, -126], [50, -66]], + [[18, -98], [31, -80]], + ], + }, +] + +const stringifyBoundingBoxes = (boxes: unknown) => JSON.stringify(boxes, null, 2) + +const matchAisstreamBboxPreset = (boxes: unknown) => { + const serialized = JSON.stringify(boxes) + return AISSTREAM_BBOX_PRESETS.find((preset) => JSON.stringify(preset.boxes) === serialized)?.value || 'custom' +} + +const formatLagSeconds = (value: number | null | undefined) => { + if (value == null) return '未知' + if (value < 60) return `${Math.round(value)} 秒` + if (value < 3600) return `${Math.round(value / 60)} 分钟` + return `${Math.round(value / 3600)} 小时` +} + function PlugConnectIcon() { return (
+
+ 连接状态 +
+ + {selectedAisRuntimeHealth.connection_state} + +
+
+
+ 本轮消息数 +
{selectedAisRuntimeHealth.message_rate ?? '未知'}
+
+
+ 最近收到 +
{formatDateTimeZhCN(selectedAisRuntimeHealth.last_seen_at)}
+
+
+ 最近成功 +
{formatDateTimeZhCN(selectedAisRuntimeHealth.last_success_at)}
+
+
+ 数据延迟 +
{formatLagSeconds(selectedAisRuntimeHealth.lag_seconds)}
+
+
+ 状态更新时间 +
{formatDateTimeZhCN(selectedAisRuntimeHealth.updated_at)}
+
+
+ ) : ( + + )} + {selectedAisRuntimeHealth?.last_error ? ( + + ) : null} + + ) : null} +
@@ -1471,6 +1676,36 @@ function Settings() { + {selectedCollector?.source === 'aisstream_vessels' ? ( + <> +
+ + + + + + +
+ + ({ value: preset.value, label: preset.label })), + { value: 'custom', label: '自定义 JSON' }, + ]} + onChange={applyAisstreamBboxPreset} + /> + + + + + + ) : null}
diff --git a/pyproject.toml b/pyproject.toml index a6b50f0c..7838310d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,12 @@ [project] name = "planet" -version = "0.46.3" +version = "0.47.0" description = "智能星球计划 - 态势感知系统" requires-python = ">=3.14" dependencies = [ "fastapi>=0.109.0", "uvicorn[standard]>=0.27.0", + "websockets>=12.0", "sqlalchemy[asyncio]>=2.0.25", "asyncpg>=0.29.0", "redis>=5.0.1", diff --git a/uv.lock b/uv.lock index b3bcdcaa..dbfb67d0 100644 --- a/uv.lock +++ b/uv.lock @@ -475,7 +475,7 @@ wheels = [ [[package]] name = "planet" -version = "0.46.3" +version = "0.47.0" source = { virtual = "." } dependencies = [ { name = "aiofiles" }, @@ -496,6 +496,7 @@ dependencies = [ { name = "redis" }, { name = "sqlalchemy", extra = ["asyncio"] }, { name = "uvicorn", extra = ["standard"] }, + { name = "websockets" }, ] [package.dev-dependencies] @@ -526,6 +527,7 @@ requires-dist = [ { name = "redis", specifier = ">=5.0.1" }, { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.25" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.27.0" }, + { name = "websockets", specifier = ">=12.0" }, ] [package.metadata.requires-dev]