Proxy Handlers¶

def on_siminfo_received(
    speed, simdt, simt, simutc, ntraf, state, scenname, sender_id=None
):
    """Process a BlueSky SIMINFO event and emit ``siminfo`` to web clients.

    Updates the per-node status/clock tracking for every sending node (feeding
    the Simulation Nodes panel via a throttled ``node_info`` emission), but only
    caches and emits the header simulation info for the active node so the
    displayed clock does not jump between nodes. Emissions to browsers are
    throttled to ``proxy.siminfo_interval``.

    Args:
        speed (float): Simulation speed multiplier.
        simdt (float): Simulation timestep in seconds.
        simt (float): Elapsed simulation time in seconds.
        simutc (str): Simulation UTC time string.
        ntraf (int): Number of aircraft currently in the simulation.
        state (int): Simulation run-state code.
        scenname (str): Name of the currently loaded scenario.
        sender_id (bytes | str | None): Identifier of the sending node; bytes
            are converted to a hex string.
    """
    proxy = active_proxy()
    if not proxy:
        return

    sender_id_str = id2str(sender_id)
    current_time = time.time()

    # Update per-node tracking for EVERY node so the Simulation Nodes panel
    # shows each node's own clock, regardless of which node is currently active.
    if sender_id_str and sender_id_str in proxy.tracked_nodes:
        simt_str = tim2txt(simt)[:-3] if simt is not None else "00:00:00"
        proxy.tracked_nodes[sender_id_str].update(
            {"status": scenname or "init", "time": simt_str}
        )
        # Wall-clock throttle: a sim-time cadence misbehaves when the sim is
        # paused or fast-forwarded.
        if (current_time - proxy.last_node_info_emit) >= proxy.node_info_interval:
            proxy.last_node_info_emit = current_time
            proxy._emit_node_info()

    # Cache and emit the header sim info solely for the active node.
    if not is_active_node(proxy, sender_id_str):
        return

    proxy.sim_data = {
        "speed": float(speed) if speed is not None else 0.0,
        "simdt": float(simdt) if simdt is not None else 0.0,
        "simt": float(simt) if simt is not None else 0.0,
        "simutc": str(simutc) if simutc is not None else "",
        "ntraf": int(ntraf) if ntraf is not None else 0,
        "state": int(state) if state is not None else 0,
        "scenname": str(scenname) if scenname is not None else "",
        "sender_id": sender_id_str,
    }

    # Throttle sim info emissions
    if (
        proxy.socketio
        and proxy.connected_clients > 0
        and (current_time - proxy.last_siminfo_emit) >= proxy.siminfo_interval
    ):
        try:
            proxy.socketio.emit("siminfo", proxy.sim_data)
            proxy.last_siminfo_emit = current_time
        except Exception as e:
            logger.error(f"Proxy→Web: Error sending SIMINFO: {e}")

def on_acdata_received(data):
    """Process a BlueSky ACDATA traffic frame and emit ``acdata`` to web clients.

    Hot path: the network timer delivers ACDATA at up to 50 Hz, but it is only
    emitted to browsers at ``acdata_interval`` (10 Hz) and only for the active
    node. ``make_json_serializable`` is the dominant per-frame cost, so it is
    deferred until after the active-node filter and the emit throttle decide the
    frame is actually sent. Set WEBATM_PERF=1 (WebATM.proxy.perf) to measure it.

    Aircraft clearing on a simulation reset is NOT handled here: BlueSky
    signals a reset on the dedicated RESET topic (events.on_reset_received),
    never via an ACDATA action.

    Args:
        data (dict): Aircraft state arrays keyed by field (``id``, ``lat``,
            ``lon``, ``alt``, ``tas``, ``trk``, ``vs``, conflict counters, ...)
            as sent by the BlueSky server.
    """
    try:
        proxy = get_bluesky_proxy()
        if not proxy:
            logger.debug("on_acdata_received called but no proxy available")
            return

        # Ignore data if reconnection is not allowed (we're disconnected)
        if not proxy.allow_reconnection:
            logger.debug("on_acdata_received ignored - reconnection not allowed")
            return

        # The sending node is set on the shared context just before this
        # synchronous dispatch.
        sender_id_str = None
        if proxy.bluesky_client and hasattr(proxy.bluesky_client, "context"):
            sender_id_str = id2str(
                getattr(proxy.bluesky_client.context, "sender_id", None)
            )

        # ACDATA from any node proves the link is alive: update liveness
        # before the active-node filter.
        proxy.last_successful_update = time.time()
        data_path_perf.record_received()

        # Only the active node's traffic is displayed (mirrors the SIMINFO
        # filter).
        if not is_active_node(proxy, sender_id_str):
            data_path_perf.record_filtered()
            return

        # Throttle BEFORE serializing: serialization is the dominant per-frame
        # cost, so pay it only for frames that are actually emitted.
        current_time = time.time()
        if not (
            proxy.socketio
            and proxy.connected_clients > 0
            and (current_time - proxy.last_acdata_emit) >= proxy.acdata_interval
        ):
            return

        t0 = time.perf_counter()
        serializable_data = make_json_serializable(data)
        data_path_perf.record_serialize(time.perf_counter() - t0)
        proxy.traffic_data = serializable_data

        try:
            t1 = time.perf_counter()
            proxy.socketio.emit("acdata", serializable_data)
            proxy.last_acdata_emit = current_time
            data_path_perf.record_emit(time.perf_counter() - t1)
        except Exception:
            logger.exception("Error emitting ACDATA")

    except Exception:
        logger.exception(f"Error in on_acdata_received (data type: {type(data)})")
    finally:
        data_path_perf.maybe_log()

def on_statechange_received(data, sender_id=None):
    """Process a BlueSky STATECHANGE event and emit ``statechange`` to web clients.

    Updates the cached simulation state (``proxy.sim_data['state']``) and
    immediately forwards the new run-state and sending node to connected
    browsers, without throttling. Like SIMINFO, only the active node's state
    changes touch the cached header state — a paused background node must not
    flip the displayed run-state.

    Args:
        data (dict): Event payload; the ``simstate`` key holds the new
            run-state code. Payloads without ``simstate`` are ignored.
        sender_id (bytes | str | None): Identifier of the sending node; bytes
            are converted to a hex string.
    """
    proxy = active_proxy()
    if not proxy:
        return

    simstate = data.get("simstate") if isinstance(data, dict) else None
    if simstate is None:
        return

    sender_id_str = id2str(sender_id)

    if not is_active_node(proxy, sender_id_str):
        logger.debug(
            f"STATECHANGE from background node {sender_id_str} ignored "
            f"(simstate={simstate})"
        )
        return

    # Only patch an existing SIMINFO cache; seeding a bare {'state': ...} would
    # let the backup timer re-emit a partial siminfo payload.
    if proxy.sim_data:
        proxy.sim_data["state"] = int(simstate)

    payload = {
        "simstate": int(simstate),
        "sender_id": sender_id_str,
    }

    if proxy.socketio and proxy.connected_clients > 0:
        try:
            proxy.socketio.emit("statechange", payload)
        except Exception as e:
            logger.error(f"Proxy->Web: Error sending STATECHANGE: {e}")

    logger.info(f"STATECHANGE from {sender_id_str}: simstate={simstate}")

def on_poly_received(data, *args, **kwargs):
    """Process a BlueSky POLY event and emit ``poly``/``polyline`` to web clients.

    Resolves the sending node and shared-state action from the BlueSky network
    context and applies the message to that node's stored shapes: Delete
    removes the named shapes, Replace/Reset/ActChange overwrite the stored
    sets, and updates merge into them (patching partial per-shape updates such
    as a colour change). At most the five most recent polygons and polylines
    are kept per node. The complete stored shape sets are emitted only when
    the sender is the currently active node.

    Args:
        data (dict): POLY payload from the BlueSky server (the shared-state
            action wrapper is already stripped by the network client): a
            ``polys`` mapping of shape name to shape info, or a list of shape
            names for a Delete action.
        *args (Any): Extra positional arguments from the network dispatch (unused).
        **kwargs (Any): Extra keyword arguments from the network dispatch (unused).
    """
    proxy = active_proxy()
    if not proxy:
        return

    try:
        sender_id, action = shared_context(proxy)
        poly_data = make_json_serializable(data)

        if sender_id:
            poly_store = proxy.poly_data_by_node.setdefault(sender_id, {"polys": {}})
            line_store = proxy.polyline_data_by_node.setdefault(
                sender_id, {"polys": {}}
            )

            if action == DELETE_ACTION:
                for name in _deleted_names(poly_data):
                    poly_store["polys"].pop(name, None)
                    line_store["polys"].pop(name, None)
            else:
                separated = _separate_poly_and_polyline_data(poly_data)
                new_polys = _shapes_of(separated["polygons"])
                new_lines = _shapes_of(separated["polylines"])

                if action in REPLACE_ACTIONS:
                    poly_store["polys"] = new_polys
                    line_store["polys"] = new_lines
                else:
                    _merge_shapes(
                        poly_store["polys"], line_store["polys"], new_polys, new_lines
                    )

                _trim_shape_store(poly_store["polys"], "polygons")
                _trim_shape_store(line_store["polys"], "polylines")

        # Only the active node's shapes are displayed; emit the complete stored
        # sets (not just this message's shapes).
        active_node_id = proxy._get_safe_active_node()
        if sender_id and active_node_id and sender_id == active_node_id:
            # Same guard as every other emit site; a client connecting later
            # gets the stored sets from its initial_data/connect envelopes.
            if proxy.socketio and proxy.connected_clients > 0:
                proxy.socketio.emit(
                    "poly", proxy.poly_data_by_node.get(sender_id, {"polys": {}})
                )
                proxy.socketio.emit(
                    "polyline",
                    proxy.polyline_data_by_node.get(sender_id, {"polys": {}}),
                )

    except Exception as e:
        logger.error(f"Error processing POLY data: {e}")
        import traceback

        traceback.print_exc()

def on_stackcmds_received(data):
    """Process a BlueSky STACKCMDS event and emit ``cmddict`` to web clients.

    BlueSky publishes STACKCMDS with replace semantics: each node answers a
    REQUEST with a full Replace of *its own* command dictionary (plugins can
    register extra commands per node). The console sends commands to the
    active node, so only the active node's dictionary is accepted — a
    background node's answer must not overwrite (or leak commands into) the
    set the console validates against. A Replace action rebuilds the stored
    dictionary; other actions merge. Other payload shapes are only logged.

    Args:
        data (dict | bytes | str): Unwrapped STACKCMDS shared-state payload; a
            dict is expected to contain a ``cmddict`` mapping of command names
            to metadata.
    """
    proxy = active_proxy()
    if not proxy:
        return

    if not isinstance(data, dict):
        logger.debug(f"Ignoring non-dict STACKCMDS payload of type {type(data)}")
        return

    cmddict = data.get("cmddict")
    if not isinstance(cmddict, dict):
        logger.warning(f"STACKCMDS payload without a cmddict mapping: {data.keys()}")
        return

    sender_id, action = shared_context(proxy)
    if not is_active_node(proxy, sender_id):
        logger.debug(f"STACKCMDS from background node {sender_id} ignored")
        return

    if action in REPLACE_ACTIONS:
        proxy.cmddict.clear()
    proxy.cmddict.update(cmddict)
    logger.debug(f"Updated cmddict with {len(cmddict)} commands (action={action})")

    if proxy.socketio and proxy.connected_clients > 0:
        try:
            proxy.socketio.emit("cmddict", {"cmddict": proxy.cmddict})
        except Exception as e:
            logger.error(f"Error emitting cmddict: {e}")

def on_stack_received(data):
    """Process a BlueSky STACK event carrying server-forwarded command lines.

    Normalizes the payload to a list of command lines and runs each non-empty
    line through ``_process_server_command``, which executes it locally and
    reports the outcome via the proxy's echo channel. Nothing is sent back to
    the BlueSky server.

    Args:
        data (str | list | tuple): One command line, or a sequence of command
            lines, forwarded by the server. Other types are logged and ignored.
    """
    proxy = active_proxy()
    if not proxy:
        return

    if isinstance(data, str):
        commands_to_process = [data]
    elif isinstance(data, (list, tuple)):
        commands_to_process = list(data)
    else:
        logger.warning(f"Unexpected STACK data format: {type(data)}")
        return

    for cmdline in commands_to_process:
        if isinstance(cmdline, str) and cmdline.strip():
            _process_server_command(proxy, cmdline)

def echo(text, flags=None, sender_id=None):
    """Handle ECHO messages (command responses) from the simulation.

    BlueSky's echo flags are an error bitmask (``BS_ARGERR=1``,
    ``BS_FUNERR=2``, ``BS_CMDERR=4`` — see ``bluesky/__init__.py``; its stack
    only ever sets them on failed commands), while the web client renders
    flags as severity levels (0 info, 1 error, 2 warning). Any set bit
    therefore maps to the error level. Delegates to ``_echo_response``, which
    emits immediately — command responses are never throttled.

    Args:
        text (str): The echo text; newlines and formatting are preserved.
        flags (int | None): BlueSky echo flags; nonzero marks an error.
        sender_id (bytes | str | None): ID of the node that sent the echo,
            forwarded in the same hex form as ``node_info`` payloads.
    """
    proxy = active_proxy()
    if not proxy:
        return

    text = str(text) if text is not None else ""
    level = 1 if flags else 0
    proxy._echo_response(text, level, sender=id2str(sender_id))

def on_routedata_received(data):
    """Handle ROUTEDATA events carrying an aircraft's route.

    Serializes the route and emits a ``routedata`` event to connected web
    clients. Frames that carry waypoints are only forwarded when the aircraft
    is part of the active node's traffic, preventing flicker when switching
    between nodes. Waypoint-less frames are BlueSky's route-clear broadcasts
    (route display toggled off, or the aircraft was deleted) and are always
    forwarded — the aircraft they refer to may already be gone from the
    traffic, and clients need them to drop their cached route.

    Args:
        data (dict): ROUTEDATA payload with the aircraft ID (``acid``) and,
            for route updates, waypoint arrays (``wplat``, ``wplon``, ...).
    """
    proxy = active_proxy()
    if not proxy:
        return

    if not proxy._get_safe_active_node():
        logger.debug("Route data ignored - no active node available")
        return

    route_aircraft_id = data.get("acid") if isinstance(data, dict) else None
    if not route_aircraft_id:
        logger.debug("Route data ignored - no aircraft ID found")
        return

    wplat = data.get("wplat")
    has_waypoints = wplat is not None and len(wplat) > 0
    if has_waypoints and proxy.traffic_data:
        if route_aircraft_id not in proxy.traffic_data.get("id", []):
            return

    route_data = make_json_serializable(data)

    if proxy.socketio and proxy.connected_clients > 0:
        try:
            proxy.socketio.emit("routedata", route_data)
        except Exception:
            # Emission errors (e.g. disconnected clients) are non-fatal
            pass

def on_reset_received(data=None, *args, sender_id=None, **kwargs):
    """Handle RESET events from the BlueSky server.

    Clears the stored polygon/polyline shapes for the node that sent the
    reset. Browsers display the active node only, so when the resetting node
    is the active one this also drops the cached aircraft and sim info and
    emits the map-clearing ``acdata``/``poly``/``polyline`` payloads plus the
    ``reset`` event — while a background node's reset must not wipe the active
    node's display. When the sender or active node can't be resolved, the
    reset is accepted so a single-node display still works (same fallback as
    the SIMINFO/ACDATA active-node filter).

    Args:
        data (Any): Optional RESET payload (unused).
        *args (Any): Additional positional payload items (unused).
        sender_id (bytes | str | None): Node that reset, from the message
            header; bytes are converted to a hex string. The shared network
            context is deliberately not consulted — it holds the sender of the
            last shared-state message, not of this RESET.
        **kwargs (Any): Additional keyword payload items (unused).
    """
    proxy = active_proxy()
    if not proxy:
        return

    try:
        sender_id = id2str(sender_id)

        reset_node_id = sender_id or proxy._get_safe_active_node()
        if not reset_node_id:
            return

        # Only the resetting node's stored shapes are stale.
        proxy.poly_data_by_node.pop(reset_node_id, None)
        proxy.polyline_data_by_node.pop(reset_node_id, None)

        if not is_active_node(proxy, sender_id):
            return

        # The active node's cached aircraft and sim info are stale too. Clear
        # both even with no browser connected — the backup timer and the next
        # initial_data snapshot serve these caches verbatim, so leaving them
        # would resurrect pre-reset aircraft (ghosts) and the pre-reset
        # clock/scenario in a header the browser just cleared.
        cleared = empty_traffic_data()
        proxy.traffic_data = cleared
        proxy.sim_data = {}

        if proxy.socketio and proxy.connected_clients > 0:
            proxy.socketio.emit("acdata", cleared)
            proxy.socketio.emit("poly", {"polys": {}})
            proxy.socketio.emit("polyline", {"polys": {}})
            proxy.socketio.emit(
                "reset",
                {"reason": "BlueSky simulation reset", "timestamp": time.time()},
            )

    except Exception as e:
        logger.error(f"Error processing RESET data: {e}")

def on_request_received(data, *args, **kwargs):
    """Handle REQUEST events from the BlueSky server.

    Currently only logs the payload; specific request handling is not yet
    implemented.

    Args:
        data (Any): The REQUEST payload.
        *args (Any): Additional positional payload items (unused).
        **kwargs (Any): Additional keyword payload items (unused).
    """
    if not active_proxy():
        return

    # TODO: Implement specific request handling logic
    logger.debug(f"REQUEST data received: {data}")

def on_plot_received(*args, **kwargs):
    """Log PLOT events; plot display is not yet implemented."""
    if not active_proxy():
        return

    logger.debug(f"PLOT data received: args={args} kwargs={kwargs}")

def on_showdialog_received(*args, **kwargs):
    """Log SHOWDIALOG events; web dialog display is not yet implemented."""
    if not active_proxy():
        return

    logger.debug(f"SHOWDIALOG data received: args={args} kwargs={kwargs}")

def on_simsettings_received(*args, **kwargs):
    """Log SIMSETTINGS events; settings handling is not yet implemented."""
    if not active_proxy():
        return

    logger.debug(f"SIMSETTINGS data received: args={args} kwargs={kwargs}")

def on_trails_received(*args, **kwargs):
    """Log TRAILS events; trail visualization is not yet implemented."""
    if not active_proxy():
        return

    logger.debug(f"TRAILS data received: args={args} kwargs={kwargs}")

def on_defwpt_received(data=None, *args, **kwargs):
    """Log DEFWPT events; waypoint rendering is not yet implemented.

    Args:
        data (Any): Unwrapped DEFWPT shared-state payload describing the
            waypoint(s).
        *args (Any): Additional positional payload items (unused).
        **kwargs (Any): Additional keyword payload items (unused).
    """
    if not active_proxy():
        return

    logger.info(f"DEFWPT data received: {data}")