274 lines
9.2 KiB
Python
274 lines
9.2 KiB
Python
"""BarentsWatch AIS collector for vessel tracking."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.vessel import VesselPosition, VesselStatic
|
|
from app.services.barentswatch import (
|
|
BARENTSWATCH_LATEST_URL,
|
|
fetch_barentswatch_access_token,
|
|
resolve_barentswatch_config,
|
|
)
|
|
from app.services.collectors.base import BaseCollector
|
|
|
|
|
|
VESSEL_TYPE_NAMES = {
|
|
30: "Fishing",
|
|
35: "Military",
|
|
60: "Passenger",
|
|
70: "Cargo",
|
|
80: "Tanker",
|
|
}
|
|
|
|
|
|
class VesselAISCollector(BaseCollector):
|
|
"""Collect latest AIS positions and append them to vessel tables."""
|
|
|
|
name = "barentswatch_vessels"
|
|
priority = "P1"
|
|
module = "L4"
|
|
frequency_hours = 1
|
|
data_type = "vessel_ais"
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return self._resolved_url or BARENTSWATCH_LATEST_URL
|
|
|
|
async def _get_access_token(self, client: httpx.AsyncClient) -> str | None:
|
|
config = await resolve_barentswatch_config(self._db_session)
|
|
return await fetch_barentswatch_access_token(client, config)
|
|
|
|
async def fetch(self) -> list[dict[str, Any]]:
|
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
|
headers: dict[str, str] = {}
|
|
token = await self._get_access_token(client)
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
|
|
response = await client.get(self.base_url, headers=headers)
|
|
if response.status_code == 401 and not token:
|
|
return self._get_sample_data()
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
|
|
if isinstance(payload, list):
|
|
return [item for item in payload if isinstance(item, dict)]
|
|
if isinstance(payload, dict):
|
|
for key in ("features", "data", "items", "vessels"):
|
|
value = payload.get(key)
|
|
if isinstance(value, list):
|
|
if key == "features":
|
|
return [
|
|
{
|
|
**(item.get("properties") or {}),
|
|
"geometry": item.get("geometry"),
|
|
}
|
|
for item in value
|
|
if isinstance(item, dict)
|
|
]
|
|
return [item for item in value if isinstance(item, dict)]
|
|
return self._get_sample_data()
|
|
|
|
def transform(self, raw_data: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
transformed = []
|
|
for item in raw_data:
|
|
record = self._normalize_record(item)
|
|
if record:
|
|
transformed.append(record)
|
|
return transformed
|
|
|
|
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
|
|
|
|
for index, item in enumerate(data):
|
|
static = await db.get(VesselStatic, item["mmsi"])
|
|
if static is None:
|
|
static = VesselStatic(mmsi=item["mmsi"])
|
|
db.add(static)
|
|
|
|
for field in (
|
|
"name",
|
|
"callsign",
|
|
"vessel_type",
|
|
"vessel_type_name",
|
|
"flag",
|
|
"length",
|
|
"width",
|
|
"draught",
|
|
"imo",
|
|
):
|
|
value = item.get(field)
|
|
if value not in (None, ""):
|
|
setattr(static, field, value)
|
|
static.updated_at = now
|
|
|
|
db.add(
|
|
VesselPosition(
|
|
mmsi=item["mmsi"],
|
|
lat=item["lat"],
|
|
lon=item["lon"],
|
|
sog=item.get("sog"),
|
|
cog=item.get("cog"),
|
|
heading=item.get("heading"),
|
|
nav_status=item.get("nav_status"),
|
|
received_at=item.get("received_at") or now,
|
|
)
|
|
)
|
|
records_added += 1
|
|
|
|
if (index + 1) % 1000 == 0:
|
|
await self.update_progress(index + 1, commit=True)
|
|
|
|
await db.execute(
|
|
delete(VesselPosition).where(VesselPosition.received_at < now - timedelta(hours=24))
|
|
)
|
|
await db.commit()
|
|
await self.update_progress(records_added, force=True)
|
|
return records_added
|
|
|
|
def _normalize_record(self, item: dict[str, Any]) -> dict[str, Any] | None:
|
|
mmsi = _as_int(_pick(item, "mmsi", "MMSI", "Mmsi"))
|
|
lat = _as_float(_pick(item, "lat", "latitude", "Latitude"))
|
|
lon = _as_float(_pick(item, "lon", "lng", "longitude", "Longitude"))
|
|
|
|
geometry = item.get("geometry")
|
|
coordinates = geometry.get("coordinates") if isinstance(geometry, dict) else None
|
|
if (lat is None or lon is None) and isinstance(coordinates, list) and len(coordinates) >= 2:
|
|
lon = _as_float(coordinates[0])
|
|
lat = _as_float(coordinates[1])
|
|
|
|
if mmsi is None or lat is None or lon is None:
|
|
return None
|
|
if not (-90 <= lat <= 90 and -180 <= lon <= 180):
|
|
return None
|
|
|
|
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)
|
|
)
|
|
received_at = _parse_datetime(_pick(item, "received_at", "timestamp", "time", "msgtime"))
|
|
|
|
return {
|
|
"mmsi": mmsi,
|
|
"name": _pick(item, "name", "shipName", "ship_name", "Name"),
|
|
"callsign": _pick(item, "callsign", "callSign", "CallSign"),
|
|
"vessel_type": vessel_type,
|
|
"vessel_type_name": vessel_type_name,
|
|
"flag": _pick(item, "flag", "country", "Flag"),
|
|
"length": _as_float(_pick(item, "length", "shipLength", "Length")),
|
|
"width": _as_float(_pick(item, "width", "shipWidth", "Width")),
|
|
"draught": _as_float(_pick(item, "draught", "draft", "Draught")),
|
|
"imo": _as_int(_pick(item, "imo", "IMO", "imoNumber")),
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"sog": _as_float(_pick(item, "sog", "speedOverGround", "SOG")),
|
|
"cog": _as_float(_pick(item, "cog", "courseOverGround", "COG")),
|
|
"heading": _as_int(_pick(item, "heading", "trueHeading", "Heading")),
|
|
"nav_status": _as_int(_pick(item, "nav_status", "navStatus", "NavigationalStatus")),
|
|
"received_at": received_at,
|
|
}
|
|
|
|
def _get_sample_data(self) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"mmsi": 257123000,
|
|
"name": "OSLO TRADER",
|
|
"lat": 59.91,
|
|
"lon": 10.73,
|
|
"sog": 12.4,
|
|
"cog": 214,
|
|
"heading": 215,
|
|
"nav_status": 0,
|
|
"vessel_type": 70,
|
|
"vessel_type_name": "Cargo",
|
|
"flag": "NO",
|
|
"length": 185,
|
|
},
|
|
{
|
|
"mmsi": 257456000,
|
|
"name": "NORDIC FJORD",
|
|
"lat": 60.39,
|
|
"lon": 5.32,
|
|
"sog": 0.2,
|
|
"cog": 82,
|
|
"heading": 80,
|
|
"nav_status": 1,
|
|
"vessel_type": 60,
|
|
"vessel_type_name": "Passenger",
|
|
"flag": "NO",
|
|
"length": 126,
|
|
},
|
|
]
|
|
|
|
|
|
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 _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
|
|
|
|
|
|
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")
|