152 lines
5.3 KiB
Python
152 lines
5.3 KiB
Python
"""Registry of target schemas supported by mapped custom data sources."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, Field, ValidationError, field_validator
|
||
|
||
|
||
class VesselAISRecord(BaseModel):
|
||
mmsi: int = Field(ge=100000000, le=999999999)
|
||
lat: float = Field(ge=-90, le=90)
|
||
lon: float = Field(ge=-180, le=180)
|
||
sog: float | None = None
|
||
cog: float | None = Field(default=None, ge=0, le=360)
|
||
heading: int | None = Field(default=None, ge=0, le=511)
|
||
name: str | None = None
|
||
vessel_type: str | int | None = None
|
||
received_at: datetime | None = None
|
||
|
||
|
||
class GeoPointRecord(BaseModel):
|
||
lat: float = Field(ge=-90, le=90)
|
||
lon: float = Field(ge=-180, le=180)
|
||
name: str | None = None
|
||
type: str | None = None
|
||
source_id: str | None = None
|
||
observed_at: datetime | None = None
|
||
metadata: dict[str, Any] = Field(default_factory=dict)
|
||
|
||
|
||
class GenericRecord(BaseModel):
|
||
data: dict[str, Any] = Field(default_factory=dict)
|
||
source_id: str | None = None
|
||
observed_at: datetime | None = None
|
||
|
||
@field_validator("data")
|
||
@classmethod
|
||
def require_payload(cls, value: dict[str, Any]) -> dict[str, Any]:
|
||
if not value:
|
||
raise ValueError("generic_records requires a non-empty data object")
|
||
return value
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TargetField:
|
||
name: str
|
||
type: str
|
||
required: bool = False
|
||
description: str = ""
|
||
example: Any = None
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"name": self.name,
|
||
"type": self.type,
|
||
"required": self.required,
|
||
"description": self.description,
|
||
"example": self.example,
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TargetSchema:
|
||
key: str
|
||
label: str
|
||
description: str
|
||
fields: tuple[TargetField, ...]
|
||
model: type[BaseModel]
|
||
destination: str
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"key": self.key,
|
||
"label": self.label,
|
||
"description": self.description,
|
||
"destination": self.destination,
|
||
"fields": [field.to_dict() for field in self.fields],
|
||
}
|
||
|
||
def validate_record(self, record: dict[str, Any]) -> tuple[dict[str, Any] | None, list[str]]:
|
||
try:
|
||
return self.model.model_validate(record).model_dump(mode="json"), []
|
||
except ValidationError as exc:
|
||
return None, [
|
||
".".join(str(part) for part in error["loc"]) + f": {error['msg']}"
|
||
for error in exc.errors()
|
||
]
|
||
|
||
|
||
TARGET_SCHEMAS: dict[str, TargetSchema] = {
|
||
"vessel_ais": TargetSchema(
|
||
key="vessel_ais",
|
||
label="船舶 AIS",
|
||
description="船只位置、航速、航向、MMSI 等 AIS 数据。",
|
||
destination="vessel_position",
|
||
model=VesselAISRecord,
|
||
fields=(
|
||
TargetField("mmsi", "integer", True, "MMSI 九位船舶标识", 257123000),
|
||
TargetField("lat", "float", True, "纬度", 59.91),
|
||
TargetField("lon", "float", True, "经度", 10.75),
|
||
TargetField("sog", "float", False, "对地航速，单位节", 12.4),
|
||
TargetField("cog", "float", False, "对地航向，0-360 度", 184.5),
|
||
TargetField("heading", "integer", False, "船首向，0-511", 186),
|
||
TargetField("name", "string", False, "船名", "OSLO EXPRESS"),
|
||
TargetField("vessel_type", "string", False, "船型", "cargo"),
|
||
TargetField("received_at", "datetime", False, "数据接收时间", "2026-04-28T00:00:00Z"),
|
||
),
|
||
),
|
||
"geo_points": TargetSchema(
|
||
key="geo_points",
|
||
label="通用地理点",
|
||
description="带经纬度的通用实体或事件点位。",
|
||
destination="generic_geo_points",
|
||
model=GeoPointRecord,
|
||
fields=(
|
||
TargetField("lat", "float", True, "纬度", 1.3),
|
||
TargetField("lon", "float", True, "经度", 103.8),
|
||
TargetField("name", "string", False, "点位名称", "Singapore"),
|
||
TargetField("type", "string", False, "点位类型", "datacenter"),
|
||
TargetField("source_id", "string", False, "来源侧 ID", "sg-1"),
|
||
TargetField("observed_at", "datetime", False, "观测时间", "2026-04-28T00:00:00Z"),
|
||
TargetField("metadata", "object", False, "扩展字段", {"provider": "example"}),
|
||
),
|
||
),
|
||
"generic_records": TargetSchema(
|
||
key="generic_records",
|
||
label="通用结构化记录",
|
||
description="未知结构数据沉淀，不直接进入 Earth 图层。",
|
||
destination="collected_data",
|
||
model=GenericRecord,
|
||
fields=(
|
||
TargetField("data", "object", True, "结构化记录主体", {"raw": "value"}),
|
||
TargetField("source_id", "string", False, "来源侧 ID", "record-1"),
|
||
TargetField("observed_at", "datetime", False, "观测时间", "2026-04-28T00:00:00Z"),
|
||
),
|
||
),
|
||
}
|
||
|
||
|
||
def list_target_schemas() -> list[dict[str, Any]]:
|
||
return [schema.to_dict() for schema in TARGET_SCHEMAS.values()]
|
||
|
||
|
||
def get_target_schema(key: str) -> TargetSchema:
|
||
try:
|
||
return TARGET_SCHEMAS[key]
|
||
except KeyError as exc:
|
||
raise ValueError(f"Unsupported target schema: {key}") from exc
|