diff --git a/CHANGELOG.md b/CHANGELOG.md index 11c27a1a..39422efc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,36 @@ All notable changes to the Lager platform are documented here. For detailed release notes, see [docs.lagerdata.com](https://docs.lagerdata.com). +## [0.31.10] - 2026-07-13 + +### Added +- **`lager battery models` lists the battery models saved on the instrument** — + occupied memory slots plus the firmware built-ins, all valid inputs to the + existing `model` command. Read-only, and also exposed in the battery TUI and + the box's `/battery/command` endpoint (`list_models`). +- **8 more Logitech webcams are detected and usable as webcam nets**: Logi 4K + Pro, BRIO 4K Stream, C925e, C922 Pro, C920, C615, C270, and StreamCam join + the existing BRIO/BRIO HD/C930e. Camera detection is now catalog-driven + (adding a model is a table entry, not code) and maps each camera to its + actual /dev/video capture node via sysfs, so mixed setups with different + per-camera node counts resolve correctly. + +### Fixed +- **Battery model readback reports the actual loaded model.** The driver read + `:BATT:STAT?`, which reports charge/discharge status rather than the model, + so `model`, `state`, and the TUI always showed "DISCHARGE" and successful + loads raised a false "slot is empty" error. Readback and verification now + use `:BATT:MOD:RCL?`, which also reliably detects the 2281S's silent + empty-slot recall failures. +- **Built-in battery models are loadable over SCPI.** The 2281S rejects the + manual's hyphenated names (`LI-ION4_2`) with a syntax error; the driver now + sends the underscore spellings the firmware accepts and takes either form + as input. +- **Battery driver errors are no longer masked as "[Errno 16] Resource busy".** + The box endpoint returned driver errors as 5xx, tripping the CLI's legacy + direct-USB fallback; they now come back as ordinary command failures so the + real message is shown. + ## [0.31.9] - 2026-07-13 ### Fixed diff --git a/box/lager/http_handlers/battery.py b/box/lager/http_handlers/battery.py index 7406df8a..4510b1ec 100644 --- a/box/lager/http_handlers/battery.py +++ b/box/lager/http_handlers/battery.py @@ -29,6 +29,25 @@ logger = logging.getLogger(__name__) +def _device_error_message(exc) -> str: + """Extract the driver's own message from a hardware_service DeviceError. + + Driver exceptions come back from /invoke as + DeviceError({'error': 'Function call failed: ', 'details': }); + return so callers see e.g. the set_model empty-slot guidance + instead of a JSON blob with a traceback in it. + """ + payload = exc.args[0] if getattr(exc, "args", None) else None + if isinstance(payload, dict): + msg = (payload.get('error') or '').strip() + prefix = 'Function call failed: ' + if msg.startswith(prefix): + msg = msg[len(prefix):] + if msg: + return msg + return describe_error(exc) + + def _resolve_battery_proxy(netname: str): """ Build a transient battery Device proxy + channel for `netname` by routing @@ -280,9 +299,21 @@ def battery_command_http(): battery.set_model(partnumber) result['message'] = f'Model set to {partnumber}' else: - model = battery._safe_query(':BATT:STAT?', '') or 'Custom' + # current_model reads :BATT:MOD:RCL? — :BATT:STAT? + # reports charge/discharge status, not the model. + model = battery.current_model() result['message'] = f'Model: {model}' + elif action in ('list_models', 'models'): + # Read-only catalog of battery models available on the + # instrument. Mirrors the 'state' action: the structured + # payload rides in the response ('models') alongside a + # human-readable message rendered by the shared formatter. + from lager.power.battery.dispatcher import format_model_catalog + models = battery.model_catalog() + result['models'] = models + result['message'] = format_model_catalog(models) + elif action == 'enable_battery': battery.enable() result['message'] = 'Battery output enabled' @@ -346,7 +377,18 @@ def _fmt(value, unit): logger.info(f"[HTTP] Battery command executed: {action} on {netname} (session {session_id})") return jsonify(result) - except (ConnectionFailed, DeviceError) as e: + except DeviceError as e: + # The driver itself raised (e.g. set_model's empty-slot + # guidance): that is an ANSWER, not a transport failure. + # Return it as success:false at 200 — a 5xx makes the CLI + # treat the endpoint as unavailable and fall through to its + # legacy direct-USB path, which always fails with + # "[Errno 16] Resource busy" while hardware_service holds + # the instrument, masking the real message. + logger.warning( + f"[HTTP] Battery command {action} on {netname} rejected by driver: {e}") + return jsonify({'success': False, 'error': _device_error_message(e)}), 200 + except ConnectionFailed as e: logger.exception(f"[HTTP] Battery command {action} on {netname} failed at hardware_service") return jsonify({'success': False, 'error': f'Hardware service error: {e}'}), 502 @@ -744,9 +786,19 @@ def handle_battery_command(data): battery.set_model(partnumber) result['message'] = f'Model set to {partnumber}' else: - model = battery._safe_query(':BATT:STAT?', '') or 'Custom' + # current_model reads :BATT:MOD:RCL? — :BATT:STAT? + # reports charge/discharge status, not the model. + model = battery.current_model() result['message'] = f'Model: {model}' + elif action == 'models': + # Read-only model catalog; same payload/message pair as + # the HTTP endpoint's 'list_models' action. + from lager.power.battery.dispatcher import format_model_catalog + models = battery.model_catalog() + result['models'] = models + result['message'] = format_model_catalog(models) + elif action == 'enable': battery.enable() result['message'] = 'Battery output enabled' @@ -782,7 +834,15 @@ def handle_battery_command(data): logger.info(f"[BATTERY-COMMAND-{thread_name}] Command completed successfully: {result}") emit('battery_command_response', result) - except (ConnectionFailed, DeviceError) as e: + except DeviceError as e: + # Driver-raised error: show the driver's message, not the + # /invoke JSON blob (see _device_error_message). + logger.warning(f"[BATTERY-COMMAND-{thread_name}] driver error: {e}") + emit('battery_command_response', { + 'success': False, + 'error': _device_error_message(e) + }) + except ConnectionFailed as e: logger.error(f"[BATTERY-COMMAND-{thread_name}] hardware_service error: {e}") emit('battery_command_response', { 'success': False, diff --git a/box/lager/http_handlers/usb_scanner.py b/box/lager/http_handlers/usb_scanner.py index 1427f0b9..3839eb23 100644 --- a/box/lager/http_handlers/usb_scanner.py +++ b/box/lager/http_handlers/usb_scanner.py @@ -136,10 +136,20 @@ def wrapper(*args, **kwargs) -> T: "YKUSH_Hub": {"vid": "04d8", "pid": "f2f7", "net_type": ["usb"]}, # eload "Rigol_DL3021": {"vid": "1ab1", "pid": "0e11", "net_type": ["eload"]}, - # camera + # camera — detection is catalog-driven: any entry with a ``webcam`` + # net_type is picked up by ``_by_camera`` below, so adding a camera + # is a one-line addition here (mirror it in cli/impl/query_instruments.py). "Logitech_BRIO_HD": {"vid": "046d", "pid": "085e", "net_type": ["webcam"]}, "Logitech_BRIO": {"vid": "046d", "pid": "0856", "net_type": ["webcam"]}, + "Logitech_BRIO_4K_Stream": {"vid": "046d", "pid": "086b", "net_type": ["webcam"]}, + "Logitech_4K_Pro": {"vid": "046d", "pid": "087f", "net_type": ["webcam"]}, "Logitech_C930e": {"vid": "046d", "pid": "0843", "net_type": ["webcam"]}, + "Logitech_C925e": {"vid": "046d", "pid": "085b", "net_type": ["webcam"]}, + "Logitech_C922_Pro": {"vid": "046d", "pid": "085c", "net_type": ["webcam"]}, + "Logitech_C920": {"vid": "046d", "pid": "082d", "net_type": ["webcam"]}, + "Logitech_C615": {"vid": "046d", "pid": "082c", "net_type": ["webcam"]}, + "Logitech_C270": {"vid": "046d", "pid": "0825", "net_type": ["webcam"]}, + "Logitech_StreamCam": {"vid": "046d", "pid": "0893", "net_type": ["webcam"]}, # watt-meter "Yocto_Watt": {"vid": "24e0", "pid": "002a", "net_type": ["watt-meter"]}, "Joulescope_JS220": {"vid": "16d0", "pid": "10ba", "net_type": ["watt-meter", "energy-analyzer"]}, @@ -545,30 +555,63 @@ def scan_usb() -> List[dict]: # Camera detection # --------------------------------------------------------------------------- -_CAM_VID = "046d" -_CAM_PIDS = {"085e", "0856", "0843"} +_WEBCAM_VIDPIDS = { + (_meta["vid"].lower(), _meta["pid"].lower()) + for _meta in SUPPORTED_USB.values() + if "webcam" in _meta["net_type"] +} -def _by_camera() -> List[dict]: - usb_cams = [ - dev for dev in scan_usb() - if dev["vid"] == _CAM_VID and dev["pid"] in _CAM_PIDS - ] - if not usb_cams: - return [] +def _read_sysfs_attr(path: Path) -> Optional[str]: + try: + return path.read_text().strip() + except OSError: + return None + +def _by_camera(v4l_root: Path = Path("/sys/class/video4linux")) -> List[dict]: + """Detect supported webcams and their /dev/video capture nodes. + + Each ``/sys/class/video4linux/videoN`` links to the USB interface that + owns it; the interface's parent directory is the USB device carrying + idVendor/idProduct/serial. Walking that link keeps the node→camera + mapping correct when models expose different numbers of video nodes + (a C920 exposes two, a BRIO four), which an index-based heuristic + cannot get right on mixed setups. + """ + results: List[dict] = [] + seen_devices: set = set() video_nodes = sorted( - (Path(p) for p in glob.glob("/dev/video*")), + v4l_root.glob("video*"), key=lambda p: int(p.name.replace("video", "")), ) - - results: List[dict] = [] - for idx, cam in enumerate(usb_cams): + for node in video_nodes: try: - video_dev = str(video_nodes[idx * 4]) - except IndexError: + usb_dev = (node / "device").resolve().parent + except OSError: + continue + vid = _read_sysfs_attr(usb_dev / "idVendor") + pid = _read_sysfs_attr(usb_dev / "idProduct") + if not vid or not pid: + continue + vid, pid = vid.lower(), pid.lower() + if (vid, pid) not in _WEBCAM_VIDPIDS: continue - results.append({**cam, "channels": {"webcam": [video_dev]}}) + if str(usb_dev) in seen_devices: + # Higher-numbered nodes on the same camera are metadata/IR + # streams; the first node is the capture stream. + continue + seen_devices.add(str(usb_dev)) + serial = _read_sysfs_attr(usb_dev / "serial") + results.append({ + "name": _VIDPID_TO_NAME[(vid, pid)], + "vid": vid, + "pid": pid, + "serial": serial, + "address": f"USB0::0x{vid.upper()}::0x{pid.upper()}::{serial or ''}::INSTR", + "net_type": ["webcam"], + "channels": {"webcam": [f"/dev/{node.name}"]}, + }) return results diff --git a/box/lager/power/battery/__init__.py b/box/lager/power/battery/__init__.py index c8a14bfb..f6878171 100644 --- a/box/lager/power/battery/__init__.py +++ b/box/lager/power/battery/__init__.py @@ -34,6 +34,8 @@ set_ovp, set_ocp, set_model, + list_models, + format_model_catalog, enable_battery, disable_battery, print_state, @@ -68,6 +70,8 @@ "set_ovp", "set_ocp", "set_model", + "list_models", + "format_model_catalog", "enable_battery", "disable_battery", "print_state", diff --git a/box/lager/power/battery/battery_net.py b/box/lager/power/battery/battery_net.py index 00d4f2ce..e7491e48 100644 --- a/box/lager/power/battery/battery_net.py +++ b/box/lager/power/battery/battery_net.py @@ -155,6 +155,27 @@ def model(self, partnumber: str | None = None) -> None: """ pass + @abstractmethod + def model_catalog(self) -> list: + """ + Read the catalog of battery models available on the instrument. + + Lists the same slot-storage mechanism that model() loads from: + numbered memory slots plus any firmware built-in models. Must be + read-only — assembling the catalog must not change instrument state. + + Keithley 2281S: + - Slot 0 ('discharge') is always available + - Slots 1-9 are reported when a model has been saved there + - The five firmware built-in models are listed without a slot + + Returns: + List of {"slot": int | None, "name": str | None} dicts, one per + available model. Slot-less entries are built-in model names; + nameless entries are custom models addressed by slot index. + """ + pass + @abstractmethod def enable(self) -> None: """ diff --git a/box/lager/power/battery/dispatcher.py b/box/lager/power/battery/dispatcher.py index 53026e10..01b7ca09 100644 --- a/box/lager/power/battery/dispatcher.py +++ b/box/lager/power/battery/dispatcher.py @@ -15,7 +15,7 @@ from lager.exceptions import BatteryBackendError, LibraryMissingError, DeviceNotFoundError from .battery_net import BatteryNet -from .keithley import KeithleyBattery +from .keithley import KeithleyBattery, GREEN, RESET class BatteryDispatcher(BaseDispatcher): @@ -207,6 +207,36 @@ def set_model(netname: str, partnumber: str | None = None, **_): drv.model(partnumber=partnumber) +def format_model_catalog(models) -> str: + """Render a structured model catalog as a small slot/name table. + + Shared by list_models (printed on the impl-script path) and the box's + /battery/command HTTP endpoint (returned as the response message) so + the two transports never drift. + """ + lines = ["Slot Model"] + for entry in models: + slot = entry.get("slot") + name = entry.get("name") + slot_str = "-" if slot is None else str(slot) + if name is None: + name = "(custom model)" + elif slot is None: + name = f"{name} (built-in)" + lines.append(f"{slot_str:<4} {name}") + lines.append("") + lines.append("Slots and names above are valid inputs to the 'model' command.") + return "\n".join(lines) + + +def list_models(netname: str, **_): + """List battery models saved on the instrument.""" + drv, _ = _dispatcher._resolve_net_and_driver(netname) + catalog = drv.model_catalog() + print(f"{GREEN}{format_model_catalog(catalog)}{RESET}") + return {"models": catalog} + + def enable_battery(netname: str, **_): """Enable battery simulator output.""" drv, _ = _dispatcher._resolve_net_and_driver(netname) diff --git a/box/lager/power/battery/keithley.py b/box/lager/power/battery/keithley.py index 1ff6b856..91f74857 100644 --- a/box/lager/power/battery/keithley.py +++ b/box/lager/power/battery/keithley.py @@ -28,6 +28,15 @@ RED = '\033[91m' RESET = '\033[0m' +# The five battery models built into 2281S firmware, recallable by name via +# :BATT:MOD:RCL (reference manual 077114601, section 7-35). They live in +# firmware, not in the numbered memory slots. NOTE: the manual prints two of +# them with hyphens (LI-ION4_2, LEAD-ACID12), but a hyphen is unparseable as +# SCPI character data (-102 "Syntax error; - :BATT:MOD:RCL LI-", +# hardware-verified) — the underscore spellings below are what the parser +# accepts and what :BATT:MOD:RCL? reports back. +BUILTIN_MODELS = ("LI_ION4_2", "NIMH1_2", "NICD1_2", "LEAD_ACID12", "NIMH12") + class KeithleyBattery(BatteryNet): """ @@ -241,13 +250,134 @@ def ocp(self, value: float | None = None) -> None: def model(self, partnumber: str | None = None) -> None: if partnumber is None: - model_str = self._safe_query(":BATT:STAT?", "") - if not model_str: - model_str = self._safe_query(":BATT:MOD?", "0") - print(f"{GREEN}{model_str}{RESET}") + print(f"{GREEN}{self.current_model()}{RESET}") return self.set_model(partnumber) + def current_model(self) -> str: + """Name of the active battery model, via :BATT:MOD:RCL?. + + (:BATT:STAT? — which older code used here — reports charge/discharge + status per the reference manual, NOT the model, so it shows + 'DISCHARGE' whenever the output is idle regardless of what is + loaded.) + + RCL? answers with the slot number while a numbered slot is active and + with the model name while a firmware built-in is active. If it gives + no reply (transient — e.g. right after a rejected command corrupts + the parser's input buffer), fall back to the name cached by the last + successful set_model. + """ + raw = self._active_model_raw() + if raw: + name = ("DISCHARGE" if raw == "0" else f"slot {raw}") if raw.isdigit() else raw + self._active_model_name = name + return name + return getattr(self, "_active_model_name", None) or "Custom" + + def _active_model_raw(self) -> str: + """Raw :BATT:MOD:RCL? reply, '' when the instrument does not answer. + + Hardware-verified 2281S behavior: the query answers with the slot + number while a numbered slot (0-9) is active and with the model name + while a firmware built-in is active. It can transiently give no reply + (e.g. a previously rejected command corrupting the parser's input + buffer leaves the next query unanswered) — shrink the VISA timeout + around the query so that silence costs <1 s instead of the full 5 s + (this runs inside the monitor tick and the Device proxy's 10 s + /invoke budget). + """ + old_timeout = None + try: + old_timeout = self.instr.timeout + self.instr.timeout = 800 # ms + except Exception: + pass + try: + return self._safe_query(":BATT:MOD:RCL?", "").strip().strip('"') + finally: + if old_timeout is not None: + try: + self.instr.timeout = old_timeout + except Exception: + pass + + def model_catalog(self) -> list[dict]: + """ + Read the catalog of battery models available on the instrument. + + The 2281S reference manual (077114601, March 2019) defines no catalog + query — :BATT:MODel:CATalog? does not exist — so the catalog is + assembled from the documented read-only model-memory queries instead: + + - Slot 0 (DISCHARGE, basic constant-voltage simulation) is always + available. + - Slots 1-9 are probed with :BATT:MOD:VOC:STEPs? (query only), + which returns the number of values stored in that model. A slot is + occupied when the length is non-zero. Internal memory stores models + by index only, so occupied slots have no retrievable name. + - The five firmware built-in models (recallable by name via + :BATT:MOD:RCL) are appended with slot None. + + Read-only: issues only queries and never forces the Battery entry + function — the model memory is not documented as mode-restricted. + + Returns: + List of {"slot": int | None, "name": str | None} dicts. + + Raises: + BatteryBackendError: If the instrument rejects the model-memory + query (older firmware) or does not answer it at all. + """ + models = [{"slot": 0, "name": "DISCHARGE"}] + for idx in range(1, 10): + try: + raw = self._query(f":BATT:MOD{idx}:VOC:STEP?") + except Exception as exc: + code = self._scpi_error_code(exc) + if code is not None and -199 <= code <= -100: + # SCPI command errors (-113 "Undefined header", etc.): + # this firmware has no model-memory queries. + raise BatteryBackendError( + "This instrument does not support the battery model " + f"memory query (:BATT:MOD{idx}:VOC:STEPs?): {exc}" + ) from exc + if code is None: + # No response at all (e.g. VISA timeout): bail out on the + # first probe instead of timing out on all nine slots. + raise BatteryBackendError( + f"Battery model catalog query failed: {exc}" + ) from exc + # Any other instrument error just means this slot holds no + # usable model (e.g. empty slot); keep probing the rest. + continue + try: + steps = int(float(raw.strip() or "0")) + except (TypeError, ValueError): + steps = 0 + if steps > 0: + models.append({"slot": idx, "name": None}) + models.extend({"slot": None, "name": name} for name in BUILTIN_MODELS) + return models + + @staticmethod + def _scpi_error_code(exc) -> int | None: + """Extract the SCPI error code from an instrument exception. + + InstrumentError carries (code, message, response) in args; other + wrappers stringify to ',""', possibly behind a + prefix (BatteryBackendError renders as '[Battery] ,"..."'). + Returns None when no code can be found (e.g. a transport-level + timeout). + """ + args = getattr(exc, "args", ()) + if args and isinstance(args[0], int): + return args[0] + match = re.search(r'(-?\d+)\s*,\s*"', str(exc)) + if match: + return int(match.group(1)) + return None + def terminal_voltage(self) -> float: try: return float(self._safe_query(":BATT:SIM:TVOL?", "0")) @@ -487,10 +617,12 @@ def set_model(self, partnumber_or_index) -> None: if name in model_map: cmd = f":BATT:MOD:RCL {model_map[name]}" else: - # Check if it's already a valid Keithley built-in - keithley_builtins = ["LI-ION4_2", "NIMH1_2", "NICD1_2", "LEAD-ACID12", "NIMH12", "DISCHARGE"] - if partnumber_or_index.upper() in keithley_builtins: - cmd = f':BATT:MOD:RCL {partnumber_or_index.upper()}' + # Check if it's already a valid Keithley built-in. Accept the + # manual's hyphenated spellings (LI-ION4_2) but always SEND + # the underscore form — the SCPI parser rejects hyphens. + builtin_token = partnumber_or_index.upper().replace("-", "_") + if builtin_token in BUILTIN_MODELS + ("DISCHARGE",): + cmd = f':BATT:MOD:RCL {builtin_token}' else: # Provide helpful suggestions for common typos suggestions = [] @@ -516,27 +648,45 @@ def set_model(self, partnumber_or_index) -> None: # Use output management since model changes may require output to be off self._set_with_output_management(cmd, ignore_codes=(711, -222)) - # Verify something actually loaded; provide helpful guidance - model_str = self._safe_query(":BATT:STAT?", "") - if not model_str or "DISCHARGE" in model_str: - # DISCHARGE is the default/empty state - # Only error if user didn't explicitly request DISCHARGE or slot 0 - requested_discharge = ( - str(partnumber_or_index).upper() == "DISCHARGE" or - str(partnumber_or_index) == "0" or - name == "discharge" - ) - if not requested_discharge: - # Provide helpful guidance about model slots + # Verify via the active-model query (:BATT:MOD:RCL?). Hardware-verified + # 2281S behavior: recalling an EMPTY slot fails silently (no error + # queued, previous model stays active), so readback is the only + # detection; RCL? answers with the slot number when a numbered slot is + # active and with the model NAME when a firmware built-in is active. + # (The old check read :BATT:STAT?, which per the manual reports + # charge/discharge status — it said "DISCHARGE" for every idle output + # and made every successful numbered-slot load raise "slot is empty".) + target = cmd.split()[-1] + actual = self._active_model_raw() + if target.lstrip("+-").isdigit(): + if actual == str(int(target)): + self._active_model_name = ( + "DISCHARGE" if int(target) == 0 else f"slot {int(target)}" + ) + else: + shown = actual if actual else "unchanged (no reply)" raise BatteryBackendError( - f"Battery model slot '{partnumber_or_index}' is empty.\n" + f"Battery model slot '{partnumber_or_index}' appears to be empty — " + f"the recall did not take effect (active model: {shown}).\n" f" The Keithley 2281S stores battery models in numbered memory slots (0-9).\n" - f" This slot doesn't contain a saved model yet.\n" f" Options:\n" + f" • Use 'models' to list the slots that hold a saved model\n" f" • Use 'discharge' for basic constant-voltage simulation\n" f" • Use '18650', 'liion', 'nimh', 'nicd', or 'lead-acid' for common battery types\n" f" • Save a custom battery model to this slot using the instrument front panel" ) + else: + # Built-in recall: RCL? echoes the built-in's name on success; a + # numbered answer (or anything else) means the recall did not + # take effect and the previous model is still loaded. + if actual.upper() != target.upper(): + shown = (f"model slot {actual}" if actual.isdigit() + else (actual or "no reply")) + raise BatteryBackendError( + f"Battery model '{partnumber_or_index}' did not load — " + f"the instrument reports: {shown}." + ) + self._active_model_name = actual # ----------------------------- state dump ----------------------------- @@ -544,7 +694,7 @@ def set_model(self, partnumber_or_index) -> None: def print_state(self) -> None: enabled = self._is_batt_output_on() mode_str = self._mode_string() - model_str = self._safe_query(":BATT:STAT?", "") or "Custom" + model_str = self.current_model() tvol = self._safe_query(":BATT:SIM:TVOL?", "0") curr = self._safe_query(":BATT:SIM:CURR?", "0") @@ -603,7 +753,7 @@ def get_monitor_state(self, channel=None) -> dict: enabled = self._is_batt_output_on() mode_str = self._mode_string() - model_str = self._safe_query(":BATT:STAT?", "") or "Custom" + model_str = self.current_model() trip = (self._safe_query(":OUTP:PROT:TRIP?", "").upper() or "") diff --git a/cli/__init__.py b/cli/__init__.py index b0ceb27f..39c7cfe4 100644 --- a/cli/__init__.py +++ b/cli/__init__.py @@ -7,4 +7,4 @@ A Command Line Interface for Lager Data """ -__version__ = '0.31.9' +__version__ = '0.31.10' diff --git a/cli/battery/battery_tui.py b/cli/battery/battery_tui.py index 506df25d..c4852c7a 100644 --- a/cli/battery/battery_tui.py +++ b/cli/battery/battery_tui.py @@ -345,6 +345,7 @@ def _show_help(self) -> None: ("ovp [VALUE] ", "bold yellow", " - Set or read over-voltage protection (e.g., 'ovp 4.5')", ""), ("mode [static|dynamic] ", "bold yellow", " - Set or read simulation mode", ""), ("model [NAME] ", "bold yellow", " - Set or read battery model (e.g., 'model 18650')", ""), + ("models ", "bold yellow", " - List battery models saved on the instrument", ""), ("enable ", "bold yellow", " - Enable battery output", ""), ("disable ", "bold yellow", " - Disable battery output", ""), ("state ", "bold yellow", " - Display current state", ""), @@ -530,6 +531,9 @@ def _parse_command(self, action: str, parts: list) -> tuple: partnumber = None return ("model", {"partnumber": partnumber}, value_set) + elif action == "models": + return ("models", {}, None) + elif action == "enable": return ("enable", {}, None) diff --git a/cli/commands/box/nets.py b/cli/commands/box/nets.py index 3d84d666..2e214d9e 100644 --- a/cli/commands/box/nets.py +++ b/cli/commands/box/nets.py @@ -383,7 +383,15 @@ def _canonical_role(role: str) -> str: # webcam "Logitech_BRIO_HD": ["webcam"], "Logitech_BRIO": ["webcam"], + "Logitech_BRIO_4K_Stream": ["webcam"], + "Logitech_4K_Pro": ["webcam"], "Logitech_C930e": ["webcam"], + "Logitech_C925e": ["webcam"], + "Logitech_C922_Pro": ["webcam"], + "Logitech_C920": ["webcam"], + "Logitech_C615": ["webcam"], + "Logitech_C270": ["webcam"], + "Logitech_StreamCam": ["webcam"], # (robot) arm "Rotrix_Dexarm": ["arm"], diff --git a/cli/commands/power/battery.py b/cli/commands/power/battery.py index 20900707..6c580353 100644 --- a/cli/commands/power/battery.py +++ b/cli/commands/power/battery.py @@ -368,6 +368,16 @@ def model(ctx, box, partnumber): _run_backend(ctx, resolved_box, 'set_model', netname=netname, partnumber=partnumber) +@battery.command() +@click.pass_context +@click.option("--box", required=False, help="Lagerbox name or IP") +def models(ctx, box): + """List battery models saved on the instrument""" + resolved_box = resolve_box(ctx, box) + netname = require_netname(ctx, "battery") + _run_backend(ctx, resolved_box, 'list_models', netname=netname) + + @battery.command() @click.pass_context @click.option("--box", required=False, help="Lagerbox name or IP") diff --git a/cli/impl/query_instruments.py b/cli/impl/query_instruments.py index 3eca7816..3521d468 100644 --- a/cli/impl/query_instruments.py +++ b/cli/impl/query_instruments.py @@ -124,10 +124,20 @@ def wrapper(*args, **kwargs) -> T: # eload "Rigol_DL3021": {"vid": "1ab1", "pid": "0e11", "net_type": ["eload"]}, # Same PID as DP821, differentiated by serial - # camera + # camera — detection is catalog-driven: any entry with a ``webcam`` + # net_type is picked up by ``_by_camera`` below, so adding a camera + # is a one-line addition here (mirror it in box/lager/http_handlers/usb_scanner.py). "Logitech_BRIO_HD": {"vid": "046d", "pid": "085e", "net_type": ["webcam"]}, "Logitech_BRIO": {"vid": "046d", "pid": "0856", "net_type": ["webcam"]}, + "Logitech_BRIO_4K_Stream": {"vid": "046d", "pid": "086b", "net_type": ["webcam"]}, + "Logitech_4K_Pro": {"vid": "046d", "pid": "087f", "net_type": ["webcam"]}, "Logitech_C930e": {"vid": "046d", "pid": "0843", "net_type": ["webcam"]}, + "Logitech_C925e": {"vid": "046d", "pid": "085b", "net_type": ["webcam"]}, + "Logitech_C922_Pro": {"vid": "046d", "pid": "085c", "net_type": ["webcam"]}, + "Logitech_C920": {"vid": "046d", "pid": "082d", "net_type": ["webcam"]}, + "Logitech_C615": {"vid": "046d", "pid": "082c", "net_type": ["webcam"]}, + "Logitech_C270": {"vid": "046d", "pid": "0825", "net_type": ["webcam"]}, + "Logitech_StreamCam": {"vid": "046d", "pid": "0893", "net_type": ["webcam"]}, # watt-meter "Yocto_Watt": {"vid": "24e0", "pid": "002a", "net_type": ["watt-meter"]}, @@ -354,46 +364,61 @@ def _by_handshake(*, exclude: Optional[set[str]] = None) -> List[dict]: # Camera detection # --------------------------------------------------------------------------- -CAM_VID = "046d" # Logitech -CAM_PIDS = {"085e", "0856", "0843"} # BRIO HD / BRIO / C930e +_WEBCAM_VIDPIDS = { + (meta["vid"].lower(), meta["pid"].lower()) + for meta in SUPPORTED_USB.values() + if "webcam" in meta["net_type"] +} -def _by_camera() -> List[dict]: - """ - Detect Logitech webcams (BRIO, BRIO HD, C930e) when udev attributes are missing. - """ - usb_cams = [ - dev for dev in _scan_usb() - if dev["vid"] == CAM_VID and dev["pid"] in CAM_PIDS - ] +def _read_sysfs_attr(path: Path) -> Optional[str]: + try: + return path.read_text().strip() + except OSError: + return None - if not usb_cams: - return [] +def _by_camera(v4l_root: Path = Path("/sys/class/video4linux")) -> List[dict]: + """Detect supported webcams and their /dev/video capture nodes. + Each ``/sys/class/video4linux/videoN`` links to the USB interface that + owns it; the interface's parent directory is the USB device carrying + idVendor/idProduct/serial. Walking that link keeps the node→camera + mapping correct when models expose different numbers of video nodes + (a C920 exposes two, a BRIO four), which an index-based heuristic + cannot get right on mixed setups. + """ + results: List[dict] = [] + seen_devices: set = set() video_nodes = sorted( - (Path(p) for p in glob.glob("/dev/video*")), + v4l_root.glob("video*"), key=lambda p: int(p.name.replace("video", "")) ) - - results: List[dict] = [] - blk_size = 4 - for idx, cam in enumerate(usb_cams): + for node in video_nodes: try: - vidx = idx * blk_size - video_dev = str(video_nodes[vidx]) - except IndexError: - print( - f"[webcam] WARNING: could not map video port for " - f"serial {cam.get('serial') or '(no-serial)'}", - file=sys.stderr - ) + usb_dev = (node / "device").resolve().parent + except OSError: continue - - entry = { - **cam, - "channels": {"webcam": [video_dev]}, - } - results.append(entry) - + vid = _read_sysfs_attr(usb_dev / "idVendor") + pid = _read_sysfs_attr(usb_dev / "idProduct") + if not vid or not pid: + continue + vid, pid = vid.lower(), pid.lower() + if (vid, pid) not in _WEBCAM_VIDPIDS: + continue + if str(usb_dev) in seen_devices: + # Higher-numbered nodes on the same camera are metadata/IR + # streams; the first node is the capture stream. + continue + seen_devices.add(str(usb_dev)) + serial = _read_sysfs_attr(usb_dev / "serial") + results.append({ + "name": _VIDPID_TO_NAME[(vid, pid)], + "vid": vid, + "pid": pid, + "serial": serial, + "address": f"USB0::0x{vid.upper()}::0x{pid.upper()}::{serial or ''}::INSTR", + "net_type": ["webcam"], + "channels": {"webcam": [f"/dev/{node.name}"]}, + }) return results def _merge_or_append(entry: dict, instruments: List[dict]) -> None: diff --git a/cli/tests/test_box_lager_imports.py b/cli/tests/test_box_lager_imports.py index 1e60ee08..71c3697b 100644 --- a/cli/tests/test_box_lager_imports.py +++ b/cli/tests/test_box_lager_imports.py @@ -190,7 +190,7 @@ def test_power_module(): battery_new_exports = [ "set_mode", "set_to_battery_mode", "set", "set_soc", "set_voc", "set_volt_full", "set_volt_empty", "set_capacity", "set_current_limit", - "set_ovp", "set_ocp", "set_model", "enable_battery", "disable_battery", + "set_ovp", "set_ocp", "set_model", "list_models", "enable_battery", "disable_battery", "print_state", "clear", "clear_ovp", "clear_ocp", "terminal_voltage", "current", "esr", "BatteryNet", "BatteryBackendError", "KeithleyBattery", "Keithley", "create_device", diff --git a/docs/docs.json b/docs/docs.json index f99a4c28..5e79e527 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -211,6 +211,7 @@ { "group": "Version History", "pages": [ + "source/release-notes/v0.31.10", "source/release-notes/v0.31.9", "source/release-notes/v0.31.8", "source/release-notes/v0.31.7", diff --git a/docs/source/reference/cli/battery.mdx b/docs/source/reference/cli/battery.mdx index 38a7e123..45fa0fb2 100644 --- a/docs/source/reference/cli/battery.mdx +++ b/docs/source/reference/cli/battery.mdx @@ -33,6 +33,7 @@ lager battery [OPTIONS] [NETNAME] COMMAND [ARGS]... | `ovp` | Set or read over-voltage protection (V) | | `ocp` | Set or read over-current protection (A) | | `model` | Set or read battery model | +| `models` | List battery models saved on the instrument | | `state` | Get comprehensive battery state | | `enable` | Enable battery simulator output | | `disable` | Disable battery simulator output | @@ -236,6 +237,23 @@ lager battery NETNAME model [PARTNUMBER] [OPTIONS] - `lead-acid` - Lead acid battery - Custom part numbers from your battery library +### `models` + +List the battery models available on the instrument: memory slots with a +saved model plus the firmware built-in models. Read-only. + +```bash +lager battery NETNAME models [OPTIONS] +``` + +**Arguments:** +- `NETNAME` - Name of the battery Net + +**Options:** +- `--box TEXT` - Lagerbox name or IP + +The printed slots and names are valid inputs to the `model` command. + ### `state` Get comprehensive battery state including all current settings and measurements. diff --git a/docs/source/release-notes/v0.31.10.mdx b/docs/source/release-notes/v0.31.10.mdx new file mode 100644 index 00000000..62d67a20 --- /dev/null +++ b/docs/source/release-notes/v0.31.10.mdx @@ -0,0 +1,33 @@ +--- +title: "Version 0.31.10" +description: "July 13, 2026" +--- + +## Features + +- **8 more Logitech webcams supported.** The Logi 4K Pro, BRIO 4K Stream, C925e, C922 Pro, C920, C615, C270, and StreamCam now appear in `lager instruments` and can be added as webcam nets, joining the existing BRIO, BRIO HD, and C930e. Detection is catalog-driven — adding a future model is a one-line table entry — and each camera is mapped to its actual `/dev/video` capture node by walking sysfs, so setups mixing cameras with different node counts (a C920 exposes two, a BRIO four) resolve to the right device. +- **`lager battery models` — list the battery models saved on the instrument.** Battery simulators store models in numbered memory slots, but until now there was no way to see which slots actually held a model without walking to the front panel. The new read-only `models` command (also available in the battery TUI and the box's `/battery/command` HTTP endpoint as `list_models`) prints each occupied slot plus the firmware's built-in models — all valid inputs to the existing `model` command. On the Keithley 2281S the catalog is assembled from query-only slot probes, so listing never changes instrument state. + +## Bug Fixes + +- **Battery model readback now reports the actual loaded model.** The driver previously read the "current model" with `:BATT:STAT?`, which per the 2281S reference manual reports charge/discharge status — not the model — so `lager battery model`, `state`, and the TUI header showed "DISCHARGE" whenever the output was idle, regardless of what was loaded. The same misread made `model ` raise a false "slot is empty" error after every successful load. Readback and verification now use `:BATT:MOD:RCL?` (e.g. `Model: slot 5` or `Model: LI_ION4_2`), and because the 2281S fails empty-slot recalls silently, a recall that doesn't take effect is now reliably detected and reported with guidance instead of pretending to succeed. +- **Built-in battery models are now loadable over SCPI.** The 2281S manual prints two built-in model names with hyphens (`LI-ION4_2`, `LEAD-ACID12`), but the instrument's SCPI parser rejects hyphens outright (error -102), so recalling those built-ins through Lager never worked. The driver now sends the underscore spellings the firmware actually accepts (`LI_ION4_2`, `LEAD_ACID12`) and accepts either spelling as input. +- **Battery command errors no longer masquerade as "Resource busy".** When a battery driver raised a real error (like the empty-slot guidance above), the box's `/battery/command` endpoint returned it as a 5xx, which made the CLI treat the endpoint as unavailable and fall through to its legacy direct-USB path — always failing with `[Errno 16] Resource busy` and burying the actual message. Driver errors are now returned as ordinary command failures, so the CLI and TUI display the real diagnosis. + +## Installation + +To install this version: + +```bash +pip install lager-cli==0.31.10 +``` + +To upgrade from a previous version: + +```bash +pip install --upgrade lager-cli +``` + +## Resources + +[View Release on PyPI](https://pypi.org/project/lager-cli/0.31.10/) diff --git a/docs/source/supported-instruments/supported-instruments.mdx b/docs/source/supported-instruments/supported-instruments.mdx index 5761c1aa..0202fd5f 100644 --- a/docs/source/supported-instruments/supported-instruments.mdx +++ b/docs/source/supported-instruments/supported-instruments.mdx @@ -273,7 +273,15 @@ scripts and CI pipelines. |---|---|---| | Logitech | BRIO HD | `lager webcam` | | Logitech | BRIO | `lager webcam` | +| Logitech | BRIO 4K Stream Edition | `lager webcam` | +| Logitech | 4K Pro (Logi 4K Pro) | `lager webcam` | | Logitech | C930e | `lager webcam` | +| Logitech | C925e | `lager webcam` | +| Logitech | C922 Pro Stream | `lager webcam` | +| Logitech | C920 | `lager webcam` | +| Logitech | C615 | `lager webcam` | +| Logitech | C270 | `lager webcam` | +| Logitech | StreamCam | `lager webcam` | **Enables:** - Visual DUT inspection diff --git a/test/unit/box/test_battery_model_catalog.py b/test/unit/box/test_battery_model_catalog.py new file mode 100644 index 00000000..e246bd46 --- /dev/null +++ b/test/unit/box/test_battery_model_catalog.py @@ -0,0 +1,387 @@ +# Copyright 2024-2026 Lager Data +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the read-only battery model catalog. + +The 2281S has no :BATT:MODel:CATalog? query (reference manual 077114601, +March 2019 — the word "catalog" appears nowhere in it), so +``KeithleyBattery.model_catalog`` assembles the catalog from the documented +query-only :BATT:MOD:VOC:STEPs? probes plus the five firmware built-in +models. These tests cover the probe parsing (occupied/empty slots), the +unsupported-firmware rejection path, and the ``list_models`` dispatcher +action that returns the structured payload to /battery/command callers. + +Modules are spec-loaded with stubbed ``lager.*`` parents so the tests run +in the CLI test environment without box-only dependencies (pyvisa etc.), +following the pattern in test_monitor_state.py. +""" + +from __future__ import annotations + +import importlib.util +import os +import sys +import types + +import pytest + +_REPO_ROOT = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "..", "..") +) + + +def _load_module(name, relpath, stubs=None): + """Spec-load a box module with stubbed dependency modules.""" + for stub_name, stub in (stubs or {}).items(): + sys.modules.setdefault(stub_name, stub) + path = os.path.join(_REPO_ROOT, *relpath.split("/")) + spec = importlib.util.spec_from_file_location(name, path) + mod = importlib.util.module_from_spec(spec) + sys.modules[name] = mod + spec.loader.exec_module(mod) + return mod + + +def _lager_exceptions_stub(): + mod = types.ModuleType("lager.exceptions") + for cls in ("SupplyBackendError", "LibraryMissingError", "DeviceNotFoundError", + "BatteryBackendError"): + setattr(mod, cls, type(cls, (Exception,), {})) + lager_pkg = types.ModuleType("lager") + lager_pkg.exceptions = mod + return {"lager": lager_pkg, "lager.exceptions": mod} + + +def _instrument_wrapper_stubs(): + wrap = types.ModuleType("lager.instrument_wrappers.instrument_wrap") + wrap.InstrumentWrapKeithley = object + defines = types.ModuleType("lager.instrument_wrappers.keithley_defines") + defines.Mode = types.SimpleNamespace( + BatterySimulator=types.SimpleNamespace(to_cmd=lambda: "BATT")) + defines.SimMethod = object + util = types.ModuleType("lager.instrument_wrappers.util") + util.InvalidEnumError = type("InvalidEnumError", (Exception,), {}) + pkg = types.ModuleType("lager.instrument_wrappers") + return { + "lager.instrument_wrappers": pkg, + "lager.instrument_wrappers.instrument_wrap": wrap, + "lager.instrument_wrappers.keithley_defines": defines, + "lager.instrument_wrappers.util": util, + } + + +@pytest.fixture(scope="module") +def keithley_mod(): + stubs = _lager_exceptions_stub() + stubs.update(_instrument_wrapper_stubs()) + for stub_name, stub in stubs.items(): + sys.modules.setdefault(stub_name, stub) + + pkg_name = "kbc_pkg" + package = types.ModuleType(pkg_name) + package.__path__ = [os.path.join(_REPO_ROOT, "box/lager/power/battery")] + sys.modules[pkg_name] = package + bn = _load_module(pkg_name + ".battery_net", "box/lager/power/battery/battery_net.py") + sys.modules[pkg_name + ".battery_net"] = bn + + path = os.path.join(_REPO_ROOT, "box/lager/power/battery/keithley.py") + spec = importlib.util.spec_from_file_location(pkg_name + ".keithley", path) + mod = importlib.util.module_from_spec(spec) + sys.modules[pkg_name + ".keithley"] = mod + spec.loader.exec_module(mod) + return mod + + +@pytest.fixture(scope="module") +def dispatcher_mod(keithley_mod): + stubs = _lager_exceptions_stub() + base = types.ModuleType("lager.dispatchers.base") + + class BaseDispatcher: + pass + + base.BaseDispatcher = BaseDispatcher + dispatchers = types.ModuleType("lager.dispatchers") + stubs.update({ + "lager.dispatchers": dispatchers, + "lager.dispatchers.base": base, + }) + for stub_name, stub in stubs.items(): + sys.modules.setdefault(stub_name, stub) + + # dispatcher.py does `from .battery_net import ...` / `from .keithley + # import ...`; reuse the modules already loaded by the keithley fixture. + pkg_name = "kbc_pkg" + path = os.path.join(_REPO_ROOT, "box/lager/power/battery/dispatcher.py") + spec = importlib.util.spec_from_file_location(pkg_name + ".dispatcher", path) + mod = importlib.util.module_from_spec(spec) + sys.modules[pkg_name + ".dispatcher"] = mod + spec.loader.exec_module(mod) + return mod + + +def _driver_with_replies(keithley_mod, replies=None, error=None): + """KeithleyBattery with a fake checked-query transport. + + ``replies`` maps SCPI command -> response string; unmapped probes answer + "0" (empty slot). ``error`` maps command -> exception to raise instead. + Writes are forbidden: the catalog must be read-only. + """ + drv = object.__new__(keithley_mod.KeithleyBattery) + drv.commands = [] + + def query(cmd): + drv.commands.append(cmd) + if error and cmd in error: + raise error[cmd] + return (replies or {}).get(cmd, "0") + + drv._query = query + + def forbidden(*args, **kwargs): + raise AssertionError("model_catalog must not write to the instrument") + + drv._write = forbidden + drv._tolerant_write = forbidden + drv._set_with_output_management = forbidden + return drv + + +class TestKeithleyModelCatalog: + def test_normal_catalog_lists_occupied_slots_and_builtins(self, keithley_mod): + drv = _driver_with_replies(keithley_mod, replies={ + ":BATT:MOD1:VOC:STEP?": "101", + ":BATT:MOD3:VOC:STEP?": "101", + }) + catalog = drv.model_catalog() + assert catalog[0] == {"slot": 0, "name": "DISCHARGE"} + assert {"slot": 1, "name": None} in catalog + assert {"slot": 3, "name": None} in catalog + # Unoccupied slots are omitted. + assert not any(entry["slot"] == 2 for entry in catalog) + # The five firmware built-ins are always listed, slot-less. + builtins = [entry["name"] for entry in catalog if entry["slot"] is None] + assert builtins == list(keithley_mod.BUILTIN_MODELS) + # All nine slots were probed, read-only. + assert drv.commands == [f":BATT:MOD{i}:VOC:STEP?" for i in range(1, 10)] + + def test_empty_slots_leave_discharge_and_builtins_only(self, keithley_mod): + drv = _driver_with_replies(keithley_mod) + catalog = drv.model_catalog() + assert catalog[0] == {"slot": 0, "name": "DISCHARGE"} + assert [e for e in catalog if e["slot"] not in (0, None)] == [] + assert len(catalog) == 1 + len(keithley_mod.BUILTIN_MODELS) + + def test_odd_step_replies_treated_as_empty(self, keithley_mod): + drv = _driver_with_replies(keithley_mod, replies={ + ":BATT:MOD1:VOC:STEP?": "garbage", + ":BATT:MOD2:VOC:STEP?": "", + ":BATT:MOD3:VOC:STEP?": "+1.010000E+02", # sci-notation counts + }) + catalog = drv.model_catalog() + slots = [e["slot"] for e in catalog if e["slot"] not in (0, None)] + assert slots == [3] + + def test_per_slot_instrument_error_means_slot_is_skipped(self, keithley_mod): + # e.g. an execution error against one slot must not abort the sweep. + drv = _driver_with_replies( + keithley_mod, + replies={":BATT:MOD2:VOC:STEP?": "101"}, + error={":BATT:MOD1:VOC:STEP?": Exception(704, "Not permitted", "")}, + ) + catalog = drv.model_catalog() + slots = [e["slot"] for e in catalog if e["slot"] not in (0, None)] + assert slots == [2] + + def test_undefined_header_raises_unsupported_firmware(self, keithley_mod): + # InstrumentError style: (code, message, response) in args. + err = Exception(-113, "Undefined header", "") + drv = _driver_with_replies( + keithley_mod, error={":BATT:MOD1:VOC:STEP?": err}) + with pytest.raises(keithley_mod.BatteryBackendError) as excinfo: + drv.model_catalog() + assert "does not support" in str(excinfo.value) + # Bailed on the first probe instead of sweeping the rest. + assert drv.commands == [":BATT:MOD1:VOC:STEP?"] + + def test_string_coded_error_also_detected(self, keithley_mod): + # Some wrappers stringify SCPI errors as ',""'. + err = keithley_mod.BatteryBackendError('-113,"Undefined header"') + drv = _driver_with_replies( + keithley_mod, error={":BATT:MOD1:VOC:STEP?": err}) + with pytest.raises(keithley_mod.BatteryBackendError) as excinfo: + drv.model_catalog() + assert "does not support" in str(excinfo.value) + + def test_no_response_fails_fast(self, keithley_mod): + # A codeless transport failure (e.g. VISA timeout on old firmware) + # must raise on the first probe, not time out nine times. + drv = _driver_with_replies( + keithley_mod, error={":BATT:MOD1:VOC:STEP?": Exception("timeout")}) + with pytest.raises(keithley_mod.BatteryBackendError) as excinfo: + drv.model_catalog() + assert "catalog query failed" in str(excinfo.value) + assert drv.commands == [":BATT:MOD1:VOC:STEP?"] + + +class TestCurrentModel: + """current_model() reads :BATT:MOD:RCL? — hardware-verified: it answers + with the slot number while a numbered slot is active and never replies + while a firmware built-in is active. (:BATT:STAT?, which older code used, + reports charge/discharge status, not the model.)""" + + def _drv(self, keithley_mod, rcl_reply): + drv = object.__new__(keithley_mod.KeithleyBattery) + + class FakeInstr: + timeout = 5000 + + drv.instr = FakeInstr() + drv._safe_query = ( + lambda cmd, default="": rcl_reply if cmd == ":BATT:MOD:RCL?" else default) + return drv + + def test_numbered_slot(self, keithley_mod): + drv = self._drv(keithley_mod, "5") + assert drv.current_model() == "slot 5" + # The shortened query timeout must be restored afterwards. + assert drv.instr.timeout == 5000 + + def test_slot_zero_reads_as_discharge(self, keithley_mod): + assert self._drv(keithley_mod, "0").current_model() == "DISCHARGE" + + def test_name_reply_passes_through(self, keithley_mod): + # Future firmware that answers with a name should just work. + assert self._drv(keithley_mod, '"LI-ION4_2"').current_model() == "LI-ION4_2" + + def test_silence_falls_back_to_cached_name(self, keithley_mod): + drv = self._drv(keithley_mod, "") + drv._active_model_name = "LI-ION4_2" + assert drv.current_model() == "LI-ION4_2" + + def test_silence_without_cache_reads_custom(self, keithley_mod): + assert self._drv(keithley_mod, "").current_model() == "Custom" + + +class TestSetModelVerification: + """set_model verifies through :BATT:MOD:RCL? because recalling an empty + slot fails silently (hardware-verified: no error queued, previous model + stays active). For built-ins, RCL? silence is the success signature.""" + + def _drv(self, keithley_mod, rcl_reply): + drv = object.__new__(keithley_mod.KeithleyBattery) + drv.writes = [] + drv._set_with_output_management = ( + lambda cmd, ignore_codes=(): drv.writes.append(cmd)) + drv._active_model_raw = lambda: rcl_reply + return drv + + def test_numeric_slot_success(self, keithley_mod): + drv = self._drv(keithley_mod, "5") + drv.set_model(5) + assert drv.writes == [":BATT:MOD:RCL 5"] + assert drv._active_model_name == "slot 5" + + def test_alias_maps_to_slot(self, keithley_mod): + drv = self._drv(keithley_mod, "1") + drv.set_model("liion") + assert drv.writes == [":BATT:MOD:RCL 1"] + assert drv._active_model_name == "slot 1" + + def test_discharge_slot_zero(self, keithley_mod): + drv = self._drv(keithley_mod, "0") + drv.set_model("discharge") + assert drv.writes == [":BATT:MOD:RCL 0"] + assert drv._active_model_name == "DISCHARGE" + + def test_empty_slot_detected_by_unchanged_readback(self, keithley_mod): + # Recall of empty slot 7 fails silently; RCL? still reports slot 5. + drv = self._drv(keithley_mod, "5") + with pytest.raises(keithley_mod.BatteryBackendError) as excinfo: + drv.set_model(7) + msg = str(excinfo.value) + assert "appears to be empty" in msg + assert "active model: 5" in msg + assert "'models'" in msg # points at the new catalog command + + def test_empty_slot_with_builtin_active_gives_no_reply(self, keithley_mod): + # Previous model was a built-in, so RCL? is silent both before and + # after the failed recall — still a verification failure. + drv = self._drv(keithley_mod, "") + with pytest.raises(keithley_mod.BatteryBackendError) as excinfo: + drv.set_model(7) + assert "unchanged (no reply)" in str(excinfo.value) + + def test_builtin_success_echoes_name(self, keithley_mod): + # RCL? echoes the built-in's name on success (hardware-verified). + drv = self._drv(keithley_mod, "LI_ION4_2") + drv.set_model("LI_ION4_2") + assert drv.writes == [":BATT:MOD:RCL LI_ION4_2"] + assert drv._active_model_name == "LI_ION4_2" + + def test_builtin_hyphen_input_sends_underscore(self, keithley_mod): + # The manual prints LI-ION4_2, but a hyphen is a SCPI syntax error + # (-102, hardware-verified) — accept it and send the underscore form. + drv = self._drv(keithley_mod, "LEAD_ACID12") + drv.set_model("lead-acid12") + assert drv.writes == [":BATT:MOD:RCL LEAD_ACID12"] + assert drv._active_model_name == "LEAD_ACID12" + + def test_builtin_failure_still_reports_previous_slot(self, keithley_mod): + drv = self._drv(keithley_mod, "5") + with pytest.raises(keithley_mod.BatteryBackendError) as excinfo: + drv.set_model("LI_ION4_2") + assert "model slot 5" in str(excinfo.value) + + +class TestListModelsAction: + def _fake_resolver(self, dispatcher_mod, catalog): + class FakeDriver: + def model_catalog(self): + return catalog + + dispatcher_mod._dispatcher._resolve_net_and_driver = ( + lambda netname: (FakeDriver(), 1)) + + def test_returns_structured_payload(self, dispatcher_mod, capsys): + catalog = [ + {"slot": 0, "name": "DISCHARGE"}, + {"slot": 1, "name": None}, + {"slot": None, "name": "LI-ION4_2"}, + ] + self._fake_resolver(dispatcher_mod, catalog) + result = dispatcher_mod.list_models("batt1") + # HTTP callers of /battery/command receive the catalog in the + # response payload, like the 'state' action's structured dict. + assert result == {"models": catalog} + out = capsys.readouterr().out + assert "DISCHARGE" in out + assert "(custom model)" in out + assert "LI-ION4_2 (built-in)" in out + assert "valid inputs to the 'model' command" in out + + def test_backend_error_propagates(self, dispatcher_mod): + class FailingDriver: + def model_catalog(self): + raise dispatcher_mod.BatteryBackendError("no catalog support") + + dispatcher_mod._dispatcher._resolve_net_and_driver = ( + lambda netname: (FailingDriver(), 1)) + with pytest.raises(dispatcher_mod.BatteryBackendError): + dispatcher_mod.list_models("batt1") + + +class TestFormatModelCatalog: + def test_table_shape(self, dispatcher_mod): + text = dispatcher_mod.format_model_catalog([ + {"slot": 0, "name": "DISCHARGE"}, + {"slot": 4, "name": None}, + {"slot": None, "name": "NIMH12"}, + ]) + lines = text.splitlines() + assert lines[0] == "Slot Model" + assert lines[1].startswith("0") and "DISCHARGE" in lines[1] + assert lines[2].startswith("4") and "(custom model)" in lines[2] + assert lines[3].startswith("-") and "NIMH12 (built-in)" in lines[3] + assert lines[-1] == ( + "Slots and names above are valid inputs to the 'model' command.") diff --git a/test/unit/box/test_webcam_detection.py b/test/unit/box/test_webcam_detection.py new file mode 100644 index 00000000..cb79634b --- /dev/null +++ b/test/unit/box/test_webcam_detection.py @@ -0,0 +1,167 @@ +# Copyright 2024-2026 Lager Data +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for sysfs-based webcam detection (``_by_camera``). + +Covers both copies of the scanner — ``cli/impl/query_instruments.py`` and +``box/lager/http_handlers/usb_scanner.py`` — against a fake sysfs tree: + +* the /dev/videoN node is resolved through the videoN → USB-interface → + USB-device sysfs chain, not by index arithmetic, so cameras exposing + different node counts (C920: two, BRIO: four) map correctly side-by-side; +* only the first (capture) node per camera is reported; +* the webcam VID:PID set is derived from SUPPORTED_USB, so every catalog + entry with a ``webcam`` net_type — including the Logi 4K Pro (046d:087f) — + is detected without touching the detection code; +* both catalogs advertise the same webcam set. + +Fully hermetic: ``serial`` (pyserial) is stubbed before the CLI script loads, +and ``_by_camera`` reads from a temp-dir sysfs stand-in via its ``v4l_root`` +parameter. +""" + +import importlib.util +import os +import sys +import tempfile +import types +import unittest +from pathlib import Path + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) +CLI_IMPL_PATH = os.path.join(REPO_ROOT, "cli", "impl", "query_instruments.py") +BOX_SCANNER_PATH = os.path.join(REPO_ROOT, "box", "lager", "http_handlers", "usb_scanner.py") + +# Stub pyserial before loading the CLI script (top-level ``from serial import``). +if "serial" not in sys.modules: + _serial_stub = types.ModuleType("serial") + _serial_stub.Serial = object + _serial_stub.SerialException = Exception + sys.modules["serial"] = _serial_stub + + +def _load_module(dotted, filepath): + if dotted in sys.modules: + return sys.modules[dotted] + spec = importlib.util.spec_from_file_location(dotted, filepath) + mod = importlib.util.module_from_spec(spec) + sys.modules[dotted] = mod + spec.loader.exec_module(mod) + return mod + + +qi = _load_module("qi_webcam_test", CLI_IMPL_PATH) +box_scanner = _load_module("box_usb_scanner_webcam_test", BOX_SCANNER_PATH) + + +class FakeSysfs: + """Builds a /sys stand-in: USB device dirs plus a video4linux dir.""" + + def __init__(self, root: Path): + self.root = root + self.usb = root / "usb" + self.v4l = root / "v4l" + self.usb.mkdir() + self.v4l.mkdir() + self._next_node = 0 + + def add_camera(self, busport, vid, pid, serial=None, nodes=1): + """Add a USB camera exposing *nodes* consecutive /dev/videoN nodes.""" + dev = self.usb / busport + iface = dev / f"{busport}:1.0" + iface.mkdir(parents=True) + (dev / "idVendor").write_text(f"{vid}\n") + (dev / "idProduct").write_text(f"{pid}\n") + if serial is not None: + (dev / "serial").write_text(f"{serial}\n") + first = self._next_node + for _ in range(nodes): + node = self.v4l / f"video{self._next_node}" + node.mkdir() + (node / "device").symlink_to(iface) + self._next_node += 1 + return first + + +class ByCameraTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.sysfs = FakeSysfs(Path(self._tmp.name)) + + def _run(self, mod): + return mod._by_camera(v4l_root=self.sysfs.v4l) + + def test_4k_pro_detected(self): + self.sysfs.add_camera("2-3", "046d", "087f", serial="ABC123", nodes=4) + for mod in (qi, box_scanner): + cams = self._run(mod) + self.assertEqual(len(cams), 1) + cam = cams[0] + self.assertEqual(cam["name"], "Logitech_4K_Pro") + self.assertEqual(cam["vid"], "046d") + self.assertEqual(cam["pid"], "087f") + self.assertEqual(cam["serial"], "ABC123") + self.assertEqual(cam["net_type"], ["webcam"]) + self.assertEqual(cam["channels"], {"webcam": ["/dev/video0"]}) + self.assertEqual(cam["address"], "USB0::0x046D::0x087F::ABC123::INSTR") + + def test_mixed_node_counts_map_correctly(self): + # A C920 exposes two nodes, a BRIO four. The old idx*4 heuristic + # would have pointed the BRIO at /dev/video4 (nonexistent here). + self.sysfs.add_camera("1-1", "046d", "082d", serial="C920SER", nodes=2) + self.sysfs.add_camera("1-2", "046d", "085e", serial="BRIOSER", nodes=4) + for mod in (qi, box_scanner): + cams = self._run(mod) + by_name = {c["name"]: c for c in cams} + self.assertEqual( + set(by_name), {"Logitech_C920", "Logitech_BRIO_HD"}) + self.assertEqual( + by_name["Logitech_C920"]["channels"], {"webcam": ["/dev/video0"]}) + self.assertEqual( + by_name["Logitech_BRIO_HD"]["channels"], {"webcam": ["/dev/video2"]}) + + def test_unknown_video_device_ignored(self): + # A capture card / unknown camera should not be reported. + self.sysfs.add_camera("3-1", "dead", "beef", nodes=2) + for mod in (qi, box_scanner): + self.assertEqual(self._run(mod), []) + + def test_no_serial_camera(self): + self.sysfs.add_camera("4-1", "046d", "0825", nodes=2) # C270, no serial + for mod in (qi, box_scanner): + cams = self._run(mod) + self.assertEqual(len(cams), 1) + self.assertIsNone(cams[0]["serial"]) + self.assertEqual( + cams[0]["address"], "USB0::0x046D::0x0825::::INSTR") + + def test_missing_v4l_root(self): + for mod in (qi, box_scanner): + self.assertEqual( + mod._by_camera(v4l_root=self.sysfs.root / "nope"), []) + + def test_every_catalog_webcam_is_detectable(self): + for mod in (qi, box_scanner): + webcams = { + name: meta for name, meta in mod.SUPPORTED_USB.items() + if "webcam" in meta["net_type"] + } + self.assertIn("Logitech_4K_Pro", webcams) + self.assertEqual( + {(m["vid"], m["pid"]) for m in webcams.values()}, + mod._WEBCAM_VIDPIDS, + ) + + def test_catalogs_advertise_same_webcams(self): + def webcam_entries(mod): + return { + name: (meta["vid"], meta["pid"]) + for name, meta in mod.SUPPORTED_USB.items() + if "webcam" in meta["net_type"] + } + self.assertEqual(webcam_entries(qi), webcam_entries(box_scanner)) + + +if __name__ == "__main__": + unittest.main()