chuanqi/apilot/stubs/netclient.py

"""netClient C++ 单例的 stub: 直接持有 Wire,负责收发 Frame。"""
from __future__ import annotations

from apilot.wire import Frame, Wire


class NetClientStub:
    def __init__(self) -> None:
        self.wire: Wire | None = None
        self._sdata: int = 0

    def connect(self, host: str, port: int, timeout: float = 10) -> None:
        self.wire = Wire(host, port, timeout)

    def disconnect(self) -> None:
        if self.wire is not None:
            self.wire.close()
        self.wire = None

    def is_connected(self) -> bool:
        return self.wire is not None

    def send_raw(self, data: bytes) -> None:
        assert self.wire is not None, "netClient not connected"
        self.wire.sock.sendall(data)

    def send_frame(self, f: Frame) -> None:
        assert self.wire is not None, "netClient not connected"
        self.wire.send(f)

    def tick(self, deadline_s: float) -> list[Frame]:
        if self.wire is None:
            return []
        return self.wire.recv(deadline_s)

    def set_sdata(self, v: int) -> None:
        self._sdata = int(v)

    def get_sdata(self) -> int:
        return self._sdata