"""Runtime settings, read from environment variables (and ``.env`` in the working directory, if present).
Every field maps to the upper-case env var of the same name, e.g. ``NDTP_PORT=9201``.
Lists are JSON: ``CORS_ORIGINS='["https://app.mowtransit.ru"]'``.
"""
from __future__ import annotations
from datetime import date, datetime
from pathlib import Path
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from .clock import DATASET_DAY
[документация]
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_ignore_empty=True)
log_level: str = "INFO"
cors_origins: list[str] = Field(default=["*"], description="browser origins allowed to call the API")
# NDTP listener (terminals / emulator connect here)
ndtp_enabled: bool = True
ndtp_host: str = "0.0.0.0"
ndtp_port: int = 9201
ndtp_idle_timeout_s: float = 300.0
ndtp_max_connections: int = 20_000
ndtp_backlog: int = 4096
# Dataset (the organizers' archive): unit registry and replay source
dataset_dir: Path | None = Field(default=None, description="folder containing train/ test/ validate/ labels/")
dataset_split: str = "validate"
# Dataset clock (see app/clock.py)
clock_day: date = DATASET_DAY
clock_start: datetime | None = Field(default=None, description="dataset time at startup, e.g. "
"2026-01-06T07:30:00; default: current Moscow time of day on clock_day")
clock_speed: float = Field(default=1.0, gt=0)
# Ingest
ingest_queue_size: int = Field(default=100_000, description="fixes buffered between the listener and ingest")
replay_enabled: bool = True
replay_backfill_s: float = Field(default=3600.0, description="dataset history replayed at once on startup")
ndtp_fresh_s: float = Field(default=60.0, description="a vehicle with NDTP this recent ignores replayed rows")
ndtp_max_clock_skew_s: float = Field(default=300.0, description="beyond this terminal-vs-server clock "
"difference, the server receive time is used")
# Vehicle state
state_tick_s: float = Field(default=2.0, description="how often arrivals/derived features are recomputed")
# Predictions (ML service: ml/src/inference_service.py)
ml_url: str = "http://ml:8001"
ml_timeout_s: float = 10.0
predict_tick_s: float = Field(default=5.0, description="how often vehicles are checked for a new target stop")
predict_retry_s: float = Field(default=30.0, description="after an ML failure, forecasts use the baseline this long")
predict_max_ping_age_s: float = Field(default=900.0, description="vehicles silent longer are treated as not in service")
# Alerts
alert_risk_threshold: float = Field(default=0.7, description="P(> 2 min late) that raises an alert (dashboard red)")
alert_tick_s: float = Field(default=5.0, description="how often active alerts are checked against arrivals")
# Dashboard
ws_tick_s: float = Field(default=1.0, description="how often vehicle.update is pushed over the WebSocket")
ui_max_ping_age_s: float = Field(default=1800.0, description="vehicles silent longer are hidden from the map")
# History (Postgres); unset = history off
database_url: str | None = Field(default=None, description="e.g. postgresql://msk:msk@postgres:5432/msk_transport")
history_flush_s: float = 1.0