Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 24 additions & 5 deletions matrixbox_simulator/device/cpstubs/socketpool.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
actually go out rather than being re-simulated.
"""

import errno
import os
import socket as _socket

Expand All @@ -11,13 +12,26 @@
_HTTP_PORT_REMAP = int(os.environ.get("MATRIXBOX_SIMULATOR_HTTP_PORT", "8080"))


def _require_connected(radio: object) -> None:
# Real matrixbox code (e.g. fetch_data.fetch) very often opens a socket
# straight away without checking wifi.radio.connected first, relying on
# the connection attempt itself to fail when there's no network. The
# host machine always has real network access, so that failure has to
# be injected here instead, or toggling `connected` off from the sim's
# own controls would be invisible to that code.
if not getattr(radio, "connected", True):
raise OSError(errno.ENETUNREACH, "Network is unreachable")


class _Socket:
"""Wraps a real socket, only overriding bind() for the port 80 remap.
Everything else (connect, send, recv_into, settimeout, ...) passes
straight through to the real socket object."""
"""Wraps a real socket, only overriding bind() for the port 80 remap
and connect() to fail while the radio is simulated as disconnected.
Everything else (send, recv_into, settimeout, ...) passes straight
through to the real socket object."""

def __init__(self, sock: _socket.socket) -> None:
def __init__(self, sock: _socket.socket, radio: object) -> None:
self._sock = sock
self._radio = radio

def bind(self, address: tuple[str, int]) -> None:
host, port = address
Expand All @@ -26,6 +40,10 @@ def bind(self, address: tuple[str, int]) -> None:

return self._sock.bind((host, port))

def connect(self, address: tuple[str, int]) -> None:
_require_connected(self._radio)
return self._sock.connect(address)

def __getattr__(self, name: str) -> object:
return getattr(self._sock, name)

Expand All @@ -49,7 +67,7 @@ def __init__(self, radio: object) -> None:
def socket(
self, family: int = _socket.AF_INET, type: int = _socket.SOCK_STREAM
) -> _Socket:
return _Socket(_socket.socket(family, type))
return _Socket(_socket.socket(family, type), self._radio)

def getaddrinfo(
self,
Expand All @@ -60,4 +78,5 @@ def getaddrinfo(
proto: int = 0,
flags: int = 0,
) -> list[tuple]:
_require_connected(self._radio)
return _socket.getaddrinfo(host, port, family or _socket.AF_INET, type)
52 changes: 49 additions & 3 deletions matrixbox_simulator/device/cpstubs/wifi.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,35 @@
f"127.0.0.1:{os.environ.get('MATRIXBOX_SIMULATOR_HTTP_PORT', '8080')}"
)

# --no-wifi boots the radio already disconnected, for testing an app's
# offline behavior from the very first frame rather than toggling it live
# with 'n' after boot. connect() refuses to succeed while disconnected
# (see below), so nothing an app does on its own — including the wifi
# setup page's own Connect button — ever flips this back on; it stays
# offline for the whole run, same as the live toggle would leave it.
_START_CONNECTED = os.environ.get("MATRIXBOX_SIMULATOR_NO_WIFI", "0") != "1"


class _ApInfo:
def __init__(self, rssi: int) -> None:
self.rssi = rssi


class _Network:
def __init__(self, ssid: str, channel: int) -> None:
self.ssid = ssid
self.channel = channel


# The sim has no way to scan real nearby networks portably, so the wifi
# setup page (matrixbox's own connect_to_wifi(), reached by disconnecting)
# is offered this fixed stand-in list instead of a real scan result.
_FAKE_SCAN_RESULTS = [_Network(ssid="matrixbox-simulator", channel=1)]


class Radio:
def __init__(self) -> None:
self.connected: bool = True
self.connected: bool = _START_CONNECTED
self.ap_active: bool = False
self.mac_address: bytes = bytes([0x02, 0x00, 0x00, 0x45, 0x53, 0x50])
self.tx_power: float = 0.0
Expand All @@ -32,12 +52,22 @@ def __init__(self) -> None:
# lands in matrixbox's own "4 of 5" signal-bar bracket (see
# web_interface._sig_bars) instead of the "no signal" 0 bars that
# an absent ap_info used to fall back to.
self.ap_info: _ApInfo | None = _ApInfo(rssi=-50)
self.ap_info: _ApInfo | None = _ApInfo(rssi=-50) if _START_CONNECTED else None

def connect(
self, ssid: str, password: str, *, channel: int = 0, timeout: float = 15
) -> None:
pass
# The sim has no real SSID/password to validate against, so it
# can't tell a wrong password from a wrong network name the way
# real hardware's own error strings do (see matrixbox's own
# connect_to_network(), which maps those apart). While simulated
# as disconnected, every attempt just fails outright instead of
# silently faking success — matching there being no real network
# to associate with at all, and keeping matrixbox from writing a
# "connected" settings.txt for a connection that never really
# happened.
if not self.connected:
raise ConnectionError("simulator offline")

def start_ap(self, ssid: str, *args: object, **kwargs: object) -> None:
self.ap_active = True
Expand All @@ -51,5 +81,21 @@ def stop_dhcp(self) -> None:
def set_ipv4_address(self, **kwargs: object) -> None:
pass

def start_scanning_networks(
self, *, start_channel: int = 1, stop_channel: int = 11
) -> list[_Network]:
return _FAKE_SCAN_RESULTS

def stop_scanning_networks(self) -> None:
pass


radio = Radio()


def set_connected(value: bool) -> None:
# Flipped live from the sim's own controls (not app code) to test how
# an app behaves with no internet. ap_info drops out along with it,
# matching a real radio that's no longer associated with any AP.
radio.connected = value
radio.ap_info = _ApInfo(rssi=-50) if value else None
29 changes: 28 additions & 1 deletion matrixbox_simulator/device/run_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -502,7 +502,7 @@ def restart_process(reason: str = "to apply the new panel geometry") -> NoReturn
def _controls_hint() -> str:
return (
"'s'/'l' button, '+'/'-' refresh-fps, '['/']' gamma, 'z' cycle size, "
"'r' reload (restarts)"
"'n' toggle wifi, 'r' reload (restarts)"
)


Expand Down Expand Up @@ -589,6 +589,19 @@ def _bump_gamma(direction: int) -> None:
print(f"matrixbox-simulator: gamma now {label}")


def _toggle_wifi() -> None:
# Same reaching-into-sys.modules trick as the gamma/refresh-fps bumps,
# to hit the exact wifi module instance the running app imported.
module = sys.modules.get("wifi")
if module is None:
print("matrixbox-simulator: nothing running yet to adjust")
return

new = not module.radio.connected
module.set_connected(new)
print(f"matrixbox-simulator: wifi now {'connected' if new else 'disconnected'}")


def _run_kernel(
framework_root: Path, args: argparse.Namespace, app_dir: Path | None = None
) -> None:
Expand Down Expand Up @@ -645,6 +658,8 @@ def _run_kernel(

http_port = int(os.environ.get("MATRIXBOX_SIMULATOR_HTTP_PORT", "8080"))
print(f"matrixbox-simulator: web UI at http://127.0.0.1:{http_port}/")
if args.no_wifi:
print("matrixbox-simulator: booting with wifi disconnected (--no-wifi)")

main_path = staged_root / "main.py"
os.chdir(staged_root) # goes through tracked_chdir, seeds sys.path[0]
Expand Down Expand Up @@ -744,6 +759,8 @@ def listen() -> None:
_bump_gamma(-1)
elif char == "z" and cycle_size is not None:
cycle_size()
elif char == "n":
_toggle_wifi()

thread = threading.Thread(target=listen, daemon=True)
thread.start()
Expand Down Expand Up @@ -826,6 +843,15 @@ def build_parser(
"try 1.8-2.8 and adjust live with '['/']'"
),
)
parser.add_argument(
"--no-wifi",
action="store_true",
help=(
"boot with wifi already disconnected, to test an app's offline "
"behavior from the first frame rather than toggling it live "
"with 'n' after boot. Stays offline for the whole run"
),
)
return parser


Expand Down Expand Up @@ -857,6 +883,7 @@ def run(args: argparse.Namespace) -> None:

os.environ["MATRIXBOX_SIMULATOR_REFRESH_FPS"] = str(args.refresh_fps)
os.environ["MATRIXBOX_SIMULATOR_GAMMA"] = str(args.gamma)
os.environ["MATRIXBOX_SIMULATOR_NO_WIFI"] = "1" if args.no_wifi else "0"

given = Path(args.app).expanduser()
if given.is_dir() and _is_os_root(given.resolve()):
Expand Down