800 lines
32 KiB
Python
800 lines
32 KiB
Python
import bcrypt as _bcrypt
|
|
if not hasattr(_bcrypt, "__about__"):
|
|
import types; _bcrypt.__about__ = types.SimpleNamespace(__version__=_bcrypt.__version__ or "4.0.0")
|
|
|
|
from datetime import datetime
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from contextlib import asynccontextmanager
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '..', '.env'))
|
|
|
|
from database import engine, SessionLocal, ensure_column
|
|
from models.aid import Aid
|
|
from models.vessel import Vessel, VesselTrack, AtonTrack, RecordingEvent
|
|
from models.lamp import Lamp
|
|
from models.contact import Contact, AlertReport
|
|
from models.org import Port, Company, BuoyOwnership
|
|
import models.aid
|
|
import models.vessel
|
|
import models.user
|
|
import models.lamp
|
|
import models.contact
|
|
import models.org
|
|
from routers import aids
|
|
from routers.auth import seed_users, get_current_user, require_admin
|
|
from models.user import User
|
|
from fastapi import Depends, Query as _Query
|
|
from database import get_db
|
|
from sqlalchemy.orm import Session
|
|
from routers import auth as auth_router
|
|
from routers import charts as charts_router
|
|
from routers import equipment as equipment_router
|
|
from routers import lamps as lamps_router
|
|
from routers import contacts as contacts_router
|
|
from routers import tracks as tracks_router
|
|
from routers import org as org_router
|
|
from services.ais_simulator import run_simulator, MIAMI_AIDS
|
|
from services.alert_engine import evaluate_vessel, evaluate_aid_movement, aid_alert_state
|
|
from services.gps_reader import GPSReader
|
|
from services.aton_decoder import decode_type21, decode_type8_aton, process_aton_message, aton_state
|
|
from services import settings_store
|
|
import services.ais_catcher as _ais_catcher
|
|
from services.ais_udp_reader import run_udp_listener
|
|
import services.ais_udp_reader as _ais_udp
|
|
|
|
# Crear tablas
|
|
models.aid.Base.metadata.create_all(bind=engine)
|
|
models.vessel.Base.metadata.create_all(bind=engine)
|
|
models.user.Base.metadata.create_all(bind=engine)
|
|
models.lamp.Base.metadata.create_all(bind=engine)
|
|
models.contact.Base.metadata.create_all(bind=engine)
|
|
models.org.Base.metadata.create_all(bind=engine)
|
|
# Additive migrations for columns added after first install
|
|
ensure_column("aids", "lamp_id", "TEXT")
|
|
ensure_column("users", "prefs_json", "TEXT")
|
|
ensure_column("users", "company_id", "TEXT")
|
|
|
|
# Each entry: {"ws": WebSocket, "company_id": str|None}
|
|
# company_id=None means superadmin/admin — sees ALL traffic.
|
|
connected_clients: list[dict] = []
|
|
|
|
# Ownership cache: company_id → set of MMSI strings the company owns.
|
|
# Rebuilt at startup and refreshed via POST /org/refresh (or when ownership changes).
|
|
_ownership_cache: dict[str, set] = {} # company_id → {mmsi, ...}
|
|
_aid_ownership_cache: dict[str, set] = {} # company_id → {aid_id, ...}
|
|
|
|
# Track throttle: last persisted point per MMSI
|
|
_vessel_track_last: dict[str, dict] = {} # mmsi → {ts, lat, lon}
|
|
_aton_track_last: dict[str, dict] = {} # mmsi → {ts, lat, lon}
|
|
|
|
# Source of truth for runtime config — mutated via POST /settings.
|
|
config = settings_store.SETTINGS
|
|
|
|
# Background AIS reader task handle (lets us stop/restart on source change)
|
|
_ais_task: asyncio.Task | None = None
|
|
|
|
# Last-known battery alert state per ATON to avoid re-emitting every msg.
|
|
# Values: None | 'WARN' | 'ALARM'
|
|
_battery_alert_state: dict[str, str | None] = {}
|
|
|
|
# Estado en memoria (posiciones actuales)
|
|
vessels_state: dict = {}
|
|
aids_state: dict = {}
|
|
|
|
def _build_ownership_cache(db):
|
|
"""Rebuild the in-memory ownership maps (MMSI and aid_id) from the DB."""
|
|
_ownership_cache.clear()
|
|
_aid_ownership_cache.clear()
|
|
rows = db.query(BuoyOwnership).all()
|
|
for row in rows:
|
|
if row.mmsi:
|
|
_ownership_cache.setdefault(row.company_id, set()).add(row.mmsi)
|
|
if row.aid_id:
|
|
_aid_ownership_cache.setdefault(row.company_id, set()).add(row.aid_id)
|
|
|
|
|
|
def seed_ports():
|
|
"""Ensure canonical Colombian ports exist in the DB."""
|
|
PORTS = [
|
|
{"id": "port-barranquilla", "name": "Barranquilla",
|
|
"center_lat": 11.0041, "center_lon": -74.8070, "default_zoom": 12.0,
|
|
"chart_name": None},
|
|
{"id": "port-cartagena", "name": "Cartagena",
|
|
"center_lat": 10.3997, "center_lon": -75.5144, "default_zoom": 12.0,
|
|
"chart_name": "BAHÍA_DE_CARTAGENA"},
|
|
{"id": "port-santamarta", "name": "Santa Marta",
|
|
"center_lat": 11.2408, "center_lon": -74.2110, "default_zoom": 12.0,
|
|
"chart_name": None},
|
|
{"id": "port-buenaventura", "name": "Buenaventura",
|
|
"center_lat": 3.8800, "center_lon": -77.0311, "default_zoom": 12.0,
|
|
"chart_name": None},
|
|
{"id": "port-tumaco", "name": "Tumaco",
|
|
"center_lat": 1.8189, "center_lon": -78.7619, "default_zoom": 12.0,
|
|
"chart_name": None},
|
|
]
|
|
db = SessionLocal()
|
|
try:
|
|
for p in PORTS:
|
|
if not db.query(Port).filter(Port.id == p["id"]).first():
|
|
db.add(Port(**p))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def seed_contacts():
|
|
"""Pre-load the operator company (INSM) as an OWNER contact so the
|
|
REPORT flow has someone to notify out of the box."""
|
|
db = SessionLocal()
|
|
try:
|
|
from models.contact import Contact
|
|
existing = db.query(Contact).filter(Contact.id == "insm-default").first()
|
|
if not existing:
|
|
db.add(Contact(
|
|
id="insm-default",
|
|
role="OWNER",
|
|
name="INSM — Ingeniería Naval y Señalización Marítima",
|
|
company_name="INSM",
|
|
email="tecnicoinsm@gmail.com",
|
|
preferred_channel="EMAIL",
|
|
notes="Operator of this AidsMonitoring deployment.",
|
|
))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def seed_aids():
|
|
db = SessionLocal()
|
|
for a in MIAMI_AIDS:
|
|
if not db.query(Aid).filter(Aid.id == a["id"]).first():
|
|
aid = Aid(
|
|
id=a["id"],
|
|
nombre=a["nombre"],
|
|
categoria=a["categoria"],
|
|
tipo=a["tipo"],
|
|
tipo_ais=a["tipo_ais"],
|
|
mmsi=a.get("mmsi"),
|
|
lat_nominal=a["lat_nominal"],
|
|
lon_nominal=a["lon_nominal"],
|
|
radio_borneo_m=a.get("radio_borneo_m", 10.0),
|
|
caracteristica_luz=a.get("caracteristica_luz"),
|
|
alcance_nm=a.get("alcance_nm"),
|
|
fuente_posicion="MANUAL",
|
|
)
|
|
db.add(aid)
|
|
db.commit()
|
|
db.close()
|
|
|
|
def _client_may_see(client: dict,
|
|
owned_mmsi: str | None = None,
|
|
owned_aid_id: str | None = None) -> bool:
|
|
"""Return True if this client is allowed to receive this message.
|
|
|
|
- owned_mmsi → vessel/AtoN traffic filtered by MMSI ownership
|
|
- owned_aid_id → aid-position traffic filtered by aid_id ownership
|
|
- neither → system/alert message: always delivered
|
|
"""
|
|
cid = client.get("company_id")
|
|
if cid is None:
|
|
return True # admin/superadmin: sees all
|
|
if owned_mmsi is not None:
|
|
return owned_mmsi in _ownership_cache.get(cid, set())
|
|
if owned_aid_id is not None:
|
|
return owned_aid_id in _aid_ownership_cache.get(cid, set())
|
|
return True # non-filtered message
|
|
|
|
|
|
async def broadcast(message: dict,
|
|
owned_mmsi: str | None = None,
|
|
owned_aid_id: str | None = None):
|
|
"""
|
|
Send *message* to connected WebSocket clients.
|
|
|
|
owned_mmsi → filter vessel/AtoN traffic by MMSI (company users see only their own)
|
|
owned_aid_id → filter aid-position updates by aid_id
|
|
Admins (company_id=None) always receive everything.
|
|
"""
|
|
data = json.dumps(message)
|
|
dead = []
|
|
for client in connected_clients:
|
|
if not _client_may_see(client, owned_mmsi, owned_aid_id):
|
|
continue
|
|
try:
|
|
await client["ws"].send_text(data)
|
|
except Exception:
|
|
dead.append(client)
|
|
for c in dead:
|
|
if c in connected_clients:
|
|
connected_clients.remove(c)
|
|
|
|
async def _persist_recording(db, alert: dict):
|
|
"""Save or close a RecordingEvent row when auto-recording triggers."""
|
|
from models.vessel import RecordingEvent
|
|
import uuid as _uuid
|
|
if alert["tipo"] == "GRABACION_INICIADA":
|
|
rec = RecordingEvent(
|
|
id=str(_uuid.uuid4()),
|
|
mmsi=alert["mmsi"],
|
|
aid_id=alert["aid_id"],
|
|
inicio=datetime.utcnow(),
|
|
trigger=alert.get("trigger", "PROXIMIDAD"),
|
|
)
|
|
db.add(rec)
|
|
db.commit()
|
|
elif alert["tipo"] == "GRABACION_FINALIZADA":
|
|
from sqlalchemy import and_
|
|
rec = db.query(RecordingEvent).filter(
|
|
and_(
|
|
RecordingEvent.mmsi == alert["mmsi"],
|
|
RecordingEvent.aid_id == alert["aid_id"],
|
|
RecordingEvent.cerrado == False,
|
|
)
|
|
).order_by(RecordingEvent.inicio.desc()).first()
|
|
if rec:
|
|
rec.fin = datetime.utcnow()
|
|
rec.distancia_min_m = alert.get("distancia_min_m")
|
|
rec.cerrado = True
|
|
db.commit()
|
|
|
|
async def process_message(msg: dict):
|
|
db = SessionLocal()
|
|
try:
|
|
if msg["type"] == "vessel":
|
|
mmsi = msg["mmsi"]
|
|
vessels_state[mmsi] = msg
|
|
aids_list = list(aids_state.values())
|
|
alerts = evaluate_vessel(msg, aids_list, config)
|
|
await broadcast(msg) # vessels = public traffic, no filter
|
|
for alert in alerts:
|
|
await broadcast({"type": "alert", **alert})
|
|
if alert["tipo"] in ("GRABACION_INICIADA", "GRABACION_FINALIZADA"):
|
|
await _persist_recording(db, alert)
|
|
|
|
# ── Auto-persist VesselTrack (DVR) ───────────────────────────────
|
|
lat, lon = msg.get("lat"), msg.get("lon")
|
|
if lat is not None and lon is not None:
|
|
now = datetime.utcnow()
|
|
prev = _vessel_track_last.get(mmsi)
|
|
elapsed = (now - prev["ts"]).total_seconds() if prev else 9999
|
|
if elapsed >= 10:
|
|
db.add(VesselTrack(
|
|
mmsi=mmsi, timestamp=now,
|
|
lat=lat, lon=lon,
|
|
sog=msg.get("sog"), cog=msg.get("cog"),
|
|
heading=msg.get("heading"),
|
|
))
|
|
db.commit()
|
|
_vessel_track_last[mmsi] = {"ts": now, "lat": lat, "lon": lon}
|
|
|
|
elif msg["type"] == "aton":
|
|
# Real AIS Type 21 / Type 8 from hardware receiver
|
|
entry = process_aton_message(msg)
|
|
if entry:
|
|
mmsi = entry["mmsi"]
|
|
aton_state[mmsi] = entry
|
|
await broadcast({"type": "aton", **entry}, owned_mmsi=mmsi)
|
|
|
|
# Auto-upsert Aid row on first sight (Type 21 carries name + position).
|
|
# The user must then assign a lamp via the right panel.
|
|
if msg.get("msg_type") == 21 and entry.get("lat") is not None:
|
|
aid = db.query(Aid).filter(Aid.mmsi == mmsi).first()
|
|
if not aid:
|
|
from models.aid import Aid as _Aid
|
|
import uuid as _uuid
|
|
aid = _Aid(
|
|
id=str(_uuid.uuid4()),
|
|
nombre=entry.get("name") or f"AIS AtoN {mmsi}",
|
|
categoria="FLOTANTE",
|
|
tipo="BOYA_LATERAL",
|
|
tipo_ais="ATON_21",
|
|
mmsi=mmsi,
|
|
lat_nominal=entry["lat"],
|
|
lon_nominal=entry["lon"],
|
|
fuente_posicion="AIS",
|
|
)
|
|
db.add(aid); db.commit()
|
|
await broadcast({
|
|
"type": "alert",
|
|
"id": str(_uuid.uuid4()),
|
|
"tipo": "ALERTA_AMARILLA",
|
|
"subtipo": "AYUDA_NUEVA_SIN_CONFIGURAR",
|
|
"mmsi": mmsi,
|
|
"aid_id": aid.id,
|
|
"aid_nombre": aid.nombre,
|
|
"mensaje": "Nueva ayuda detectada — falta asignar lámpara",
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
})
|
|
|
|
# Battery threshold check — per-aid lamp values if assigned,
|
|
# otherwise the global defaults from settings_store.
|
|
v = entry.get("voltage_v")
|
|
if v is not None:
|
|
aid = db.query(Aid).filter(Aid.mmsi == mmsi).first()
|
|
lamp = None
|
|
if aid and aid.lamp_id:
|
|
lamp = db.query(Lamp).filter(Lamp.id == aid.lamp_id).first()
|
|
if lamp:
|
|
rng = lamp.voltage_max - lamp.voltage_min
|
|
warn_v = lamp.voltage_min + rng * 0.20
|
|
alarm_v = lamp.voltage_min + rng * 0.10
|
|
threshold_source = f"lamp:{lamp.manufacturer} {lamp.model}"
|
|
else:
|
|
warn_v = config["battery_warn_v"]
|
|
alarm_v = config["battery_alarm_v"]
|
|
threshold_source = "global"
|
|
prev = _battery_alert_state.get(mmsi)
|
|
new_state = None
|
|
if v <= alarm_v: new_state = "ALARM"
|
|
elif v <= warn_v: new_state = "WARN"
|
|
if new_state and new_state != prev:
|
|
import uuid as _uuid
|
|
await broadcast({
|
|
"type": "alert",
|
|
"id": str(_uuid.uuid4()),
|
|
"tipo": "ALERTA_ROJA" if new_state == "ALARM" else "ALERTA_AMARILLA",
|
|
"subtipo": "BATERIA_BAJA",
|
|
"mmsi": mmsi,
|
|
"aid_nombre": entry.get("name"),
|
|
"voltage_v": v,
|
|
"umbral": alarm_v if new_state == "ALARM" else warn_v,
|
|
"fuente_umbral": threshold_source,
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
})
|
|
_battery_alert_state[mmsi] = new_state
|
|
|
|
# ── Auto-persist AtonTrack (DVR) ─────────────────────────────
|
|
at_lat = entry.get("lat")
|
|
at_lon = entry.get("lon")
|
|
if at_lat is not None and at_lon is not None:
|
|
now = datetime.utcnow()
|
|
prev = _aton_track_last.get(mmsi)
|
|
elapsed = (now - prev["ts"]).total_seconds() if prev else 9999
|
|
if elapsed >= 30: # AtoN updates less frequent → 30s threshold
|
|
db.add(AtonTrack(
|
|
mmsi=mmsi, timestamp=now,
|
|
lat=at_lat, lon=at_lon,
|
|
voltage_v=entry.get("voltage_v"),
|
|
off_position=entry.get("off_position"),
|
|
))
|
|
db.commit()
|
|
_aton_track_last[mmsi] = {"ts": now, "lat": at_lat, "lon": at_lon}
|
|
|
|
elif msg["type"] == "aid_position":
|
|
aid_id = msg["id"]
|
|
if aid_id in aids_state:
|
|
aids_state[aid_id].update(msg)
|
|
aid = db.query(Aid).filter(Aid.id == aid_id).first()
|
|
if aid:
|
|
alert_list = evaluate_aid_movement(
|
|
aid_id, msg["lat_actual"], msg["lon_actual"],
|
|
aid.lat_nominal, aid.lon_nominal, config
|
|
)
|
|
aid.lat_actual = msg["lat_actual"]
|
|
aid.lon_actual = msg["lon_actual"]
|
|
aid.desplazamiento_m = msg["desplazamiento_m"]
|
|
aid.en_posicion = msg["en_posicion"]
|
|
db.commit()
|
|
await broadcast(msg) # aid positions = public, no filter
|
|
for alert in alert_list:
|
|
await broadcast({"type": "alert", **alert})
|
|
finally:
|
|
db.close()
|
|
|
|
async def _start_ais_source():
|
|
"""Start the AIS reader task matching the current config['ais_source']."""
|
|
src = config.get("ais_source", "SIMULATOR").upper()
|
|
if src == "SIMULATOR":
|
|
return asyncio.create_task(run_simulator(process_message))
|
|
if src == "SDR":
|
|
# Auto-launch AIS-catcher if hardware is present and not already running
|
|
if _ais_catcher.exe_exists() and not _ais_catcher.is_running():
|
|
result = _ais_catcher.launch()
|
|
if result["ok"]:
|
|
print(f"[ais] AIS-catcher launched (pid {result.get('pid')})")
|
|
else:
|
|
print(f"[ais] AIS-catcher launch failed: {result.get('error')}")
|
|
return asyncio.create_task(run_udp_listener(process_message, port=10110))
|
|
return None
|
|
|
|
|
|
async def _stop_ais_source():
|
|
global _ais_task
|
|
if _ais_task and not _ais_task.done():
|
|
_ais_task.cancel()
|
|
try: await _ais_task
|
|
except asyncio.CancelledError: pass
|
|
_ais_task = None
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
settings_store.init()
|
|
seed_aids()
|
|
seed_users()
|
|
seed_contacts()
|
|
seed_ports()
|
|
db = SessionLocal()
|
|
_build_ownership_cache(db)
|
|
db.close()
|
|
db = SessionLocal()
|
|
for aid in db.query(Aid).all():
|
|
aids_state[aid.id] = {
|
|
"id": aid.id, "nombre": aid.nombre, "categoria": aid.categoria,
|
|
"tipo": aid.tipo, "tipo_ais": aid.tipo_ais, "mmsi": aid.mmsi,
|
|
"lat_nominal": aid.lat_nominal, "lon_nominal": aid.lon_nominal,
|
|
"radio_borneo_m": aid.radio_borneo_m,
|
|
"lat_actual": aid.lat_actual, "lon_actual": aid.lon_actual,
|
|
"desplazamiento_m": aid.desplazamiento_m or 0,
|
|
"en_posicion": aid.en_posicion, "en_movimiento": aid.en_movimiento,
|
|
"caracteristica_luz": aid.caracteristica_luz, "alcance_nm": aid.alcance_nm,
|
|
"lamp_id": aid.lamp_id,
|
|
"puerto_responsable": aid.puerto_responsable,
|
|
"empresa_responsable": aid.empresa_responsable,
|
|
"ultima_senal": str(aid.ultima_senal) if aid.ultima_senal else None,
|
|
}
|
|
db.close()
|
|
|
|
global _ais_task
|
|
_ais_task = await _start_ais_source()
|
|
|
|
# GPS — settings_store now holds the port (mirrored from .env on first run)
|
|
gps_port = config.get("gps_port") or None
|
|
gps_baud = config.get("gps_baud") or None
|
|
gps = GPSReader(broadcast, forced_port=gps_port, forced_baud=gps_baud)
|
|
await gps.start()
|
|
app.state.gps = gps
|
|
|
|
yield
|
|
|
|
app = FastAPI(title="AidsMonitoring", lifespan=lifespan)
|
|
|
|
app.add_middleware(CORSMiddleware, allow_origins=["*"],
|
|
allow_methods=["*"], allow_headers=["*"])
|
|
|
|
app.include_router(auth_router.router)
|
|
app.include_router(aids.router)
|
|
app.include_router(charts_router.router)
|
|
app.include_router(equipment_router.router)
|
|
app.include_router(lamps_router.router)
|
|
app.include_router(contacts_router.router)
|
|
app.include_router(tracks_router.router)
|
|
app.include_router(org_router.router)
|
|
|
|
@app.post("/org/refresh")
|
|
async def refresh_ownership(
|
|
_user: User = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Rebuild the in-memory MMSI ownership cache from DB. Call after assigning/removing buoys."""
|
|
_build_ownership_cache(db)
|
|
return {"ok": True, "companies": len(_ownership_cache),
|
|
"total_mmsis": sum(len(v) for v in _ownership_cache.values())}
|
|
|
|
|
|
def _require_recording_permission(user: User):
|
|
"""ADMIN, SUPERADMIN, or CLIENT_ADMIN may start/stop recordings."""
|
|
from models.user import Role
|
|
if user.role not in (Role.ADMIN, Role.SUPERADMIN, Role.CLIENT_ADMIN):
|
|
from fastapi import HTTPException
|
|
raise HTTPException(status_code=403,
|
|
detail="Only ADMIN or CLIENT_ADMIN may control recordings")
|
|
|
|
@app.post("/recordings/start/{mmsi}")
|
|
async def manual_start_recording(
|
|
mmsi: str, aid_id: str = "MANUAL",
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
_require_recording_permission(current_user)
|
|
from services.alert_engine import active_recordings
|
|
from models.vessel import RecordingEvent
|
|
import uuid as _uuid
|
|
key = f"{mmsi}_{aid_id}"
|
|
if key in active_recordings:
|
|
return {"ok": False, "detail": "Already recording"}
|
|
active_recordings[key] = {
|
|
"mmsi": mmsi, "aid_id": aid_id, "aid_nombre": aid_id,
|
|
"inicio": datetime.utcnow().isoformat(),
|
|
"trigger": "MANUAL", "min_dist": 0, "pre_buffer": [],
|
|
}
|
|
db = SessionLocal()
|
|
try:
|
|
rec = RecordingEvent(
|
|
id=str(_uuid.uuid4()), mmsi=mmsi, aid_id=aid_id,
|
|
inicio=datetime.utcnow(), trigger="MANUAL",
|
|
)
|
|
db.add(rec); db.commit()
|
|
finally:
|
|
db.close()
|
|
vessel = vessels_state.get(mmsi, {})
|
|
await broadcast({"type": "alert", "tipo": "GRABACION_INICIADA",
|
|
"mmsi": mmsi, "aid_id": aid_id, "aid_nombre": aid_id,
|
|
"distancia_m": 0, "trigger": "MANUAL",
|
|
"timestamp": datetime.utcnow().isoformat()})
|
|
return {"ok": True}
|
|
|
|
@app.post("/recordings/stop/{mmsi}")
|
|
async def manual_stop_recording(
|
|
mmsi: str, aid_id: str = "MANUAL",
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
from services.alert_engine import active_recordings
|
|
from models.vessel import RecordingEvent
|
|
from sqlalchemy import and_
|
|
_require_recording_permission(current_user)
|
|
key = f"{mmsi}_{aid_id}"
|
|
rec_mem = active_recordings.pop(key, None)
|
|
min_dist = round(rec_mem["min_dist"], 1) if rec_mem else None
|
|
db = SessionLocal()
|
|
try:
|
|
rec = db.query(RecordingEvent).filter(
|
|
and_(RecordingEvent.mmsi == mmsi, RecordingEvent.aid_id == aid_id,
|
|
RecordingEvent.cerrado == False)
|
|
).order_by(RecordingEvent.inicio.desc()).first()
|
|
if rec:
|
|
rec.fin = datetime.utcnow()
|
|
rec.distancia_min_m = min_dist
|
|
rec.cerrado = True
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
await broadcast({"type": "alert", "tipo": "GRABACION_FINALIZADA",
|
|
"mmsi": mmsi, "aid_id": aid_id,
|
|
"distancia_min_m": min_dist,
|
|
"timestamp": datetime.utcnow().isoformat()})
|
|
return {"ok": True}
|
|
|
|
@app.get("/debug")
|
|
async def debug_state():
|
|
from services.alert_engine import aid_alert_state, aid_position_history
|
|
return {
|
|
"alert_states": dict(aid_alert_state),
|
|
"history_sizes": {k: len(v) for k, v in aid_position_history.items()},
|
|
"aids_in_state": list(aids_state.keys()),
|
|
"vessels_tracked": len(vessels_state),
|
|
}
|
|
|
|
@app.get("/settings")
|
|
async def get_settings(current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db)):
|
|
"""Return system defaults merged with this user's saved preferences."""
|
|
import json as _json
|
|
base = settings_store.get_all()
|
|
try:
|
|
user_row = db.query(User).filter(User.id == current_user.id).first()
|
|
prefs = _json.loads(user_row.prefs_json or '{}') if user_row else {}
|
|
except Exception:
|
|
prefs = {}
|
|
return {**base, **prefs}
|
|
|
|
|
|
@app.post("/settings")
|
|
async def update_settings(payload: dict,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db)):
|
|
"""Save user-specific preferences to the DB. System-level keys (GPS port,
|
|
AIS source) are also applied to the runtime config so they take effect
|
|
without a restart. Returns {ok, settings, applied: [...]}."""
|
|
import json as _json
|
|
global _ais_task
|
|
|
|
# Split: system keys go to settings_store; everything stays in user prefs
|
|
SYSTEM_KEYS = {"ais_source","ais_serial_port","ais_baud","ais_net_addr",
|
|
"gps_port","gps_baud","smtp_host","smtp_port","smtp_user",
|
|
"smtp_password","smtp_from","smtp_from_name","smtp_use_tls"}
|
|
|
|
system_patch = {k: v for k, v in (payload or {}).items() if k in SYSTEM_KEYS}
|
|
prev = settings_store.get_all()
|
|
if system_patch:
|
|
settings_store.apply_patch(system_patch)
|
|
new = settings_store.get_all()
|
|
applied = []
|
|
|
|
# AIS source change
|
|
if system_patch.get("ais_source") and new["ais_source"] != prev["ais_source"]:
|
|
await _stop_ais_source()
|
|
_ais_task = await _start_ais_source()
|
|
applied.append(f"ais_source → {new['ais_source']}")
|
|
|
|
# GPS port change
|
|
if "gps_port" in system_patch or "gps_baud" in system_patch:
|
|
gps: GPSReader = getattr(app.state, "gps", None)
|
|
if gps:
|
|
gps._forced_port = new["gps_port"] or None
|
|
gps._forced_baud = new["gps_baud"] or None
|
|
gps._port = None
|
|
gps.status = "SEARCHING"
|
|
gps.reconnect()
|
|
applied.append(f"gps_port → {new['gps_port'] or 'auto'} @ {new['gps_baud']}")
|
|
|
|
# Save full payload as this user's preferences
|
|
try:
|
|
user_row = db.query(User).filter(User.id == current_user.id).first()
|
|
if user_row:
|
|
existing = _json.loads(user_row.prefs_json or '{}')
|
|
existing.update(payload or {})
|
|
user_row.prefs_json = _json.dumps(existing)
|
|
db.commit()
|
|
except Exception:
|
|
pass
|
|
|
|
return {"ok": True, "settings": {**new, **(payload or {})}, "applied": applied}
|
|
|
|
|
|
@app.get("/gps/status")
|
|
async def gps_status(request: Request):
|
|
gps: GPSReader = getattr(request.app.state, "gps", None)
|
|
if not gps:
|
|
return {"status": "DISABLED", "port": None, "last_fix": None}
|
|
return {
|
|
"status": gps.status,
|
|
"port": gps._port or gps._forced_port,
|
|
"forced_port": gps._forced_port,
|
|
"baud": gps._baud,
|
|
"last_fix": gps.last_fix,
|
|
}
|
|
|
|
@app.post("/gps/port")
|
|
async def set_gps_port(request: Request, port: str = "", baud: int = 9600):
|
|
"""Hot-change GPS port without server restart."""
|
|
gps: GPSReader = getattr(request.app.state, "gps", None)
|
|
if not gps:
|
|
return {"ok": False, "detail": "GPS reader not running"}
|
|
gps._forced_port = port.strip() or None
|
|
gps._forced_baud = baud or None
|
|
gps._port = None # force reconnect on next loop iteration
|
|
gps.status = "SEARCHING"
|
|
gps.reconnect() # close active serial so _read_loop exits and re-probes
|
|
# Persist to .env for next restart
|
|
env_path = os.path.join(os.path.dirname(__file__), '..', '.env')
|
|
try:
|
|
lines = open(env_path).readlines()
|
|
def _set(lines, key, val):
|
|
for i, l in enumerate(lines):
|
|
if l.startswith(f"{key}="):
|
|
lines[i] = f"{key}={val}\n"; return lines
|
|
lines.append(f"{key}={val}\n"); return lines
|
|
lines = _set(lines, "GPS_PORT", port.strip())
|
|
lines = _set(lines, "GPS_BAUD", str(baud))
|
|
open(env_path, 'w').writelines(lines)
|
|
except Exception as e:
|
|
pass # non-fatal — in-memory change already applied
|
|
return {"ok": True, "port": gps._forced_port, "baud": baud}
|
|
|
|
@app.get("/ais/status")
|
|
async def ais_status():
|
|
"""Return AIS-catcher process status and SDR hardware info."""
|
|
return _ais_catcher.status()
|
|
|
|
|
|
@app.get("/ais/stats")
|
|
async def ais_stats():
|
|
"""Return live UDP listener stats: sentence counts, last receive time."""
|
|
import time as _time
|
|
udp = _ais_udp.get_stats()
|
|
catcher = _ais_catcher.status()
|
|
last_ts = udp.get("last_sentence_ts")
|
|
seconds_since = round(_time.time() - last_ts, 1) if last_ts else None
|
|
return {
|
|
**udp,
|
|
"seconds_since_last": seconds_since,
|
|
"signal": last_ts is not None and seconds_since < 60,
|
|
"catcher_running": catcher["running"],
|
|
"sdr_devices": catcher["sdr_devices"],
|
|
"ais_source": config.get("ais_source", "SIMULATOR"),
|
|
}
|
|
|
|
|
|
@app.get("/ais/raw")
|
|
async def ais_raw():
|
|
"""Return last 30 raw NMEA sentences received."""
|
|
return {"sentences": _ais_udp.get_raw()}
|
|
|
|
|
|
@app.post("/ais/launch")
|
|
async def ais_launch(udp_port: int = 10110, web_port: int = 8100):
|
|
"""Launch AIS-catcher.exe (auto-switches ais_source to SDR)."""
|
|
global _ais_task
|
|
result = _ais_catcher.launch(udp_port=udp_port, web_port=web_port)
|
|
# Always sync source to SDR after a successful launch (or already-running),
|
|
# otherwise the catcher streams to UDP:10110 with no listener attached.
|
|
if result["ok"] and config.get("ais_source", "SIMULATOR").upper() != "SDR":
|
|
settings_store.apply_patch({"ais_source": "SDR"})
|
|
await _stop_ais_source()
|
|
_ais_task = await _start_ais_source()
|
|
return result
|
|
|
|
|
|
@app.post("/ais/stop")
|
|
async def ais_stop():
|
|
"""Stop AIS-catcher.exe."""
|
|
return _ais_catcher.stop()
|
|
|
|
|
|
@app.websocket("/ws")
|
|
async def websocket_endpoint(
|
|
ws: WebSocket,
|
|
token: str | None = _Query(default=None),
|
|
):
|
|
await ws.accept()
|
|
|
|
# ── Resolve company_id from optional JWT token ────────────────────────────
|
|
company_id = None
|
|
if token:
|
|
try:
|
|
from jose import jwt as _jwt, JWTError as _JWTError
|
|
from routers.auth import SECRET_KEY, ALGORITHM
|
|
payload = _jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
username = payload.get("sub")
|
|
if username:
|
|
_db = SessionLocal()
|
|
try:
|
|
_u = _db.query(User).filter(User.username == username,
|
|
User.activo == True).first()
|
|
if _u:
|
|
company_id = getattr(_u, "company_id", None)
|
|
finally:
|
|
_db.close()
|
|
except Exception:
|
|
pass # invalid/expired token → treat as anonymous (admin-level view)
|
|
|
|
client = {"ws": ws, "company_id": company_id}
|
|
connected_clients.append(client)
|
|
|
|
# ── Build filtered init snapshot ──────────────────────────────────────────
|
|
# company users see ONLY their company's aids/vessels/atons
|
|
# admins (company_id=None) see everything
|
|
owned_mmsis = _ownership_cache.get(company_id) if company_id else None
|
|
owned_aid_ids= _aid_ownership_cache.get(company_id) if company_id else None
|
|
|
|
# aids = official positions → visible to ALL users (public nav info)
|
|
# vessels= ship AIS traffic → visible to ALL users
|
|
# atons = live AIS AtoN msgs → filtered by company MMSI ownership
|
|
init_aids = list(aids_state.values())
|
|
init_vessels = list(vessels_state.values())
|
|
|
|
if company_id is None:
|
|
# Admin / superadmin → all AtoNs
|
|
init_atons = list(aton_state.values())
|
|
else:
|
|
# Client role: get their owned MMSIs (empty set = no buoys assigned = sees nothing)
|
|
client_mmsis = _ownership_cache.get(company_id, set())
|
|
init_atons = [a for a in aton_state.values()
|
|
if a.get("mmsi") in client_mmsis]
|
|
|
|
await ws.send_text(json.dumps({
|
|
"type": "init",
|
|
"aids": init_aids,
|
|
"vessels": init_vessels,
|
|
"atons": init_atons,
|
|
}))
|
|
|
|
# Re-emit any active aid alerts so new clients see current state
|
|
from datetime import datetime as _dt
|
|
for aid_id, state in aid_alert_state.items():
|
|
if state == 'RED':
|
|
await ws.send_text(json.dumps({"type": "alert", "tipo": "ALERTA_ROJA",
|
|
"subtipo": "AYUDA_EN_MOVIMIENTO", "aid_id": aid_id,
|
|
"mensaje": "Movimiento continuo detectado",
|
|
"timestamp": _dt.utcnow().isoformat()}))
|
|
elif state == 'YELLOW':
|
|
aid = aids_state.get(aid_id, {})
|
|
await ws.send_text(json.dumps({"type": "alert", "tipo": "ALERTA_AMARILLA",
|
|
"subtipo": "AYUDA_FUERA_POSICION", "aid_id": aid_id,
|
|
"desplazamiento_m": round(aid.get("desplazamiento_m", 0), 1),
|
|
"timestamp": _dt.utcnow().isoformat()}))
|
|
try:
|
|
while True:
|
|
await ws.receive_text()
|
|
except WebSocketDisconnect:
|
|
if client in connected_clients:
|
|
connected_clients.remove(client)
|
|
|
|
app.mount("/", StaticFiles(directory=os.path.join(os.path.dirname(__file__), '..', 'frontend'), html=True), name="frontend")
|