rjmcmahon
/
FiWiManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fi-Wi: site setup, relay module, map/SSH consolidation 

- Add site_setup entry (root shim + fiwi.site_setup) with SiteSetupApp and in-process
  relay/calibrate; FiWiRelayBootstrapApp in fiwi_relay/relay_app.py.
- CLI: run_panel_calibrate, parse_panel_calibrate_argv, shared BrainStem load helper;
  fiber_map_io.resolve_calibrate_ssh_targets for one merge of calibrate SSH hosts.
- Maps default under maps/; config and remote_ssh.env.example updates; concentrator
  and panel calibrate fixes; radiohead refactor; tests/docs refresh.
- Remove fiber_radio_port.py (unused example), empty fiwi_link stub, orphan peak_detect.reflex.

Made-with: Cursor
This commit is contained in:
Robert McMahon 2026-04-04 18:18:27 -07:00
parent 93cd346a07
commit 957078039d
26 changed files with 1918 additions and 397 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -27,6 +27,8 @@ htmlcov/
# Operator maps; INI profiles live in config/ (see config/default.ini). SSH dotenv: remote_ssh.env.example
fiber_map.json
fiber_map.json.bak*
maps/fiber_map.json
maps/fiber_map.json.*
*.json.bak*
panel_map.json
remote_ssh.env

5
config/default.ini
View File

@ -7,10 +7,11 @@
description = Generic default profile
[paths]
fiber_map = fiber_map.json
; Under the Fi-Wi install root (directory containing fiwi.py). Backups: same folder, timestamp suffix.
fiber_map = maps/fiber_map.json
[patch_panel]
; Front-panel port count when fiber_map.json has no patch_panel.slots yet
; Front-panel port count when the map has no patch_panel.slots yet
default_ports = 24
[remote_ssh]

2
config/uax24.ini
View File

@ -4,7 +4,7 @@
description = UAX 24-port USB concentrator
[paths]
fiber_map = fiber_map.json
fiber_map = maps/fiber_map.json
[patch_panel]
; Front-panel ports for this concentrator when map has no patch_panel yet

2
config/uax4.ini
View File

@ -4,7 +4,7 @@
description = UAX 4-port USB concentrator
[paths]
fiber_map = fiber_map.json
fiber_map = maps/fiber_map.json
[patch_panel]
; Front-panel ports for this concentrator when map has no patch_panel yet

6
docs/fiwi-design.md
View File

@ -12,14 +12,14 @@ The Fi-Wi stack ties together **USB power-control hubs** (BrainStem API), a **pa
|--------|------|
| `fiwi.py` | Entry point; configures `fiwi.paths` with the install directory, then runs `fiwi.cli.main`. |
| `fiwi/` | Library: concentrator, SSH transport, fiber map I/O, USB/wireless probes, diagnostics. |
| `fiber_map.json` | Lives next to `fiwi.py` (see `fiwi.paths`). Operator-edited or calibrate-produced. |
| `maps/fiber_map.json` | Default path under the install root (see `fiwi.paths`; override with `FIWI_FIBER_MAP`). Operator-edited or calibrate-produced; timestamped backups sit in `maps/` too. |
| `config/*.ini` | Optional profiles under `config/` (`default.ini` when `FIWI_CONFIG` unset). `[remote_hubs] hosts` sets `FIWI_REMOTE_HUBS` (merged with `FIWI_CALIBRATE_REMOTES` for panel calibrate). Merged after `remote_ssh.env` via `setdefault`. |
| `remote_ssh.env` / `.fiwi_remote` | Optional dotenv next to the install; merged into process env for SSH defaults (see `SshNodeConfig`). |
## Core components
- **`FiWiConcentrator`** (`fiwi/concentrator.py`): Main Fi-Wi object — BrainStem discovery, power, reboot, panel + fiber workflows, and **panel calibrate** (interactive and hybrid local/remote ordering).
- **`FiberRadioPort`** (`fiwi/fiber_radio_port.py`): View of one `fiber_ports[]` entry: power routing, SSH node resolution, previews for map-driven UIs.
- **`RadioHeadEntry`** (`fiwi/radiohead.py`): View of one `fiber_ports[]` record (map key + data). **`RadioHead`** pairs that with a `FiWiConcentrator` for power/inrush (via `patch_panel().head` / `heads`).
- **`SshNode` / `SshNodeConfig`** (`fiwi/ssh.py`): Builds `ssh` argv (optional `FIWI_SSH_OPTS`), runs remote `FIWI_REMOTE_PYTHON` + `FIWI_REMOTE_SCRIPT` with forwarded subcommands, or **raw** `ssh target …` for helpers like `dmesg` / `lspci` in diagnostics. Supports **deferred** capture (`RemoteCallHandle`, `asyncio.Task`) when `defer=True` or `FIWI_REMOTE_DEFER` / `--async` so overlapping subprocesses do not require Python threads.
- **`ssh_dispatch`** (`fiwi/ssh_dispatch.py`): Early CLI path: if `fiber_map.json` routes a **power fiber-port** (or certain fiber flows) to another host, dispatch runs `SshNode.invoke` **without** importing BrainStem on the workstation.
- **`fiber_map_io`**: Load/save map, PCIe prompt helpers during calibrate, previews.
@ -40,7 +40,7 @@ A **central, append-only log** records ordered events (`NoteEvent`, `DmesgEvent`
## Extension points
- New **CLI subcommands**: `fiwi/cli.py` dispatch; heavy logic stays in `FiWiConcentrator` or focused modules.
- New **map-driven behavior**: extend `fiber_ports` schema carefully; teach `FiberRadioPort` / `ssh_dispatch` if routing changes.
- New **map-driven behavior**: extend `fiber_ports` schema carefully; teach `RadioHeadEntry` / `ssh_dispatch` if routing changes.
- New **remote machine-readable commands**: add branches in `fiwi/cli.py` (remote runs the same script).
- New **diagnostic event kinds**: add a frozen dataclass, widen `DiagEvent`, keep `dump_jsonl` as dict-per-line.

2
docs/fiwi-test-authoring.md
View File

@ -7,7 +7,7 @@ The repository does not yet ship a pytest suite for `fiwi/`; the patterns below
## Choose a control style
1. **Shell / subprocess** — Easiest to run exactly what operators run. Parse JSON from stdout for machine-readable commands (`calibrate-ports-json`, `lsusb-lines-json`, `wlan-info-json`). Stable for CI-style wrappers.
2. **Python imports** — Use `FiWiConcentrator`, `SshNode`, `FiberRadioPort`, and `fiwi.diag_log` for tighter integration, error handling, and async (`ainvoke_capture`, `alog_dmesg`, and so on).
2. **Python imports** — Use `FiWiConcentrator`, `SshNode`, `RadioHead` / `RadioHeadEntry`, and `fiwi.diag_log` for tighter integration, error handling, and async (`ainvoke_capture`, `alog_dmesg`, and so on).
Mix both: for example import `SshNode` to run remote JSON probes while keeping local steps in the shell.

10
fiber_map.example.json
View File

@ -1,5 +1,13 @@
{
"patch_panel": { "slots": 24, "label": "Example rack" },
"fiwi_site": {
"concentrator_name": "example-concentrator",
"concentrator_location": "Lab / rack label"
},
"patch_panel": {
"slots": 24,
"label": "Example rack",
"location": "Lab bench A"
},
"calibrate_remotes": ["rjmcmahon@192.168.1.39"],
"fiber_ports": {
"1": {

30
fiwi.py
View File

@ -4,6 +4,36 @@
import os
import sys
def _consume_early_config_argv(argv: list[str]) -> list[str]:
"""
Apply ``-c`` / ``--config`` (INI profile or ``*.ini`` path) as ``FIWI_CONFIG``, and drop those
tokens so :mod:`getopt` in :mod:`fiwi.cli` does not treat ``-c`` as unknown.
"""
out: list[str] = []
i = 0
while i < len(argv):
a = argv[i]
if a in ("-c", "--config"):
if i + 1 < len(argv):
os.environ["FIWI_CONFIG"] = argv[i + 1].strip()
i += 2
continue
out.append(a)
i += 1
continue
if a.startswith("--config="):
os.environ["FIWI_CONFIG"] = a.partition("=")[2].strip()
i += 1
continue
out.append(a)
i += 1
return out
if len(sys.argv) > 1:
sys.argv[1:] = _consume_early_config_argv(sys.argv[1:])
import fiwi.paths as _paths
_paths.configure(os.path.dirname(os.path.abspath(__file__)))

16
fiwi/__init__.py
View File

@ -18,9 +18,10 @@ from fiwi.diag_log import (
kernel_dump_event,
note_event,
)
from fiwi.fiber_radio_port import FiberRadioPort
from fiwi.concentrator import FiWiConcentrator
from fiwi.patch_panel import PatchPanel
from fiwi.radiohead import InrushSample, RadioHead, RadioHeadEntry, load_radio_head_entries
from fiwi.concentrator import BoundHubPlane, FiWiConcentrator, HubDownstreamPort
from fiwi.patch_panel import BoundPatchPanel, PatchPanel
from fiwi.site_setup import open_fiwi_stack
from fiwi.ssh import (
FetchCalibratePortsHandle,
RemoteCallHandle,
@ -36,9 +37,15 @@ __all__ = [
"DiagLog",
"DmesgEvent",
"KernelDumpEvent",
"FiberRadioPort",
"BoundHubPlane",
"BoundPatchPanel",
"FetchCalibratePortsHandle",
"FiWiConcentrator",
"HubDownstreamPort",
"InrushSample",
"RadioHead",
"RadioHeadEntry",
"load_radio_head_entries",
"NoteEvent",
"PatchPanel",
"PcieEvent",
@ -57,5 +64,6 @@ __all__ = [
"get_diag_log",
"kernel_dump_event",
"note_event",
"open_fiwi_stack",
"resolve_remote_defer",
]

110
fiwi/cli.py
View File

@ -108,7 +108,8 @@ def _parse_global_cli(argv: list[str]) -> _GlobalCli | None:
def _print_cli_help() -> None:
print(
"Fi-Wi test framework — CLI\n"
"Usage: fiwi.py [--async] <command> [target]\n"
"Usage: fiwi.py [-c PROFILE] [--async] <command> [target]\n"
" -c / --config PROFILE set FIWI_CONFIG before loading config/*.ini (same as site_setup.py)\n"
" --async set FIWI_REMOTE_DEFER: deferred calls spawn ssh child processes immediately;\n"
" panel calibrate overlaps them (join via handle.result(); no Python threads).\n"
" Or set FIWI_REMOTE_DEFER=1 / remote_ssh.env / config/*.ini (FIWI_CONFIG=profile).\n"
@ -117,6 +118,7 @@ def _print_cli_help() -> None:
" discover — USB power-control hubs (serial, port count); no port I/O\n"
" show_hostcards — same as discover, concentrator 'hostcards' label\n"
" port-metrics-json — JSON list: per-port power, current (mA), voltage (V) on connected hubs\n"
" hub-inrush-json <hub.port> — JSON {peak_ma, duration_ms} short current capture (local hubs)\n"
" usb-hub-tty-json — JSON: Acroname (24ff) hubs: serial, tty[], bus, dev, id, manufacturer, product\n"
" show_radioheads — all fiber_ports rows + power (same table as fiber status)\n"
" status [target] — default command; target like all, 1.3, all.2\n"
@ -149,10 +151,14 @@ def _print_cli_help() -> None:
)
def _parse_panel_calibrate_argv(args):
def parse_panel_calibrate_argv(args: list[str]) -> tuple[bool, int | None, list[str]]:
"""
Parse argv tail after ``panel calibrate``:
``panel calibrate [merge] [N] [--ssh user@host] …``
Returns ``(merge, limit, calibrate_ssh_hosts)``.
Returns ``(merge, limit, calibrate_ssh_hosts)``. On bad tokens, prints to stderr and raises
:class:`SystemExit` (2) — intended for CLI use; call :func:`run_panel_calibrate` with explicit
keyword args when embedding.
"""
merge = False
limit = None
@ -181,6 +187,68 @@ def _parse_panel_calibrate_argv(args):
return merge, limit, hosts
def _try_load_brainstem_for_cli() -> int | None:
"""
Import BrainStem for local USB hub control.
Returns:
``None`` on success, or ``1`` after printing the usual install hints.
"""
try:
load_brainstem()
except Exception as exc:
print(f"fiwi: failed to import brainstem: {exc}", file=sys.stderr, flush=True)
if isinstance(exc, ImportError):
print(
" Install the BrainStem package for **this** Python (the one running fiwi.py), e.g.\n"
" ./env/bin/python3 -m pip install -r requirements.txt\n"
" or: python3 -m pip install --user -r requirements.txt\n"
" Check with: python3 -c \"import brainstem\" (same python you use for fiwi.py).\n"
" If you ran `fiwi.py --ssh` from another machine, set FIWI_REMOTE_PYTHON there to\n"
" a Python on **this** host where brainstem is installed (e.g. ~/…/env/bin/python3).",
file=sys.stderr,
flush=True,
)
return 1
return None
def run_panel_calibrate(
*,
merge: bool = False,
limit: int | None = None,
calibrate_ssh_hosts: list[str] | None = None,
emit_start_line: bool = False,
) -> int:
"""
Run :meth:`~fiwi.concentrator.FiWiConcentrator.panel_calibrate` in-process (same work as
``python fiwi.py panel calibrate …`` for local hubs + SSH walk).
Expects :mod:`fiwi.paths` to be configured (e.g. ``paths.configure(install_root)``) and
``FIWI_CONFIG`` set if you use profiles. Does not re-parse ``sys.argv``.
Returns:
``0`` on normal completion. May raise :class:`SystemExit` for the same reasons as the CLI
(e.g. Ctrl-C during calibrate uses exit 130 inside ``panel_calibrate``).
"""
if emit_start_line and sys.stderr.isatty():
os.write(2, b"fiwi: start\n")
bs_rc = _try_load_brainstem_for_cli()
if bs_rc is not None:
return bs_rc
hosts = [s.strip() for s in (calibrate_ssh_hosts or []) if (s or "").strip()]
concentrator = FiWiConcentrator()
try:
concentrator.panel_calibrate(
merge=merge,
limit=limit,
calibrate_ssh_hosts=hosts,
)
finally:
concentrator.disconnect()
return 0
def main() -> int:
ga = _parse_global_cli(sys.argv[1:])
if ga is None:
@ -217,22 +285,9 @@ def main() -> int:
# Skip when stderr is a pipe (SSH capture, subprocess): avoids interleaving with stdout consumers.
if sys.stderr.isatty():
os.write(2, b"fiwi: start\n")
try:
load_brainstem()
except Exception as exc:
print(f"fiwi: failed to import brainstem: {exc}", file=sys.stderr, flush=True)
if isinstance(exc, ImportError):
print(
" Install the BrainStem package for **this** Python (the one running fiwi.py), e.g.\n"
" ./env/bin/python3 -m pip install -r requirements.txt\n"
" or: python3 -m pip install --user -r requirements.txt\n"
" Check with: python3 -c \"import brainstem\" (same python you use for fiwi.py).\n"
" If you ran `fiwi.py --ssh` from another machine, set FIWI_REMOTE_PYTHON there to\n"
" a Python on **this** host where brainstem is installed (e.g. ~/…/env/bin/python3).",
file=sys.stderr,
flush=True,
)
return 1
bs_rc = _try_load_brainstem_for_cli()
if bs_rc is not None:
return bs_rc
concentrator = FiWiConcentrator()
try:
cmd = pos[0].lower() if pos else "status"
@ -241,6 +296,21 @@ def main() -> int:
concentrator.status(target)
elif cmd == "port-metrics-json":
print(json.dumps(concentrator.port_metrics_snapshot()), flush=True)
elif cmd == "hub-inrush-json":
if len(pos) < 2:
print(
"Usage: fiwi.py hub-inrush-json <hub.port>\n"
" Example: fiwi.py hub-inrush-json 1.0 (hub 1, port 0; local BrainStem only)",
file=sys.stderr,
flush=True,
)
return 2
try:
sample = concentrator.sample_inrush_hub_port_str(pos[1])
except (ConnectionError, ValueError) as exc:
print(f"fiwi: {exc}", file=sys.stderr, flush=True)
return 1
print(json.dumps(sample.as_json_dict()), flush=True)
elif cmd == "calibrate-ports-json":
if not concentrator.hubs and not concentrator.connect():
print("[]", flush=True)
@ -335,7 +405,7 @@ def main() -> int:
concentrator.panel_status()
elif sub == "calibrate":
cal_args = pos[2:]
merge, limit, cal_hosts = _parse_panel_calibrate_argv(cal_args)
merge, limit, cal_hosts = parse_panel_calibrate_argv(cal_args)
concentrator.panel_calibrate(merge=merge, limit=limit, calibrate_ssh_hosts=cal_hosts)
elif sub in ("on", "off"):
if len(pos) < 3:

432
fiwi/concentrator.py
View File

@ -1,5 +1,7 @@
"""Fi-Wi concentrator: BrainStem USB power, fiber/radio map, patch panel, remote nodes."""
from __future__ import annotations
import asyncio
import copy
import json
@ -7,15 +9,15 @@ import os
import shutil
import sys
import time
from dataclasses import dataclass
import fiwi.brainstem_loader as stemmod
from fiwi.patch_panel import PatchPanel, default_panel_ports, effective_panel_slots
from fiwi.patch_panel import BoundPatchPanel, PatchPanel, default_panel_ports, effective_panel_slots
from fiwi.paths import fiber_map_path
from fiwi.fiber_radio_port import FiberRadioPort
from fiwi.radiohead import InrushSample, RadioHeadEntry
from fiwi import fiber_map_io as fm
from fiwi.ssh import (
SshNode,
SshNodeConfig,
parse_status_line_for_hub_port,
resolve_remote_defer,
)
@ -28,7 +30,8 @@ class FiWiConcentrator:
Main Fi-Wi object: BrainStem-driven USB power-control hub plane (host Python API). On-device
**Reflex** programs are separate — compile with SDK **arc** (``fiwi.py reflex compile``), load
with ReflexLoader / HubTool; FiWi still uses ``brainstem`` for port power and metrics.
``FiberRadioPort`` / :class:`fiwi.ssh.SshNode` routing, and calibration.
:class:`~fiwi.radiohead.RadioHeadEntry` (map) and :class:`~fiwi.radiohead.RadioHead` (via :meth:`patch_panel`),
:class:`HubDownstreamPort` (via :meth:`hub_plane`), and :class:`fiwi.ssh.SshNode` routing.
Local USB reboot staggering uses :mod:`asyncio`. Remote SSH work during ``panel calibrate``
(fetching hub/port lists and baseline power-off) runs concurrent ``SshNode`` coroutines via
@ -380,6 +383,19 @@ class FiWiConcentrator:
print("Radioheads — fiber_map.json fiber_ports (per-strand attachments)", flush=True)
self.fiber_map_status()
def patch_panel(self, doc: dict | None = None) -> BoundPatchPanel:
"""Bound panel + this concentrator + map — use :meth:`~fiwi.patch_panel.BoundPatchPanel.head` / ``heads``."""
d = doc if doc is not None else fm.load_fiber_map_or_exit()
blob = d.get("patch_panel")
pp = PatchPanel.from_map_blob(blob)
if pp is None:
pp = PatchPanel(slots=effective_panel_slots(d), label="")
return pp.bound(self, d)
def hub_plane(self) -> BoundHubPlane:
"""Local USB hub downstream ports — use :meth:`~BoundHubPlane.port` / :meth:`~BoundHubPlane.port_at`."""
return BoundHubPlane(self)
def _sample_inrush(self, stem, port, sample_duration=0.3):
"""Captures peak current and duration for the reboot report."""
samples = []
@ -456,6 +472,13 @@ class FiWiConcentrator:
)
return rows
def hub_port_metrics(self, hub_1: int, port_0: int) -> dict[str, object] | None:
"""One row from :meth:`port_metrics_snapshot` for ``hub_1`` / ``port_0``, or ``None``."""
for row in self.port_metrics_snapshot():
if row.get("hub") == hub_1 and row.get("port") == port_0:
return row
return None
async def _async_reboot_port(self, h_idx, stem, port, delay, stagger, skip_empty):
"""Reboots a port and captures inrush vs steady state data."""
current_ma = stem.usb.getPortCurrent(port).value / 1000.0
@ -551,9 +574,9 @@ class FiWiConcentrator:
"""Rack positions 1…N (N from fiber_map patch_panel.slots or default): mapping and power."""
doc = fm.load_fiber_map_or_exit()
n_slots = effective_panel_slots(doc)
slot_frp = [FiberRadioPort.from_port_id(doc, n) for n in range(1, n_slots + 1)]
slot_entries = [RadioHeadEntry.from_port_id(doc, n) for n in range(1, n_slots + 1)]
need_local = any(
x.hub_port() is not None and x.ssh_target() is None for x in slot_frp
x.hub_port() is not None and x.ssh_target() is None for x in slot_entries
)
if need_local and not self.hubs and not self.connect():
return
@ -563,12 +586,12 @@ class FiWiConcentrator:
flush=True,
)
print("-" * 120)
for idx, frp in enumerate(slot_frp):
for idx, map_ent in enumerate(slot_entries):
panel_n = idx + 1
tup = frp.hub_port()
ssh = frp.ssh_target()
chip_s = frp.chip_preview()
pcie_s = frp.pcie_preview()
tup = map_ent.hub_port()
ssh = map_ent.ssh_target()
chip_s = map_ent.chip_preview()
pcie_s = map_ent.pcie_preview()
if tup is None:
print(
f"{panel_n:<7} | {'—':<10} | {'—':<18} | {'—':<5} | {'—':<8} | {chip_s:<28} | {pcie_s:<22}",
@ -627,15 +650,15 @@ class FiWiConcentrator:
if panel_1based < 1 or panel_1based > n_slots:
print(f"Panel port must be 1–{n_slots}, got {panel_1based}", file=sys.stderr, flush=True)
sys.exit(1)
frp = FiberRadioPort.from_port_id(doc, panel_1based)
if not frp.is_mapped():
map_ent = RadioHeadEntry.from_port_id(doc, panel_1based)
if not map_ent.is_mapped():
print(
f"Panel {panel_1based} is not mapped (no fiber_ports[{frp.map_key!r}] in fiber_map.json).",
f"Panel {panel_1based} is not mapped (no fiber_ports[{map_ent.map_key!r}] in fiber_map.json).",
file=sys.stderr,
flush=True,
)
sys.exit(1)
ssh = frp.ssh_target()
ssh = map_ent.ssh_target()
if ssh:
sys.exit(
SshNode.parse(ssh).invoke(
@ -643,7 +666,7 @@ class FiWiConcentrator:
defer=False,
)
)
hub_1, port_0 = frp.hub_port()
hub_1, port_0 = map_ent.hub_port()
assert hub_1 is not None and port_0 is not None
tgt = f"{hub_1}.{port_0}"
print(f"Panel {panel_1based} → hub target {tgt} ({mode})", flush=True)
@ -653,23 +676,23 @@ class FiWiConcentrator:
def fiber_power(self, mode, fiber_port):
"""Power via fiber_map.json fiber_ports key (any positive integer id)."""
doc = fm.load_fiber_map_or_exit()
frp = FiberRadioPort.from_port_id(doc, int(fiber_port))
if not frp.is_mapped():
map_ent = RadioHeadEntry.from_port_id(doc, int(fiber_port))
if not map_ent.is_mapped():
print(
f"Fiber port {fiber_port} is not mapped (missing fiber_ports[{frp.map_key!r}] in fiber_map.json).",
f"Fiber port {fiber_port} is not mapped (missing fiber_ports[{map_ent.map_key!r}] in fiber_map.json).",
file=sys.stderr,
flush=True,
)
sys.exit(1)
ssh = frp.ssh_target()
ssh = map_ent.ssh_target()
if ssh:
sys.exit(
SshNode.parse(ssh).invoke(
["power", "fiber-port", frp.map_key, mode.lower()],
["power", "fiber-port", map_ent.map_key, mode.lower()],
defer=False,
)
)
hub_1, port_0 = frp.hub_port()
hub_1, port_0 = map_ent.hub_port()
assert hub_1 is not None and port_0 is not None
tgt = f"{hub_1}.{port_0}"
print(f"Fiber port {fiber_port} → hub target {tgt} ({mode})", flush=True)
@ -682,16 +705,16 @@ class FiWiConcentrator:
SSH-mapped fibers use PCIe metadata from calibrate / fiber_map.json — not forwarded.
"""
doc = fm.load_fiber_map_or_exit()
frp = FiberRadioPort.from_port_id(doc, int(fiber_port))
key = frp.map_key
if not frp.is_mapped():
map_ent = RadioHeadEntry.from_port_id(doc, int(fiber_port))
key = map_ent.map_key
if not map_ent.is_mapped():
print(
f"Fiber port {fiber_port} is not mapped (missing fiber_ports[{key!r}] in fiber_map.json).",
file=sys.stderr,
flush=True,
)
sys.exit(1)
ssh = frp.ssh_target()
ssh = map_ent.ssh_target()
if ssh:
print(
"fiber chip: this fiber is SSH-mapped (PCIe/fiber path). Use panel calibrate PCIe prompts "
@ -709,7 +732,7 @@ class FiWiConcentrator:
sys.exit(1)
if not self.hubs and not self.connect():
return
hub_1, port_0 = frp.hub_port()
hub_1, port_0 = map_ent.hub_port()
assert hub_1 is not None and port_0 is not None
h_idx = hub_1 - 1
if h_idx < 0 or h_idx >= len(self.hubs):
@ -773,7 +796,7 @@ class FiWiConcentrator:
def fiber_map_status(self):
"""All fiber_ports entries with hub.port and live power (local BrainStem or ssh status)."""
doc = fm.load_fiber_map_or_exit()
all_frp = list(FiberRadioPort.each_from_document(doc))
all_entries = list(RadioHeadEntry.each_from_document(doc))
print(
f"{'Fiber':<8} | {'Hub.Port':<10} | {'Route':<18} | {'Pwr':<5} | {'mA':<8} | "
f"{'Chip (saved)':<28} | {'PCIe (saved)':<22}",
@ -781,16 +804,16 @@ class FiWiConcentrator:
)
print("-" * 120)
need_local = any(
frp.hub_port() is not None and frp.ssh_target() is None for frp in all_frp
e.hub_port() is not None and e.ssh_target() is None for e in all_entries
)
if need_local and not self.hubs and not self.connect():
return
for frp in all_frp:
key = frp.map_key
tup = frp.hub_port()
ssh = frp.ssh_target()
chip_s = frp.chip_preview()
pcie_s = frp.pcie_preview()
for map_ent in all_entries:
key = map_ent.map_key
tup = map_ent.hub_port()
ssh = map_ent.ssh_target()
chip_s = map_ent.chip_preview()
pcie_s = map_ent.pcie_preview()
if tup is None:
print(
f"{key!s:<8} | {'—':<10} | {'—':<18} | {'—':<5} | {'—':<8} | {chip_s:<28} | {pcie_s:<22}",
@ -849,15 +872,15 @@ class FiWiConcentrator:
if panel_1based < 1 or panel_1based > n_slots:
print(f"Panel port must be 1–{n_slots}, got {panel_1based}", file=sys.stderr, flush=True)
sys.exit(1)
frp = FiberRadioPort.from_port_id(doc, panel_1based)
if not frp.is_mapped():
map_ent = RadioHeadEntry.from_port_id(doc, panel_1based)
if not map_ent.is_mapped():
print(
f"Panel {panel_1based} is not mapped (no fiber_ports[{frp.map_key!r}] in fiber_map.json).",
f"Panel {panel_1based} is not mapped (no fiber_ports[{map_ent.map_key!r}] in fiber_map.json).",
file=sys.stderr,
flush=True,
)
sys.exit(1)
ssh = frp.ssh_target()
ssh = map_ent.ssh_target()
sub = "reboot" if skip_empty else "reboot-force"
if ssh:
sys.exit(
@ -866,7 +889,7 @@ class FiWiConcentrator:
defer=False,
)
)
hub_1, port_0 = frp.hub_port()
hub_1, port_0 = map_ent.hub_port()
assert hub_1 is not None and port_0 is not None
tgt = f"{hub_1}.{port_0}"
print(f"Panel {panel_1based} → hub target {tgt} ({sub})", flush=True)
@ -889,6 +912,46 @@ class FiWiConcentrator:
else:
stem.usb.setPortDisable(port_0)
def set_hub_port_power(self, hub_1: int, port_0: int, enable: bool) -> None:
"""
Enable or disable one local downstream USB port (1-based hub index, 0-based port index).
Connects to local hubs if needed. Used by :class:`~fiwi.radiohead.RadioHead` and tests.
"""
if not self.hubs and not self.connect():
raise ConnectionError("no USB power-control hubs connected")
h_idx = hub_1 - 1
if h_idx < 0 or h_idx >= len(self.hubs):
raise ValueError(f"hub {hub_1} is out of range (have {len(self.hubs)} hub(s))")
stem = self.hubs[h_idx]
n = self._port_count(stem)
if port_0 < 0 or port_0 >= n:
raise ValueError(f"hub {hub_1} has ports 0–{n - 1}; got {port_0}")
self._set_hub_port_power(hub_1, port_0, enable)
def sample_inrush_hub_port(
self, hub_1: int, port_0: int, *, sample_duration: float = 0.3
) -> InrushSample:
"""
Sample downstream current for ``sample_duration`` seconds on a **local** hub port.
Remote hub hosts use CLI ``hub-inrush-json`` (same JSON shape).
"""
if not self.hubs and not self.connect():
raise ConnectionError("no USB power-control hubs connected")
h_idx = hub_1 - 1
if h_idx < 0 or h_idx >= len(self.hubs):
raise ValueError(f"hub {hub_1} is out of range (have {len(self.hubs)} hub(s))")
stem = self.hubs[h_idx]
n = self._port_count(stem)
if port_0 < 0 or port_0 >= n:
raise ValueError(f"hub {hub_1} has ports 0–{n - 1}; got {port_0}")
raw = self._sample_inrush(stem, port_0, sample_duration)
return InrushSample(raw["peak"], raw["duration"])
def sample_inrush_hub_port_str(self, hub_dot_port: str, *, sample_duration: float = 0.3) -> InrushSample:
"""Parse ``hub.port`` (e.g. ``1.0``) and sample; for CLI ``hub-inrush-json``."""
hp = self.hub_plane().port_at(hub_dot_port)
return hp.measure_inrush(sample_duration=sample_duration)
def _calibrate_step_power_off(self, ssh_host, hub_1, port_0):
"""Turn off downstream port at end of one calibrate step (after PCIe prompts if mapped)."""
if ssh_host is None:
@ -906,6 +969,77 @@ class FiWiConcentrator:
flush=True,
)
@staticmethod
def _calibrate_interruptible_sleep(total_s: float, chunk_s: float = 0.25) -> None:
"""Sleep in small steps so Ctrl-C (SIGINT) is handled between BrainStem / blocking calls."""
if total_s <= 0:
return
end = time.monotonic() + total_s
while True:
remaining = end - time.monotonic()
if remaining <= 0:
return
time.sleep(min(chunk_s, remaining))
def _calibrate_power_down_all_ports(
self, remote_hosts: list[str], *, context: str = "Calibrate shutdown"
) -> None:
"""
Disable every local downstream port, then run ``fiwi off all`` on each SSH hub host.
Intended for calibrate cleanup / site-setup baseline so the rack is not left hot.
"""
if self.hubs:
print(
f"\n>>> {context}: all local downstream USB ports OFF…",
flush=True,
)
for stem in self.hubs:
n = self._port_count(stem)
for p in range(n):
try:
stem.usb.setPortDisable(p)
except Exception:
pass
self._calibrate_interruptible_sleep(0.15, 0.05)
seen: set[str] = set()
for host in remote_hosts:
h = str(host).strip()
if not h or h in seen:
continue
seen.add(h)
try:
print(f">>> {context}: `{h}` off all…", flush=True)
code, _out, err = SshNode.parse(h).invoke_capture(
["off", "all"], timeout=120, defer=False
)
if code != 0 and (err or "").strip():
print(err.strip()[:240], file=sys.stderr, flush=True)
except Exception as exc:
print(
f" ({context}: remote off all {h!r} failed: {exc})",
file=sys.stderr,
flush=True,
)
def power_down_all_downstream_ports(
self, remote_hosts: list[str] | None = None, *, context: str = "Site setup"
) -> None:
"""
Turn off every downstream port on **connected** local hubs, then run ``fiwi off all`` on
each ``user@host`` in *remote_hosts* (deduped). Opens local hubs via :meth:`connect` if none
are connected yet. Safe to call before site metadata edits or calibrate.
"""
if not self.hubs:
self.connect(quiet=True)
seen: list[str] = []
dup: set[str] = set()
for h in remote_hosts or []:
s = str(h).strip()
if s and "@" in s and s not in dup:
dup.add(s)
seen.append(s)
self._calibrate_power_down_all_ports(seen, context=context)
def _port_power_feedback(self, hub_1, port_0):
"""Return short status string after a change (hub power state + optional current)."""
h_idx = hub_1 - 1
@ -923,10 +1057,7 @@ class FiWiConcentrator:
def _write_fiber_map_document(self, doc):
path = fiber_map_path()
out = fm.ensure_fiber_map_document(doc)
with open(path, "w", encoding="utf-8") as f:
json.dump(out, f, indent=2)
f.write("\n")
fm.write_fiber_map_document(doc, path=path)
print(f"Wrote {path}", flush=True)
def _prompt_patch_panel(self, doc: dict) -> PatchPanel:
@ -989,11 +1120,23 @@ class FiWiConcentrator:
lab_in = ""
if lab_in:
label = lab_in
panel = PatchPanel(slots=n, label=label)
location = ""
if have and have.location:
location = have.location
try:
loc_in = input(
" Optional site / rack location [Enter to skip]: "
).strip()
except EOFError:
loc_in = ""
if loc_in:
location = loc_in
panel = PatchPanel(slots=n, label=label, location=location)
doc["patch_panel"] = panel.to_map_blob()
print(
f" Patch panel set: {panel.slots} position(s)"
+ (f" ({panel.label})" if panel.label else "")
+ (f" @ {panel.location}" if panel.location else "")
+ ".\n---\n",
flush=True,
)
@ -1019,7 +1162,64 @@ class FiWiConcentrator:
doc = {"fiber_ports": {}}
doc = fm.ensure_fiber_map_document(doc)
mp = fiber_map_path()
print(f"fiwi: fiber map: {mp}", file=sys.stderr, flush=True)
_map_hosts = fm.calibrate_remotes_hosts(doc)
if _map_hosts:
print(
f"fiwi: calibrate_remotes from map: {', '.join(_map_hosts)}",
file=sys.stderr,
flush=True,
)
elif doc.get("calibrate_remotes") is not None:
print(
"fiwi: calibrate_remotes in map is empty or not a list/string of user@host — "
"no SSH hub hosts from JSON.",
file=sys.stderr,
flush=True,
)
try:
if merge:
have_pp = PatchPanel.from_map_blob(doc.get("patch_panel"))
if have_pp is not None:
bits = [f"{have_pp.slots} position(s)"]
if (have_pp.label or "").strip():
bits.append(f"label {have_pp.label.strip()!r}")
if (have_pp.location or "").strip():
bits.append(f"location {have_pp.location.strip()!r}")
summ = ", ".join(bits)
print(
"\n--- Patch panel (from map) ---\n"
f" {summ}\n"
" [Enter]=keep and continue to USB hub walk · e = edit panel fields",
flush=True,
)
if sys.stdin.isatty():
try:
ans = input(
" Your choice (then Enter): "
).strip().lower()
except EOFError:
ans = ""
else:
print(
" (stdin is not a TTY — keeping patch_panel from map; "
"run in a real terminal for prompts.)\n",
flush=True,
)
ans = ""
if ans != "e":
doc["patch_panel"] = have_pp.to_map_blob()
print(
f" Keeping patch_panel ({summ}).\n---\n",
flush=True,
)
else:
self._prompt_patch_panel(doc)
else:
self._prompt_patch_panel(doc)
else:
self._prompt_patch_panel(doc)
except KeyboardInterrupt:
print(
@ -1031,34 +1231,14 @@ class FiWiConcentrator:
raise SystemExit(130)
self._write_fiber_map_document(doc)
seen_h = set()
cli_hosts = []
for h in calibrate_ssh_hosts:
s = str(h).strip()
if s and s not in seen_h:
seen_h.add(s)
cli_hosts.append(s)
cr = doc.get("calibrate_remotes")
if isinstance(cr, list):
for x in cr:
s = str(x).strip()
if s and s not in seen_h:
seen_h.add(s)
cli_hosts.append(s)
env_rem = SshNodeConfig.load().calibrate_remotes
if env_rem:
added_from_env = []
for part in env_rem.split(","):
s = part.strip()
if s and s not in seen_h:
seen_h.add(s)
cli_hosts.append(s)
added_from_env.append(s)
if added_from_env:
plan = fm.resolve_calibrate_ssh_targets(
doc, extra_cli_hosts=calibrate_ssh_hosts
)
cli_hosts = list(plan.hosts)
if plan.first_introduced_via_ssh_config:
print(
f"fiwi: calibrate remotes from FIWI_CALIBRATE_REMOTES / FIWI_REMOTE_HUBS: "
f"{', '.join(added_from_env)}",
"fiwi: calibrate remotes from FIWI_CALIBRATE_REMOTES / FIWI_REMOTE_HUBS / "
f"[remote_hubs]: {', '.join(plan.first_introduced_via_ssh_config)}",
file=sys.stderr,
flush=True,
)
@ -1164,8 +1344,19 @@ class FiWiConcentrator:
if sh is not None and sh not in remote_hosts_ordered:
remote_hosts_ordered.append(sh)
if remote_hosts_ordered:
print(
"\nRemote USB baseline over SSH (turning off each hub.port on the Pi)…",
flush=True,
)
for host in remote_hosts_ordered:
print(f" → {host}", flush=True)
print(
" Each hub.port line appears as that SSH session finishes (first connect can take up to the timeout).\n"
" Stuck with no new lines? Test `ssh <host> true` — password prompts do not work here; use keys or "
"ssh-agent. Fail fast: FIWI_SSH_OPTS='-o BatchMode=yes -o ConnectTimeout=15'.\n",
flush=True,
)
use_def = resolve_remote_defer(None)
by_host = {h: [] for h in remote_hosts_ordered}
if use_def:
meta_off: list[tuple[str, int, int]] = []
handles = []
@ -1175,35 +1366,48 @@ class FiWiConcentrator:
for h, p in pairs:
meta_off.append((host, h, p))
handles.append(node.remote_hub_port_power(h, p, False, defer=True))
res_off = [h.result() for h in handles]
for (host, h, p), r in zip(meta_off, res_off):
by_host[host].append(((h, p), r))
n_tot = len(handles)
print(
f" {n_tot} SSH off command(s) started in parallel "
f"(FIWI_REMOTE_DEFER); collecting in start order…",
flush=True,
)
for i, (handle, (host, h, p)) in enumerate(zip(handles, meta_off), start=1):
print(f" [{i}/{n_tot}] awaiting {host} hub {h} port {p}…", flush=True)
r = handle.result()
rc, rerr = r
if rc != 0:
print(
f" → exit {rc}: {(rerr or '').strip()[:120]}",
flush=True,
)
else:
print(" → ok", flush=True)
for host in remote_hosts_ordered:
time.sleep(0.6)
print(f" Remote baseline done for {host}.", flush=True)
else:
for host in remote_hosts_ordered:
pairs = sorted({(h, p) for s_h, h, p in steps if s_h == host})
for h, p in pairs:
n_off = len(pairs)
print(
f"Turning OFF every downstream USB port on {host} "
f"(baseline, {n_off} SSH call(s), sequential)…",
flush=True,
)
for i, (h, p) in enumerate(pairs, start=1):
print(f" [{i}/{n_off}] off hub {h} port {p} …", flush=True)
r = SshNode.parse(host).remote_hub_port_power(
h, p, False, defer=False
)
by_host[host].append(((h, p), r))
for host in remote_hosts_ordered:
chunk = by_host.get(host, [])
n_off = len(chunk)
mode_s = "concurrent (deferred)" if use_def else "sequential"
print(
f"Turning OFF every downstream USB port on {host} "
f"(baseline, {n_off} SSH call(s), {mode_s})…",
flush=True,
)
for i, ((hp, rp), (rc, rerr)) in enumerate(chunk, start=1):
h, p = hp
rc, rerr = r
if rc != 0:
print(
f" [{i}/{n_off}] off {h}.{p} → exit {rc}: {(rerr or '').strip()[:120]}",
f" → exit {rc}: {(rerr or '').strip()[:120]}",
flush=True,
)
else:
print(f" [{i}/{n_off}] off {h}.{p} ok", flush=True)
print(" → ok", flush=True)
time.sleep(0.6)
print(f" Remote baseline done for {host}.", flush=True)
@ -1476,3 +1680,49 @@ class FiWiConcentrator:
def disconnect(self):
for stem in self.hubs: stem.disconnect()
@dataclass
class HubDownstreamPort:
"""
One local BrainStem downstream USB port (1-based hub index, 0-based port index).
Obtain via :meth:`FiWiConcentrator.hub_plane` → :meth:`BoundHubPlane.port` / :meth:`BoundHubPlane.port_at`.
For map-aware / SSH routing, use :meth:`FiWiConcentrator.patch_panel` → :class:`~fiwi.radiohead.RadioHead`.
"""
concentrator: FiWiConcentrator
hub_1: int
port_0: int
def power_on(self) -> None:
self.concentrator.set_hub_port_power(self.hub_1, self.port_0, True)
def power_off(self) -> None:
self.concentrator.set_hub_port_power(self.hub_1, self.port_0, False)
def measure_inrush(self, *, sample_duration: float = 0.3) -> InrushSample:
return self.concentrator.sample_inrush_hub_port(
self.hub_1, self.port_0, sample_duration=sample_duration
)
@dataclass
class BoundHubPlane:
"""Bound view of this concentrator’s local hub/port space (symmetric to :class:`~fiwi.patch_panel.BoundPatchPanel`)."""
concentrator: FiWiConcentrator
def port(self, hub_1: int, port_0: int) -> HubDownstreamPort:
return HubDownstreamPort(self.concentrator, hub_1, port_0)
def port_at(self, hub_dot_port: str) -> HubDownstreamPort:
s = hub_dot_port.strip()
parts = s.split(".", 1)
if len(parts) != 2:
raise ValueError("expected hub.port, e.g. 1.0")
try:
h = int(parts[0].strip())
p = int(parts[1].strip())
except ValueError as exc:
raise ValueError("expected hub.port with integer hub and port, e.g. 1.0") from exc
return self.port(h, p)

120
fiwi/fiber_map_io.py
View File

@ -1,10 +1,15 @@
"""Load/save fiber_map.json; parse hub/port bindings and chip metadata."""
from __future__ import annotations
import json
import os
import re
import sys
import time
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from typing import Any
from fiwi.adnacom_pcie_catalog import (
ADNACOM_KNOWN_CARDS,
@ -13,7 +18,85 @@ from fiwi.adnacom_pcie_catalog import (
print_catalog_menu,
short_bdf,
)
from fiwi.paths import fiber_map_path
from fiwi.paths import base_dir, fiber_map_path
def calibrate_remotes_hosts(doc) -> list[str]:
"""
Normalize ``doc['calibrate_remotes']`` to ``user@host`` strings.
Accepts a JSON **array** of strings or a single comma-separated **string**
(panel calibrate and site setup both use this).
"""
if not isinstance(doc, dict):
return []
cr = doc.get("calibrate_remotes")
if isinstance(cr, list):
return [str(x).strip() for x in cr if str(x).strip() and "@" in str(x)]
if isinstance(cr, str) and cr.strip():
return [p.strip() for p in cr.split(",") if p.strip() and "@" in p]
return []
@dataclass(frozen=True)
class CalibrateSshTargetPlan:
"""
Resolved ``user@host`` hub targets for hybrid Fi-Wi (panel calibrate, ``off all``, relay).
Order is stable: explicit CLI hosts, then ``doc['calibrate_remotes']``, then
:attr:`~fiwi.ssh.SshNodeConfig.calibrate_remotes` (INI + ``FIWI_CALIBRATE_REMOTES`` /
``FIWI_REMOTE_HUBS``), each phase deduped.
"""
hosts: tuple[str, ...]
first_introduced_via_ssh_config: tuple[str, ...]
"""Hosts whose first appearance was only from config/env, not map or ``extra_cli_hosts``."""
def resolve_calibrate_ssh_targets(
doc: Mapping[str, Any] | None,
*,
extra_cli_hosts: Iterable[str] | None = None,
) -> CalibrateSshTargetPlan:
"""
Single place that merges calibrate SSH targets (map + profile + optional ``--ssh``).
Call after :func:`fiwi.paths.configure` / INI load so :class:`~fiwi.ssh.SshNodeConfig` resolves.
"""
from fiwi.ssh import SshNodeConfig
hosts: list[str] = []
seen: set[str] = set()
introduced_cfg: list[str] = []
for raw in extra_cli_hosts or ():
t = str(raw).strip()
if not t or "@" not in t or t in seen:
continue
seen.add(t)
hosts.append(t)
if isinstance(doc, dict):
for t in calibrate_remotes_hosts(doc):
if t in seen:
continue
seen.add(t)
hosts.append(t)
cfg_line = (SshNodeConfig.load().calibrate_remotes or "").strip()
if cfg_line:
for part in cfg_line.split(","):
t = part.strip()
if not t or "@" not in t or t in seen:
continue
seen.add(t)
hosts.append(t)
introduced_cfg.append(t)
return CalibrateSshTargetPlan(
hosts=tuple(hosts),
first_introduced_via_ssh_config=tuple(introduced_cfg),
)
def ensure_fiber_map_document(doc):
@ -29,12 +112,38 @@ def ensure_fiber_map_document(doc):
def load_fiber_map_document():
"""Load ``fiber_map.json`` if present; return None if missing."""
"""
Load the configured fiber map if present; return None if missing.
If the configured path (usually ``maps/fiber_map.json``) is absent but a legacy
``fiber_map.json`` exists in the install root, that file is loaded so older
layouts keep working until you save (which writes the configured path).
"""
fpath = fiber_map_path()
if not os.path.isfile(fpath):
return None
if os.path.isfile(fpath):
with open(fpath, encoding="utf-8") as f:
return ensure_fiber_map_document(json.load(f))
leg = os.path.join(base_dir(), "fiber_map.json")
if (
os.path.normpath(os.path.abspath(leg))
!= os.path.normpath(os.path.abspath(fpath))
and os.path.isfile(leg)
):
with open(leg, encoding="utf-8") as f:
return ensure_fiber_map_document(json.load(f))
return None
def write_fiber_map_document(doc, *, path=None):
"""Write the fiber map (default :func:`~fiwi.paths.fiber_map_path`) from a document dict."""
p = path if path is not None else fiber_map_path()
out = ensure_fiber_map_document(doc)
parent = os.path.dirname(os.path.abspath(p))
if parent:
os.makedirs(parent, exist_ok=True)
with open(p, "w", encoding="utf-8") as f:
json.dump(out, f, indent=2)
f.write("\n")
def load_fiber_map_or_exit():
@ -42,7 +151,8 @@ def load_fiber_map_or_exit():
if doc is None:
print(
f"Missing {fiber_map_path()}.\n"
" Copy fiber_map.example.json → fiber_map.json and set fiber_ports "
" Copy fiber_map.example.json → maps/fiber_map.json (or set FIWI_FIBER_MAP) "
"and set fiber_ports "
'(e.g. "5": {"hub": 1, "port": 2, "ssh": "rjmcmahon@192.168.1.39"}). '
"Use ssh / remote / host+user when USB hubs are reached via SSH.",
file=sys.stderr,

94
fiwi/fiber_radio_port.py
View File

@ -1,94 +0,0 @@
"""
Fiber + radio port: central domain object for a row in fiber_map.json.
Power (USB hub downstream ports), SSH routing, PCIe / wlan / USB metadata all hang off this
aggregate. ``FiWiConcentrator`` supplies BrainStem power; not the conceptual center.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, Iterator, List, Optional, Tuple
from fiwi import fiber_map_io as fm
from fiwi.ssh import SshNode
@dataclass
class FiberRadioPort:
"""
One logical fiber/radio attachment: ``fiber_ports[<map_key>]`` in the map document.
``entry`` is the live dict from the document (mutations persist when the doc is saved).
"""
map_key: str
entry: Optional[Dict[str, Any]]
@property
def port_id(self) -> Optional[int]:
"""Integer fiber id when ``map_key`` is decimal; else None."""
if self.map_key.isdigit():
return int(self.map_key)
return None
def hub_port(self) -> Optional[Tuple[int, int]]:
"""(hub_1based, port_0based) or None if unmapped / invalid."""
return fm.fiber_entry_hub_port(self.entry)
def ssh_target(self) -> Optional[str]:
"""user@host when this port’s hubs are reached via SSH; else None."""
return fm.fiber_ssh_target(self.entry) if isinstance(self.entry, dict) else None
def ssh_node(self) -> Optional[SshNode]:
"""Remote Fi-Wi host (``SshNode``) for this port, or None when mapped locally."""
t = self.ssh_target()
if not t:
return None
try:
return SshNode.parse(t)
except ValueError:
return None
def is_mapped(self) -> bool:
"""True when hub.port is valid in the entry."""
return self.hub_port() is not None
def chip_preview(self, width: int = 26) -> str:
return fm.stored_chip_preview(self.entry) if isinstance(self.entry, dict) else ""
def pcie_preview(self, width: int = 22) -> str:
return fm.stored_pcie_preview(self.entry) if isinstance(self.entry, dict) else ""
@classmethod
def from_map_key(cls, doc: Dict[str, Any], map_key: str) -> FiberRadioPort:
ports = doc.get("fiber_ports") if isinstance(doc, dict) else None
if not isinstance(ports, dict):
return cls(str(map_key), None)
ent = ports.get(str(map_key))
return cls(
str(map_key),
ent if isinstance(ent, dict) else None,
)
@classmethod
def from_port_id(cls, doc: Dict[str, Any], port_id: int) -> FiberRadioPort:
return cls.from_map_key(doc, str(int(port_id)))
@staticmethod
def each_from_document(doc: Dict[str, Any]) -> Iterator[FiberRadioPort]:
"""All ``fiber_ports`` rows (sorted), including unmapped / empty values."""
ports = doc.get("fiber_ports") if isinstance(doc, dict) else None
if not isinstance(ports, dict):
return
for key in sorted(ports.keys(), key=fm.fiber_sort_key):
ent = ports[key]
yield FiberRadioPort(
str(key),
ent if isinstance(ent, dict) else None,
)
def load_fiber_radio_ports(doc: Dict[str, Any]) -> List[FiberRadioPort]:
"""Registry of all fiber map rows (sorted keys)."""
return list(FiberRadioPort.each_from_document(doc))

3
fiwi/fiwi_relay/__init__.py
View File

@ -6,10 +6,13 @@ from .host import (
probe_remote_hub_readiness,
suggested_remote_venv_python,
)
from .relay_app import FiWiRelayBootstrapApp, main as run_fiwi_relay
__all__ = [
"FiWiRelay",
"FiWiRelayBootstrapApp",
"RemoteReadiness",
"probe_remote_hub_readiness",
"run_fiwi_relay",
"suggested_remote_venv_python",
]

125
fiwi/fiwi_relay/__main__.py
View File

@ -1,129 +1,8 @@
"""
Run as ``python -m fiwi.fiwi_relay`` — builds :class:`~.host.FiWiRelay` from config and calls :meth:`~.host.FiWiRelay.setup`.
"""
"""``python -m fiwi.fiwi_relay`` — thin entry; see :mod:`fiwi.fiwi_relay.relay_app`."""
from __future__ import annotations
import argparse
import os
import sys
_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".."))
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
def main() -> int:
p = argparse.ArgumentParser(
description="Fi-Wi hub relay: FiWiRelay.setup() over SSH or locally (auto when user@host is this machine).",
)
p.add_argument(
"-c",
"--config",
metavar="PROFILE_OR_INI",
help="INI profile or path (sets FIWI_CONFIG before loading config)",
)
p.add_argument(
"--ssh",
metavar="USER@HOST",
help="Override SSH target (default: first host from merged [remote_hubs] / env); ignored with --local",
)
p.add_argument(
"-n",
"--name",
metavar="LABEL",
help="Override hub label (default: [remote_hubs] label or derived from host)",
)
p.add_argument(
"--all",
action="store_true",
help="Run setup() on every configured remote host (local vs ssh auto per host)",
)
p.add_argument(
"--timeout",
type=float,
default=600.0,
metavar="SEC",
help="SSH session timeout (default: 600)",
)
p.add_argument(
"--dry-run",
action="store_true",
help="Print what would run; do not SSH / do not run bash",
)
p.add_argument(
"--local",
action="store_true",
help="Force local execution (this machine); also used if from_config fails and no [remote_hubs]",
)
p.add_argument(
"--remote",
action="store_true",
help="Force SSH even if this machine matches the configured user@host",
)
args = p.parse_args()
if args.config:
os.environ["FIWI_CONFIG"] = args.config.strip()
if args.local and args.remote:
print("FAIL: use only one of --local and --remote.", file=sys.stderr)
return 2
try:
os.chdir(_ROOT)
except OSError as exc:
print(f"FAIL: cannot chdir to repo root {_ROOT!r}: {exc}", file=sys.stderr)
return 1
from fiwi.fiwi_relay.host import FiWiRelay
try:
if args.local and args.all:
print("FAIL: --local does not support --all (would run local bootstrap for every listed host).", file=sys.stderr)
return 2
if args.all:
hosts = FiWiRelay.each_from_config(_ROOT)
if not hosts:
print(
"No remote hosts in config. Set [remote_hubs] hosts or FIWI_REMOTE_HUBS.",
file=sys.stderr,
)
return 2
rc = 0
for h in hosts:
use_local = not args.remote and FiWiRelay.ssh_target_points_here(h.ssh_target)
mode = "local" if use_local else "ssh"
print(f"\n--- {h.name} ({h.ssh_target}) [{mode}] ---\n", flush=True)
code = h.setup(timeout=args.timeout, dry_run=args.dry_run, local=use_local)
if code != 0:
rc = code
return rc
if args.local:
try:
host = FiWiRelay.from_config(_ROOT, ssh_target=args.ssh, name=args.name)
except ValueError:
host = FiWiRelay.for_local_machine(_ROOT, name=args.name)
print(
f"FiWiRelay(name={host.name!r}, ssh_target={host.ssh_target!r}, local=True) [forced --local]",
flush=True,
)
return host.setup(timeout=args.timeout, dry_run=args.dry_run, local=True)
host = FiWiRelay.from_config(_ROOT, ssh_target=args.ssh, name=args.name)
use_local = not args.remote and FiWiRelay.ssh_target_points_here(host.ssh_target)
if use_local:
tag = " [auto local]"
elif args.remote:
tag = " [forced ssh]"
else:
tag = " [ssh]"
print(f"FiWiRelay(name={host.name!r}, ssh_target={host.ssh_target!r}){tag}", flush=True)
return host.setup(timeout=args.timeout, dry_run=args.dry_run, local=use_local)
except ValueError as exc:
print(f"FAIL: {exc}", file=sys.stderr)
return 2
from fiwi.fiwi_relay.relay_app import main
if __name__ == "__main__":
raise SystemExit(main())

174
fiwi/fiwi_relay/relay_app.py Normal file
View File

@ -0,0 +1,174 @@
"""
Hub relay bootstrap as an object: parse argv, chdir to install root, run :class:`~.host.FiWiRelay` setup.
Used by ``python -m fiwi.fiwi_relay`` and by :mod:`fiwi.site_setup` (embed with explicit ``argv``).
"""
from __future__ import annotations
import argparse
import os
import sys
from dataclasses import dataclass
from fiwi.fiwi_relay.host import FiWiRelay
def default_install_root() -> str:
"""Directory containing ``fiwi.py`` (repo / install root)."""
return os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".."))
def build_relay_argument_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
description=(
"Fi-Wi hub relay: FiWiRelay.setup() over SSH or locally "
"(auto when user@host is this machine)."
),
)
p.add_argument(
"-c",
"--config",
metavar="PROFILE_OR_INI",
help="INI profile or path (sets FIWI_CONFIG before loading config)",
)
p.add_argument(
"--ssh",
metavar="USER@HOST",
help=(
"Override SSH target (default: first host from merged [remote_hubs] / env); "
"ignored with --local"
),
)
p.add_argument(
"-n",
"--name",
metavar="LABEL",
help="Override hub label (default: [remote_hubs] label or derived from host)",
)
p.add_argument(
"--all",
action="store_true",
help="Run setup() on every configured remote host (local vs ssh auto per host)",
)
p.add_argument(
"--timeout",
type=float,
default=600.0,
metavar="SEC",
help="SSH session timeout (default: 600)",
)
p.add_argument(
"--dry-run",
action="store_true",
help="Print what would run; do not SSH / do not run bash",
)
p.add_argument(
"--local",
action="store_true",
help=(
"Force local execution (this machine); also used if from_config fails "
"and no [remote_hubs]"
),
)
p.add_argument(
"--remote",
action="store_true",
help="Force SSH even if this machine matches the configured user@host",
)
return p
@dataclass
class FiWiRelayBootstrapApp:
"""
One relay bootstrap run: ``argv`` is either ``None`` (use ``sys.argv[1:]``) or an explicit
token list so a parent CLI is not misparsed.
"""
argv: list[str] | None = None
install_root: str | None = None
def _parse_args(self) -> argparse.Namespace:
p = build_relay_argument_parser()
if self.argv is None:
return p.parse_args()
return p.parse_args(self.argv)
def run(self) -> int:
args = self._parse_args()
if args.config:
os.environ["FIWI_CONFIG"] = args.config.strip()
if args.local and args.remote:
print("FAIL: use only one of --local and --remote.", file=sys.stderr)
return 2
root = self.install_root if self.install_root is not None else default_install_root()
if root not in sys.path:
sys.path.insert(0, root)
try:
os.chdir(root)
except OSError as exc:
print(f"FAIL: cannot chdir to repo root {root!r}: {exc}", file=sys.stderr)
return 1
try:
if args.local and args.all:
print(
"FAIL: --local does not support --all "
"(would run local bootstrap for every listed host).",
file=sys.stderr,
)
return 2
if args.all:
hosts = FiWiRelay.each_from_config(root)
if not hosts:
print(
"No remote hosts in config. Set [remote_hubs] hosts or FIWI_REMOTE_HUBS.",
file=sys.stderr,
)
return 2
rc = 0
for h in hosts:
use_local = not args.remote and FiWiRelay.ssh_target_points_here(h.ssh_target)
mode = "local" if use_local else "ssh"
print(f"\n--- {h.name} ({h.ssh_target}) [{mode}] ---\n", flush=True)
code = h.setup(timeout=args.timeout, dry_run=args.dry_run, local=use_local)
if code != 0:
rc = code
return rc
if args.local:
try:
host = FiWiRelay.from_config(root, ssh_target=args.ssh, name=args.name)
except ValueError:
host = FiWiRelay.for_local_machine(root, name=args.name)
print(
f"FiWiRelay(name={host.name!r}, ssh_target={host.ssh_target!r}, local=True) "
f"[forced --local]",
flush=True,
)
return host.setup(timeout=args.timeout, dry_run=args.dry_run, local=True)
host = FiWiRelay.from_config(root, ssh_target=args.ssh, name=args.name)
use_local = not args.remote and FiWiRelay.ssh_target_points_here(host.ssh_target)
if use_local:
tag = " [auto local]"
elif args.remote:
tag = " [forced ssh]"
else:
tag = " [ssh]"
print(
f"FiWiRelay(name={host.name!r}, ssh_target={host.ssh_target!r}){tag}",
flush=True,
)
return host.setup(timeout=args.timeout, dry_run=args.dry_run, local=use_local)
except ValueError as exc:
print(f"FAIL: {exc}", file=sys.stderr)
return 2
def main(argv: list[str] | None = None, *, install_root: str | None = None) -> int:
"""CLI / embed entry: same as ``FiWiRelayBootstrapApp(...).run()``."""
return FiWiRelayBootstrapApp(argv=argv, install_root=install_root).run()

52
fiwi/patch_panel.py
View File

@ -1,7 +1,7 @@
"""
Physical patch panel: front-panel position count for the rack (field workflow).
Stored in fiber_map.json as ``patch_panel``: ``{ "slots": N, "label": "…" }``.
Stored in fiber_map.json as ``patch_panel``: ``{ "slots": N, "label": "…", "location": "…" }``.
USB hub calibrate still walks hub ports; map keys 1…N align with panel positions.
"""
@ -9,10 +9,15 @@ from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any, Dict, Optional
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from fiwi.constants import PANEL_SLOTS
if TYPE_CHECKING:
from fiwi.concentrator import FiWiConcentrator
from fiwi.radiohead import RadioHead, RadioHeadEntry
_MAX_SLOTS = 256
@ -35,6 +40,7 @@ class PatchPanel:
slots: int
label: str = ""
location: str = ""
def __post_init__(self) -> None:
if self.slots < 1:
@ -46,6 +52,8 @@ class PatchPanel:
d: Dict[str, Any] = {"slots": self.slots}
if self.label.strip():
d["label"] = self.label.strip()
if self.location.strip():
d["location"] = self.location.strip()
return d
@classmethod
@ -61,7 +69,45 @@ class PatchPanel:
return None
lab = blob.get("label")
label = lab.strip() if isinstance(lab, str) else ""
return cls(slots=s, label=label)
loc = blob.get("location")
location = loc.strip() if isinstance(loc, str) else ""
return cls(slots=s, label=label, location=location)
def bound(self, concentrator: FiWiConcentrator, doc: Dict[str, Any]) -> BoundPatchPanel:
"""Attach a :class:`FiWiConcentrator` and loaded fiber map for :class:`~fiwi.radiohead.RadioHead` access."""
return BoundPatchPanel(panel=self, concentrator=concentrator, doc=doc)
@dataclass
class BoundPatchPanel:
"""
Patch panel (slot count) plus fiber map document and concentrator — the supported way to get a
:class:`~fiwi.radiohead.RadioHead` for a front-panel position.
"""
panel: PatchPanel
concentrator: FiWiConcentrator
doc: Dict[str, Any]
def head(self, position_1based: int) -> Optional[RadioHead]:
"""Radio head for panel position ``1…panel.slots``, or ``None`` if unmapped."""
if position_1based < 1 or position_1based > self.panel.slots:
raise ValueError(
f"panel position must be 1–{self.panel.slots}, got {position_1based}"
)
map_ent = RadioHeadEntry.from_port_id(self.doc, position_1based)
if not map_ent.is_mapped():
return None
return RadioHead(map_ent, self.concentrator, patch_panel_port=position_1based)
def heads(self) -> List[RadioHead]:
"""Mapped positions only, in panel order (1…slots)."""
out: List[RadioHead] = []
for n in range(1, self.panel.slots + 1):
rh = self.head(n)
if rh is not None:
out.append(rh)
return out
def effective_panel_slots(doc: Optional[Dict[str, Any]]) -> int:

9
fiwi/paths.py
View File

@ -35,9 +35,14 @@ def config_dir() -> str:
def fiber_map_path() -> str:
name = (os.environ.get("FIWI_FIBER_MAP") or "fiber_map.json").strip()
"""
Active map file path. Default: ``maps/fiber_map.json`` under :func:`base_dir`
(override with ``FIWI_FIBER_MAP`` or ``[paths] fiber_map`` in config).
Timestamped backups from site setup live beside this file in ``maps/``.
"""
name = (os.environ.get("FIWI_FIBER_MAP") or "maps/fiber_map.json").strip()
if not name:
name = "fiber_map.json"
name = "maps/fiber_map.json"
if os.path.isabs(name):
return name
return os.path.join(base_dir(), name)

320
fiwi/radiohead.py Normal file
View File

@ -0,0 +1,320 @@
"""
Radio head control and map entries.
:class:`RadioHeadEntry` is one ``fiber_ports[]`` row (map key + JSON object). :class:`RadioHead`
pairs that with a :class:`FiWiConcentrator` for power and inrush — obtain via
:meth:`FiWiConcentrator.patch_panel` → :meth:`~fiwi.patch_panel.BoundPatchPanel.head` / ``heads``.
"""
from __future__ import annotations
import json
import time
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Tuple
from fiwi import fiber_map_io as fm
from fiwi.ssh import SshNode
if TYPE_CHECKING:
from fiwi.concentrator import FiWiConcentrator
@dataclass(frozen=True)
class InrushSample:
"""Peak downstream current (mA) and approximate settle time (ms) over a short capture window."""
peak_ma: float
duration_ms: float
def as_json_dict(self) -> dict[str, float]:
return {"peak_ma": self.peak_ma, "duration_ms": self.duration_ms}
@classmethod
def from_json_text(cls, text: str) -> InrushSample:
d = json.loads(text.strip())
if not isinstance(d, dict):
raise ValueError("inrush JSON must be an object")
return cls(float(d["peak_ma"]), float(d["duration_ms"]))
@dataclass
class RadioHeadEntry:
"""
One ``fiber_ports[<map_key>]`` record from ``fiber_map.json`` (no BrainStem / SSH capability).
``data`` is the live dict from the document (mutations persist when the doc is saved).
"""
map_key: str
data: Optional[Dict[str, Any]]
@property
def port_id(self) -> Optional[int]:
"""Integer fiber id when ``map_key`` is decimal; else None."""
if self.map_key.isdigit():
return int(self.map_key)
return None
def hub_port(self) -> Optional[Tuple[int, int]]:
"""(hub_1based, port_0based) or None if unmapped / invalid."""
return fm.fiber_entry_hub_port(self.data)
def ssh_target(self) -> Optional[str]:
"""user@host when this head’s hubs are reached via SSH; else None."""
return fm.fiber_ssh_target(self.data) if isinstance(self.data, dict) else None
def ssh_node(self) -> Optional[SshNode]:
"""Remote Fi-Wi host (``SshNode``) for this head, or None when mapped locally."""
t = self.ssh_target()
if not t:
return None
try:
return SshNode.parse(t)
except ValueError:
return None
def is_mapped(self) -> bool:
"""True when hub.port is valid in the entry."""
return self.hub_port() is not None
def chip_preview(self, width: int = 26) -> str:
return fm.stored_chip_preview(self.data) if isinstance(self.data, dict) else ""
def pcie_preview(self, width: int = 22) -> str:
return fm.stored_pcie_preview(self.data) if isinstance(self.data, dict) else ""
@property
def chip_type(self) -> Optional[str]:
"""Saved radio / USB chip label from the map (``chip_type``, wlan, or ``usb_id``), if any."""
d = self.data
if not isinstance(d, dict):
return None
for k in ("chip_type", "chip_label", "usb_id"):
v = d.get(k)
if isinstance(v, str) and v.strip():
return v.strip()
wlan = d.get("wlan")
if isinstance(wlan, dict):
prim = wlan.get("primary")
if isinstance(prim, dict):
for k in ("chip_label", "product"):
v = prim.get(k)
if isinstance(v, str) and v.strip():
return v.strip()
v = d.get("usb_description")
if isinstance(v, str) and v.strip():
return v.strip()
return None
@property
def radio_name(self) -> Optional[str]:
"""Optional human label for this strand / RRH (``fiber_ports[].name``)."""
d = self.data
if not isinstance(d, dict):
return None
v = d.get("name")
return v.strip() if isinstance(v, str) and v.strip() else None
@property
def radio_location(self) -> Optional[str]:
"""Optional physical strand / antenna location (``fiber_ports[].location``)."""
d = self.data
if not isinstance(d, dict):
return None
v = d.get("location")
return v.strip() if isinstance(v, str) and v.strip() else None
@classmethod
def from_map_key(cls, doc: Dict[str, Any], map_key: str) -> RadioHeadEntry:
ports = doc.get("fiber_ports") if isinstance(doc, dict) else None
if not isinstance(ports, dict):
return cls(str(map_key), None)
ent = ports.get(str(map_key))
return cls(
str(map_key),
ent if isinstance(ent, dict) else None,
)
@classmethod
def from_port_id(cls, doc: Dict[str, Any], port_id: int) -> RadioHeadEntry:
return cls.from_map_key(doc, str(int(port_id)))
@staticmethod
def each_from_document(doc: Dict[str, Any]) -> Iterator[RadioHeadEntry]:
"""All ``fiber_ports`` rows (sorted), including unmapped / empty values."""
ports = doc.get("fiber_ports") if isinstance(doc, dict) else None
if not isinstance(ports, dict):
return
for key in sorted(ports.keys(), key=fm.fiber_sort_key):
ent = ports[key]
yield RadioHeadEntry(
str(key),
ent if isinstance(ent, dict) else None,
)
def load_radio_head_entries(doc: Dict[str, Any]) -> List[RadioHeadEntry]:
"""All ``fiber_ports`` rows (sorted keys)."""
return list(RadioHeadEntry.each_from_document(doc))
@dataclass
class RadioHead:
"""
Map entry plus concentrator: USB hub power, inrush sampling, and SSH routing from the map row.
Live ``power`` / ``voltage`` / ``current`` read BrainStem (or ``port-metrics-json`` over SSH).
Values are cached until :meth:`refresh_live`, or after :meth:`power_on` / :meth:`power_off` /
:meth:`power_cycle` / :meth:`measure_inrush`.
``chip_type`` and ``patch_panel_port`` come from the map / panel binding when known.
"""
map_entry: RadioHeadEntry
concentrator: FiWiConcentrator
patch_panel_port: Optional[int] = None
"""1-based front-panel position when this head was resolved via :meth:`~fiwi.patch_panel.BoundPatchPanel.head`."""
_electrical_cache: Optional[Tuple[Optional[bool], Optional[float], Optional[float]]] = field(
default=None, init=False, repr=False
)
def _invalidate_electrical_cache(self) -> None:
self._electrical_cache = None
def refresh_live(self) -> None:
"""Clear cached ``power`` / ``voltage`` / ``current`` so the next read hits the hub again."""
self._invalidate_electrical_cache()
def _hub_port(self) -> Tuple[int, int]:
tup = self.map_entry.hub_port()
if tup is None:
raise ValueError("radio head is not mapped to hub.port")
return tup
@property
def chip_type(self) -> Optional[str]:
"""Chip / NIC identity from ``fiber_map.json`` (not a live probe)."""
return self.map_entry.chip_type
@property
def power(self) -> Optional[bool]:
"""Downstream port power: ``True`` ON, ``False`` OFF, ``None`` if unknown (no hub / parse error)."""
p, _v, _c = self._read_live_electrical()
return p
@property
def voltage(self) -> Optional[float]:
"""Port voltage in volts, if the hub reports it."""
_p, v, _c = self._read_live_electrical()
return v
@property
def current(self) -> Optional[float]:
"""Port current in milliamps (BrainStem µA-based), if the hub reports it."""
_p, _v, c = self._read_live_electrical()
return c
def _read_live_electrical(self) -> Tuple[Optional[bool], Optional[float], Optional[float]]:
if self._electrical_cache is not None:
return self._electrical_cache
if not self.map_entry.is_mapped():
self._electrical_cache = (None, None, None)
return self._electrical_cache
hub_1, port_0 = self._hub_port()
if self.map_entry.ssh_target():
node = self.map_entry.ssh_node()
if node is None:
return None, None, None
code, out, err = node.invoke_capture(["port-metrics-json"], defer=False, timeout=90)
if code != 0 or not (out or "").strip():
self._electrical_cache = (None, None, None)
return self._electrical_cache
try:
rows = json.loads(out.strip())
except json.JSONDecodeError:
self._electrical_cache = (None, None, None)
return self._electrical_cache
if not isinstance(rows, list):
self._electrical_cache = (None, None, None)
return self._electrical_cache
for r in rows:
if not isinstance(r, dict):
continue
if r.get("hub") == hub_1 and r.get("port") == port_0:
pwr = r.get("power")
on = True if pwr == "ON" else False if pwr == "OFF" else None
v_raw = r.get("voltage_v")
v = float(v_raw) if isinstance(v_raw, (int, float)) else None
c_raw = r.get("current_ma")
c = float(c_raw) if isinstance(c_raw, (int, float)) else None
self._electrical_cache = (on, v, c)
return self._electrical_cache
self._electrical_cache = (None, None, None)
return self._electrical_cache
row = self.concentrator.hub_port_metrics(hub_1, port_0)
if row is None:
self._electrical_cache = (None, None, None)
return self._electrical_cache
pwr = row.get("power")
on = True if pwr == "ON" else False if pwr == "OFF" else None
v_raw = row.get("voltage_v")
v = float(v_raw) if isinstance(v_raw, (int, float)) else None
c_raw = row.get("current_ma")
c = float(c_raw) if isinstance(c_raw, (int, float)) else None
self._electrical_cache = (on, v, c)
return self._electrical_cache
def power_on(self) -> None:
hub_1, port_0 = self._hub_port()
if self.map_entry.ssh_target():
node = self.map_entry.ssh_node()
if node is None:
raise ValueError("invalid ssh target in fiber map")
code, msg = node.remote_hub_port_power(hub_1, port_0, True, defer=False)
if code != 0:
raise RuntimeError(msg.strip() if msg else f"remote power on failed (exit {code})")
else:
self.concentrator.set_hub_port_power(hub_1, port_0, True)
self._invalidate_electrical_cache()
def power_off(self) -> None:
hub_1, port_0 = self._hub_port()
if self.map_entry.ssh_target():
node = self.map_entry.ssh_node()
if node is None:
raise ValueError("invalid ssh target in fiber map")
code, msg = node.remote_hub_port_power(hub_1, port_0, False, defer=False)
if code != 0:
raise RuntimeError(msg.strip() if msg else f"remote power off failed (exit {code})")
else:
self.concentrator.set_hub_port_power(hub_1, port_0, False)
self._invalidate_electrical_cache()
def power_cycle(self, off_s: float = 0.25) -> None:
self.power_off()
time.sleep(off_s)
self.power_on()
def measure_inrush(self, *, sample_duration: float = 0.3) -> InrushSample:
hub_1, port_0 = self._hub_port()
if self.map_entry.ssh_target():
node = self.map_entry.ssh_node()
if node is None:
raise ValueError("invalid ssh target in fiber map")
code, out, err = node.invoke_capture(
["hub-inrush-json", f"{hub_1}.{port_0}"],
defer=False,
timeout=90,
)
blob = (out or "").strip()
if code != 0:
raise RuntimeError((err or blob or f"remote hub-inrush-json exit {code}").strip())
self._invalidate_electrical_cache()
return InrushSample.from_json_text(blob)
sample = self.concentrator.sample_inrush_hub_port(
hub_1, port_0, sample_duration=sample_duration
)
self._invalidate_electrical_cache()
return sample

620
fiwi/site_setup.py Normal file
View File

@ -0,0 +1,620 @@
"""
**Primary operator entry** for Fi-Wi on a rig: site metadata, fiber map layout, optional
USB power baseline, optional hub relay bootstrap, optional panel calibrate — usually as::
site_setup.py -c uax24
Optional ``--relay`` / ``--calibrate`` only **trigger** those steps; relay **hosts** come from
``config/*.ini`` (``[remote_hubs] hosts``) and env. ``--relay`` runs
:class:`~fiwi.fiwi_relay.relay_app.FiWiRelayBootstrapApp` in-process (same as ``python -m fiwi.fiwi_relay``).
``--calibrate`` runs :func:`fiwi.cli.run_panel_calibrate` in-process.
``--ssh`` is only for **extra** calibrate targets beyond ``calibrate_remotes`` in the map / profile.
Writes the fiber map (default ``maps/fiber_map.json`` under the install root) keys:
* ``fiwi_site`` — ``concentrator_name``, ``concentrator_location``
* ``patch_panel`` — ``slots``, ``label`` (panel name), ``location``
* ``calibrate_remotes`` — ``user@host`` list for hybrid panel calibrate
Interactive setup can optionally **power down** local downstream USB ports and run ``fiwi off all``
on configured SSH hub hosts **after** saving the map, only when you are **not** continuing to panel
calibrate (``panel calibrate`` repeats that baseline itself).
``calibrate_remotes`` defaults from the active INI profile (e.g. ``[remote_hubs] hosts`` in
``config/uax24.ini``) when you use ``site_setup.py -c uax24``.
Per-strand labels use optional ``fiber_ports[<id>].name`` and ``.location`` (see
:class:`~fiwi.radiohead.RadioHeadEntry`).
"""
from __future__ import annotations
import argparse
import os
import shutil
import socket
import sys
from datetime import datetime
from typing import TYPE_CHECKING, Any, Dict, List, Tuple
from fiwi import fiber_map_io as fm
from fiwi.paths import fiber_map_path
from fiwi.patch_panel import BoundPatchPanel, PatchPanel, default_panel_ports
if TYPE_CHECKING:
from fiwi.concentrator import FiWiConcentrator
from fiwi.radiohead import RadioHead
FIWI_SITE_KEY = "fiwi_site"
def _fiwi_repo_root() -> str:
"""Directory containing ``fiwi.py`` (parent of package ``fiwi``)."""
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def merge_calibrate_remotes(doc: Dict[str, Any], hosts: List[str]) -> Dict[str, Any]:
"""Set ``calibrate_remotes`` to deduped ``user@host`` list, or remove key if empty."""
doc = fm.ensure_fiber_map_document(doc)
seen: List[str] = []
for h in hosts:
s = (h or "").strip()
if s and "@" in s and s not in seen:
seen.append(s)
if seen:
doc["calibrate_remotes"] = seen
elif "calibrate_remotes" in doc:
del doc["calibrate_remotes"]
return doc
def read_fiwi_site(doc: Dict[str, Any]) -> Dict[str, str]:
"""Return ``fiwi_site`` object from a map document, or empty dict."""
raw = doc.get(FIWI_SITE_KEY)
if not isinstance(raw, dict):
return {}
out: Dict[str, str] = {}
for k in ("concentrator_name", "concentrator_location"):
v = raw.get(k)
if isinstance(v, str) and v.strip():
out[k] = v.strip()
return out
def merge_fiwi_site(
doc: Dict[str, Any],
*,
concentrator_name: str,
concentrator_location: str,
) -> Dict[str, Any]:
"""Attach ``fiwi_site``; empty strings omit that field (existing key removed if both empty)."""
doc = fm.ensure_fiber_map_document(doc)
site: Dict[str, str] = {}
if concentrator_name.strip():
site["concentrator_name"] = concentrator_name.strip()
if concentrator_location.strip():
site["concentrator_location"] = concentrator_location.strip()
if site:
doc[FIWI_SITE_KEY] = site
elif FIWI_SITE_KEY in doc:
del doc[FIWI_SITE_KEY]
return doc
def merge_patch_panel_meta(
doc: Dict[str, Any],
*,
slots: int,
label: str,
location: str,
) -> Dict[str, Any]:
"""Merge ``patch_panel`` slots / name / location; preserves unrelated keys in the blob."""
doc = fm.ensure_fiber_map_document(doc)
if slots < 1 or slots > 256:
raise ValueError("patch panel slots must be 1–256")
pp = PatchPanel(slots=slots, label=label.strip(), location=location.strip())
blob: Dict[str, Any] = {}
prev = doc.get("patch_panel")
if isinstance(prev, dict):
blob.update(prev)
blob.update(pp.to_map_blob())
doc["patch_panel"] = blob
return doc
def open_fiwi_stack() -> Tuple["FiWiConcentrator", BoundPatchPanel, List["RadioHead"]]:
"""
Load ``fiber_map.json``, open a :class:`~fiwi.concentrator.FiWiConcentrator`, bound patch panel,
and all mapped :class:`~fiwi.radiohead.RadioHead` instances (panel order, then off-panel fiber ids).
Raises:
FileNotFoundError: if ``fiber_map.json`` is missing.
"""
from fiwi.concentrator import FiWiConcentrator
from fiwi.radiohead import RadioHead, RadioHeadEntry
doc = fm.load_fiber_map_document()
if doc is None:
raise FileNotFoundError(
f"Missing {fiber_map_path()}. Run site_setup.py or copy fiber_map.example.json."
)
c = FiWiConcentrator()
bound = c.patch_panel(doc)
heads: List[RadioHead] = list(bound.heads())
seen = {h.map_entry.map_key for h in heads}
for ent in RadioHeadEntry.each_from_document(doc):
if ent.is_mapped() and ent.map_key not in seen:
heads.append(RadioHead(ent, c))
seen.add(ent.map_key)
def _sort_key(h: RadioHead) -> tuple:
p = h.patch_panel_port
if p is not None:
return (0, p, h.map_entry.map_key)
mk = h.map_entry.map_key
if mk.isdigit():
return (1, int(mk), mk)
return (2, mk, mk)
heads.sort(key=_sort_key)
return c, bound, heads
def _prompt(default: str, label: str) -> str:
try:
raw = input(f" {label} [{default}]: ").strip()
except EOFError:
raw = ""
return raw if raw else default
def _prompt_usb_power_down(*, remotes: List[str], has_local_hubs: bool) -> bool:
bits: List[str] = []
if has_local_hubs:
bits.append("every downstream port on connected local USB hubs")
if remotes:
bits.append("`fiwi off all` on " + ", ".join(remotes))
print(
"\n--- USB power baseline ---\n\n"
"Turn OFF " + " and ".join(bits) + "\n"
" (You are not running panel calibrate this session; calibrate would do this baseline for you.)\n",
flush=True,
)
try:
raw = input(" Power down now? [Y/n]: ").strip().lower()
except EOFError:
return True
return raw not in ("n", "no")
def _run_early_usb_power_down_if_requested(doc: Dict[str, Any], args: argparse.Namespace) -> None:
if args.no_power_down:
return
from fiwi.concentrator import FiWiConcentrator
remotes = list(fm.resolve_calibrate_ssh_targets(doc, extra_cli_hosts=()).hosts)
c = FiWiConcentrator()
try:
if not c.hubs:
c.connect(quiet=True)
has_local = bool(c.hubs)
if not has_local and not remotes:
print(
"\n(USB power baseline skipped: no local hubs opened and no SSH hub hosts in profile/map.)\n",
flush=True,
)
return
if not _prompt_usb_power_down(remotes=remotes, has_local_hubs=has_local):
print(" Skipping USB power-down.\n", flush=True)
return
c.power_down_all_downstream_ports(remotes, context="Site setup")
print(" USB baseline done.\n", flush=True)
finally:
c.disconnect()
def _timestamped_fiber_map_backup_path(original: str) -> str:
"""``fiber_map.json`` → ``fiber_map.json.20260403T143022`` (suffix if collision)."""
d = os.path.dirname(os.path.abspath(original))
base = os.path.basename(original)
stamp = datetime.now().strftime("%Y%m%dT%H%M%S")
candidate = os.path.join(d, f"{base}.{stamp}")
if not os.path.exists(candidate):
return candidate
for i in range(2, 10_000):
alt = os.path.join(d, f"{base}.{stamp}.{i}")
if not os.path.exists(alt):
return alt
raise OSError("could not allocate a unique backup filename")
def _prompt_existing_map_action(path: str) -> str:
"""
Ask what to do when ``fiber_map.json`` already exists.
Returns:
``\"merge\"`` — load and update metadata only.
``\"delete\"`` — move file to a timestamped backup and start ``{\"fiber_ports\": {}}``.
``\"exit\"`` — caller should exit without writing.
"""
print(
f"\n{path} already exists.\n\n"
" [m] Merge — keep fiber_ports and everything else; update only\n"
" fiwi_site, patch_panel, and calibrate_remotes\n"
" [d] New map — move this file to a timestamped backup, then start fresh\n"
" [q] Quit — exit without changing anything\n",
flush=True,
)
while True:
try:
raw = input(" Continue? [m/d/q] (default: m): ").strip().lower()
except EOFError:
return "exit"
if not raw or raw in ("m", "merge"):
return "merge"
if raw in ("d", "delete", "fresh"):
return "delete"
if raw in ("q", "quit", "exit", "x"):
return "exit"
print(" Enter m, d, or q.", flush=True)
def _run_interactive(args: argparse.Namespace) -> Dict[str, Any]:
from fiwi.config import resolved_config_path
from fiwi import paths as paths_mod
from fiwi.ssh import SshNodeConfig
ini_path = resolved_config_path(paths_mod.base_dir())
prof = (os.environ.get("FIWI_CONFIG") or "").strip()
if ini_path:
print(
f"Active profile: {ini_path} (FIWI_CONFIG={prof or 'default'})\n",
flush=True,
)
elif prof:
print(
f"FIWI_CONFIG={prof!r} — no matching config/*.ini; using env / code defaults only.\n",
flush=True,
)
path = fiber_map_path()
exists = os.path.isfile(path)
if exists:
action = _prompt_existing_map_action(path)
if action == "exit":
raise SystemExit(0)
if action == "delete":
backup = _timestamped_fiber_map_backup_path(path)
try:
shutil.move(path, backup)
except OSError as exc:
print(f"Could not move {path} → {backup}: {exc}", file=sys.stderr, flush=True)
raise SystemExit(1) from exc
exists = False
print(f"Moved {path} → {backup}\nStarting a new map.\n", flush=True)
doc = fm.load_fiber_map_document() if exists else {"fiber_ports": {}}
doc = fm.ensure_fiber_map_document(doc)
site = read_fiwi_site(doc)
pp = PatchPanel.from_map_blob(doc.get("patch_panel"))
def_cn = site.get("concentrator_name", args.concentrator_name or "") or socket.gethostname()
def_cl = site.get("concentrator_location", args.concentrator_location or "")
def_slots = pp.slots if pp is not None else default_panel_ports()
def_plab = (pp.label if pp else "") or args.panel_name or ""
def_ploc = (pp.location if pp else "") or args.panel_location or ""
print("\n--- Fi-Wi site (concentrator / this machine) ---\n", flush=True)
cn = _prompt(def_cn, "Concentrator name (e.g. rack or hostname label)")
cl = _prompt(def_cl, "Concentrator location (e.g. lab, datacenter row)")
print("\n--- Patch panel ---\n", flush=True)
slots_s = _prompt(str(def_slots), "Number of front-panel positions (slots)")
try:
slots = int(slots_s)
except ValueError:
print(" Invalid slot count.", file=sys.stderr, flush=True)
raise SystemExit(2) from None
plab = _prompt(def_plab, "Patch panel name / label")
ploc = _prompt(def_ploc, "Patch panel location (e.g. bay, room)")
print("\n--- Hub hosts (hybrid panel calibrate) ---\n", flush=True)
ssh_cfg = SshNodeConfig.load()
from_profile: List[str] = []
if ssh_cfg.calibrate_remotes:
from_profile = [
x.strip()
for x in ssh_cfg.calibrate_remotes.split(",")
if x.strip() and "@" in x
]
if from_profile and ini_path:
print(
f" Discovered from profile: {', '.join(from_profile)}\n",
flush=True,
)
elif from_profile:
print(
f" Discovered from environment: {', '.join(from_profile)}\n",
flush=True,
)
existing_cr = fm.calibrate_remotes_hosts(doc)
if existing_cr:
def_rel = ",".join(existing_cr)
elif from_profile:
def_rel = ",".join(from_profile)
else:
def_rel = ""
rel_raw = _prompt(
def_rel,
"Hub SSH targets (comma-separated user@host); Enter=default; -=omit",
)
if rel_raw.strip() == "-":
relay_hosts: List[str] = []
elif rel_raw.strip():
relay_hosts = [
x.strip()
for x in rel_raw.split(",")
if x.strip() and "@" in x
]
else:
relay_hosts = [
x.strip()
for x in def_rel.split(",")
if x.strip() and "@" in x
]
doc = merge_fiwi_site(doc, concentrator_name=cn, concentrator_location=cl)
doc = merge_patch_panel_meta(doc, slots=slots, label=plab, location=ploc)
doc = merge_calibrate_remotes(doc, relay_hosts)
return doc
def _run_batch(args: argparse.Namespace) -> Dict[str, Any]:
path = fiber_map_path()
exists = os.path.isfile(path)
if exists and not args.merge:
print(
f"{path} exists; use --merge for batch updates.",
file=sys.stderr,
flush=True,
)
raise SystemExit(2)
doc = fm.load_fiber_map_document() if exists else {"fiber_ports": {}}
doc = fm.ensure_fiber_map_document(doc)
if args.slots is None or args.slots < 1:
print("--batch requires --slots N (>= 1).", file=sys.stderr, flush=True)
raise SystemExit(2)
doc = merge_fiwi_site(
doc,
concentrator_name=args.concentrator_name or "",
concentrator_location=args.concentrator_location or "",
)
doc = merge_patch_panel_meta(
doc,
slots=int(args.slots),
label=args.panel_name or "",
location=args.panel_location or "",
)
return doc
def _profile_argv_prefix() -> List[str]:
prof = (os.environ.get("FIWI_CONFIG") or "").strip()
return ["-c", prof] if prof else []
def build_site_setup_argument_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
description=(
"Fi-Wi site setup — map metadata, optional chained relay + calibrate. "
"Relay SSH targets are read from the INI (e.g. [remote_hubs] hosts); "
"--relay/--calibrate only choose whether to run those steps after saving the map. "
"Run from the install directory next to fiwi.py; use -c PROFILE."
),
)
p.add_argument(
"-c",
"--config",
metavar="PROFILE",
help=(
"INI profile or path (sets FIWI_CONFIG before loading paths — e.g. uax24 → "
"patch_panel defaults, remote_hubs hosts for calibrate_remotes prompt)"
),
)
p.add_argument(
"--merge",
action="store_true",
help="Required with --batch when fiber_map.json already exists (interactive uses m/d/q prompt).",
)
p.add_argument(
"--relay",
action="store_true",
help=(
"After saving the map, run fiwi relay bootstrap in-process (FiWiRelayBootstrapApp; "
"same -c profile). Uses [remote_hubs] / env from config — this flag does not set hosts."
),
)
p.add_argument(
"--calibrate",
action="store_true",
help=(
"After saving the map (and after --relay if given), run panel calibrate merge in-process "
"(fiwi.cli.run_panel_calibrate) plus any --ssh. "
"Interactive runs also ask [Y/n] to calibrate unless --no-calibrate-prompt."
),
)
p.add_argument(
"--batch",
action="store_true",
help="Non-interactive: supply --slots and optional name/location strings",
)
p.add_argument("--concentrator-name", default="", metavar="STR")
p.add_argument("--concentrator-location", default="", metavar="STR")
p.add_argument("--panel-name", default="", metavar="STR")
p.add_argument("--panel-location", default="", metavar="STR")
p.add_argument("--slots", type=int, default=None, metavar="N")
p.add_argument(
"--ssh",
action="append",
default=[],
metavar="user@host",
help="Only with --calibrate: extra --ssh user@host for panel calibrate (repeatable)",
)
p.add_argument(
"--no-power-down",
action="store_true",
help=(
"Skip the optional USB power-down prompt when not calibrating (interactive only; "
"redundant with panel calibrate’s baseline if you use --calibrate)"
),
)
p.add_argument(
"--no-calibrate-prompt",
action="store_true",
help=(
"After saving the map, do not ask to start panel calibrate (interactive only); "
"use --calibrate to run it without a prompt."
),
)
return p
class SiteSetupApp:
"""Orchestrates fiber map metadata, optional USB power-down, relay bootstrap, panel calibrate."""
def __init__(self, args: argparse.Namespace) -> None:
self._args = args
def run(self) -> int:
import fiwi.paths as paths_mod
args = self._args
paths_mod.configure(_fiwi_repo_root())
if args.batch:
doc = _run_batch(args)
else:
if not sys.stdin.isatty():
print(
"site_setup: need a TTY for prompts, or use --batch with --slots.",
file=sys.stderr,
flush=True,
)
return 2
doc = _run_interactive(args)
out_path = fiber_map_path()
fm.write_fiber_map_document(doc)
print(f"\nWrote {out_path}", flush=True)
site = read_fiwi_site(doc)
pp = PatchPanel.from_map_blob(doc.get("patch_panel"))
if site:
print(
f" fiwi_site: {site.get('concentrator_name', '—')!r} @ "
f"{site.get('concentrator_location', '—')!r}",
flush=True,
)
if pp:
print(
f" patch_panel: {pp.slots} ports, name={pp.label or '—'!r}, "
f"location={pp.location or '—'!r}",
flush=True,
)
cr = fm.calibrate_remotes_hosts(doc)
if cr:
print(f" calibrate_remotes: {', '.join(cr)}", flush=True)
map_p = fiber_map_path()
prof = (os.environ.get("FIWI_CONFIG") or "").strip()
ex_prof = prof if prof else "uax24"
do_calibrate = bool(args.calibrate)
if (
not args.batch
and sys.stdin.isatty()
and not do_calibrate
and not args.no_calibrate_prompt
):
try:
ans = input(
"\nStart panel calibrate now (merge from map)? [Y/n]: "
).strip().lower()
except EOFError:
ans = "n"
if not ans or ans in ("y", "yes"):
do_calibrate = True
if not do_calibrate:
if not args.batch and sys.stdin.isatty():
_run_early_usb_power_down_if_requested(doc, args)
print(
"\nStopped after saving the map (no calibrate this run).\n"
"Optional chained steps — relay hosts come from config/ INI, not the CLI:\n"
f" site_setup.py -c {ex_prof} --relay\n"
f" site_setup.py -c {ex_prof} --calibrate\n"
f" site_setup.py -c {ex_prof} --relay --calibrate [--ssh user@host …]\n"
f"If calibrate skips the Pi: check FIWI_FIBER_MAP vs {map_p}; "
f"bootstrap with --relay if needed.\n"
"Per-strand labels: fiber_ports[\"<id>\"].name / .location\n",
flush=True,
)
else:
print(
"\nContinuing to panel calibrate…\n"
"(With `merge`, fiwi asks once to confirm patch panel from the map — Enter to keep, "
"or `e` to change slots/label/location.)\n",
flush=True,
)
root = _fiwi_repo_root()
if args.relay:
if not sys.stdin.isatty():
print("site_setup: --relay needs a TTY.", file=sys.stderr, flush=True)
return 2
from fiwi.fiwi_relay.relay_app import FiWiRelayBootstrapApp
relay_argv = _profile_argv_prefix()
print(
"\nRunning fiwi relay bootstrap (in-process"
+ (f"; {' '.join(relay_argv)}" if relay_argv else "")
+ ")…\n",
flush=True,
)
relay_rc = FiWiRelayBootstrapApp(argv=relay_argv, install_root=root).run()
if relay_rc != 0:
return relay_rc
if do_calibrate:
if not sys.stdin.isatty():
print("site_setup: panel calibrate needs a TTY.", file=sys.stderr, flush=True)
return 2
from fiwi.cli import run_panel_calibrate
extra_ssh = [str(h).strip() for h in (args.ssh or []) if str(h).strip()]
print(
"\nRunning panel calibrate (in-process, merge=True)"
+ (f"; extra --ssh: {', '.join(extra_ssh)}" if extra_ssh else "")
+ ".\n",
flush=True,
)
return run_panel_calibrate(
merge=True,
limit=None,
calibrate_ssh_hosts=extra_ssh,
emit_start_line=True,
)
return 0
def main(argv: List[str] | None = None) -> int:
args = build_site_setup_argument_parser().parse_args(argv)
if args.config:
os.environ["FIWI_CONFIG"] = args.config.strip()
return SiteSetupApp(args).run()
if __name__ == "__main__":
raise SystemExit(main())

8
fiwi/ssh_dispatch.py
View File

@ -5,7 +5,7 @@ import sys
from fiwi.patch_panel import effective_panel_slots
from fiwi import fiber_map_io as fm
from fiwi.fiber_radio_port import FiberRadioPort
from fiwi.radiohead import RadioHeadEntry
from fiwi.ssh import SshNode, apply_fiwi_ssh_env
@ -32,7 +32,7 @@ def dispatch_fiber_mapped_ssh_if_needed(argv):
fid = int(argv[3])
except ValueError:
return None
node = FiberRadioPort.from_port_id(doc, fid).ssh_node()
node = RadioHeadEntry.from_port_id(doc, fid).ssh_node()
if node:
return node.invoke(
["power", "fiber-port", str(fid), argv[4].lower()],
@ -44,7 +44,7 @@ def dispatch_fiber_mapped_ssh_if_needed(argv):
fid = int(argv[2])
except ValueError:
return None
node = FiberRadioPort.from_port_id(doc, fid).ssh_node()
node = RadioHeadEntry.from_port_id(doc, fid).ssh_node()
if node:
print(
"fiber chip: not forwarded over SSH — radios are PCIe/fiber; use panel calibrate PCIe prompts "
@ -64,7 +64,7 @@ def dispatch_fiber_mapped_ssh_if_needed(argv):
n_slots = effective_panel_slots(doc)
if pn < 1 or pn > n_slots:
return None
node = FiberRadioPort.from_port_id(doc, pn).ssh_node()
node = RadioHeadEntry.from_port_id(doc, pn).ssh_node()
if node:
return node.invoke(["panel", sub, str(pn)], defer=False)

0
fiwi_link/fiwi_relay/__main__.py
View File

28
peak_detect.reflex
View File

@ -1,28 +0,0 @@
#include <aUSBHub3p.reflex>
// Create a reference to the module
aUSBHub3p hub;
// Map scratchpad locations
// pad[offset : length]
pad[0:4] signed int peak;
pad[4:4] signed int duration;
reflex every_1ms() {
// Current in micro-amps
signed int current = hub.usb.getPortCurrent(4);
if (current > peak) {
peak = current;
}
// If drawing more than 50mA, increment duration counter
if (current > 50000) {
duration++;
}
}
reflex mapEnable() {
peak = 0;
duration = 0;
}

5
remote_ssh.env.example
View File

@ -18,7 +18,8 @@ FIWI_REMOTE_SCRIPT=~/Code/FiWiManager/fiwi.py
#
# Optional SSH client tweaks:
# FIWI_SSH_BIN=ssh
# FIWI_SSH_OPTS=-o BatchMode=yes
# Non-interactive panel calibrate uses captured SSH (no password prompt). Use keys + optional:
# FIWI_SSH_OPTS=-o BatchMode=yes -o ConnectTimeout=15
#
# Optional: deferred remote calls (spawn ssh children; panel calibrate overlaps SSH via Popen):
# FIWI_REMOTE_DEFER=1
@ -31,6 +32,6 @@ FIWI_REMOTE_SCRIPT=~/Code/FiWiManager/fiwi.py
#
# Remote BrainStem hosts (comma-separated user@host); merged with FIWI_CALIBRATE_REMOTES for panel calibrate:
# FIWI_REMOTE_HUBS=rjmcmahon@192.168.1.39
# FIWI_FIBER_MAP=fiber_map.json
# FIWI_FIBER_MAP=maps/fiber_map.json
# FIWI_DEFAULT_PANEL_PORTS=24
# FIWI_DEFAULT_PANEL_SLOTS=24 ; legacy alias, same meaning

7
site_setup.py Normal file
View File

@ -0,0 +1,7 @@
#!/usr/bin/env python3
"""CLI entry: Fi-Wi site_setup — map + optional relay + optional calibrate (see fiwi.site_setup)."""
from fiwi.site_setup import main
if __name__ == "__main__":
raise SystemExit(main())

109
tests/check_concentrator.py
View File

@ -29,6 +29,10 @@ and ``port-metrics-json``; the remote tree must include that command (same revis
``FIWI_CALIBRATE_REMOTES`` / merged hub hosts: all ports OFF (verify), then all ON (verify),
then prints the port power table (snapshot after the test).
After the per-port power table, the script builds a :class:`fiwi.radiohead.RadioHead` for every
mapped ``fiber_ports`` row (panel slots first, then any other keys) and prints panel, saved chip
type, and live power / mA / V.
``--inrush HUB PORT`` (after the report, or after ``--powercycle``) runs the same USB inrush probe as
``tests/check_inrush.py`` on **local** hubs only (1-based hub index, 0-based port). Use
``--inrush-host-sample`` for host-side polling; default is on-hub Reflex scratchpad.
@ -908,6 +912,109 @@ def _print_per_port_power_table(
return rc
_RADIO_HEADS_HDR = (
f"{'Panel':<8} | {'chip_type (map)':<28} | {'Pwr':<5} | {'mA':>10} | {'V':>8} | {'Hub.Pt':<8}"
)
def _print_radio_heads_section(c: FiWiConcentrator) -> None:
"""
Walk every mapped radio head: panel-bound via :meth:`~fiwi.concentrator.FiWiConcentrator.patch_panel`
``.heads()``, then any other mapped ``fiber_ports`` rows (off-panel fiber ids).
Uses :class:`fiwi.radiohead.RadioHead` ``power`` / ``current`` / ``voltage`` (BrainStem or SSH
``port-metrics-json`` per head).
"""
from fiwi import fiber_map_io as fm
from fiwi.radiohead import RadioHead, RadioHeadEntry
print("Radio heads (all mapped fiber_ports → RadioHead)", flush=True)
print("-" * len(_RADIO_HEADS_HDR), flush=True)
doc = fm.load_fiber_map_document()
if not doc:
print(" (no fiber_map.json — skip.)", flush=True)
print(flush=True)
return
try:
from fiwi.site_setup import read_fiwi_site
site = read_fiwi_site(doc)
if site:
cn = site.get("concentrator_name") or "—"
cl = site.get("concentrator_location") or "—"
print(
f" Fi-Wi concentrator: {cn} | Location: {cl}",
flush=True,
)
bound = c.patch_panel(doc)
pp = bound.panel
name_s = pp.label.strip() if pp.label.strip() else "—"
loc_s = pp.location.strip() if pp.location.strip() else "—"
print(
f" Patch panel name: {name_s} | Location: {loc_s} | Ports: {pp.slots}",
flush=True,
)
print(flush=True)
heads: list[RadioHead] = list(bound.heads())
seen_map_keys = {rh.map_entry.map_key for rh in heads}
for ent in RadioHeadEntry.each_from_document(doc):
if not ent.is_mapped() or ent.map_key in seen_map_keys:
continue
heads.append(RadioHead(ent, c))
seen_map_keys.add(ent.map_key)
def _rh_sort_key(rh: RadioHead) -> tuple:
p = rh.patch_panel_port
if p is not None:
return (0, p, rh.map_entry.map_key)
mk = rh.map_entry.map_key
if mk.isdigit():
return (1, int(mk), mk)
return (2, mk, mk)
heads.sort(key=_rh_sort_key)
except (OSError, ValueError) as exc:
print(f" ! Could not load patch panel / map: {exc}", flush=True)
print(flush=True)
return
if not heads:
print(" (no mapped fiber_ports entries.)", flush=True)
print(flush=True)
return
print(_RADIO_HEADS_HDR, flush=True)
print("-" * len(_RADIO_HEADS_HDR), flush=True)
for rh in heads:
panel_s = str(rh.patch_panel_port) if rh.patch_panel_port is not None else "—"
chip = rh.chip_type or "—"
if len(chip) > 28:
chip = chip[:25] + "..."
hp = rh.map_entry.hub_port()
hub_pt = f"{hp[0]}.{hp[1]}" if hp else "—"
rh.refresh_live()
pwr_b = rh.power
if pwr_b is True:
pwr_s = "ON"
elif pwr_b is False:
pwr_s = "OFF"
else:
pwr_s = "?"
cur = rh.current
ma_s = f"{cur:.2f}" if isinstance(cur, (int, float)) else "—"
vv = rh.voltage
v_s = f"{vv:.3f}" if isinstance(vv, (int, float)) else "—"
print(
f"{panel_s:<8} | {chip:<28} | {pwr_s:<5} | {ma_s:>10} | {v_s:>8} | {hub_pt:<8}",
flush=True,
)
print(flush=True)
def _print_usb_hub_tables(c: FiWiConcentrator, hub_rows: list[ConsolidatedHubRow], base_rc: int) -> int:
"""Print consolidated hub table and port power / current / voltage table. Returns updated rc."""
rc = base_rc
@ -1125,6 +1232,7 @@ def _print_consolidated_report(c: FiWiConcentrator) -> int:
hub_rows, remote_rc = _build_consolidated_hub_rows(c)
hub_rows = _enrich_hub_rows_usb(hub_rows)
remote_rc = _print_usb_hub_tables(c, hub_rows, remote_rc)
_print_radio_heads_section(c)
_print_pcie_catalog_section()
return remote_rc
@ -1376,6 +1484,7 @@ def main() -> int:
remote_fail = remote_fail or brc
hub_rows = _enrich_hub_rows_usb(hub_rows)
remote_fail = _print_per_port_power_table(c, hub_rows, remote_fail)
_print_radio_heads_section(c)
else:
remote_fail = _print_consolidated_report(c)
if args.inrush is not None: