backend/runtime.py

from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Deque, Dict, Optional

from websocket import manager as ws_manager


@dataclass
class NS3State:
    socket_path: str = ""
    server_available: bool = False
    client_connected: bool = False
    total_messages: int = 0
    total_flows: int = 0
    last_event_at: Optional[str] = None
    last_flow_at: Optional[str] = None
    last_error: Optional[str] = None
    events: Deque[Dict[str, Any]] = field(default_factory=lambda: deque(maxlen=200))

    def log(self, event_type: str, message: str, **extra: Any) -> None:
        now = datetime.utcnow().isoformat()
        event = {"timestamp": now, "type": event_type, "message": message, **extra}
        self.last_event_at = now
        self.events.appendleft(event)


flow_queue = None
ml_worker = None
ns3_reader = None
ns3_state = NS3State()