From a7d432579bfa46b25791de79d912dbbbaa60b6f3 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Mar 2026 23:06:44 +0000 Subject: [PATCH 1/2] Initial plan From b5882ef285461f262290ef3236d89f8b276ae1bb Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Mar 2026 23:18:22 +0000 Subject: [PATCH 2/2] feat: implement standalone desktop GUI for wireless modem RF parameter management Co-authored-by: deilert00 <139483703+deilert00@users.noreply.github.com> --- .gitignore | 5 + desktop/README.md | 145 +++++++ desktop/modem_manager.py | 637 ++++++++++++++++++++++++++++ desktop/modem_utils.py | 369 ++++++++++++++++ desktop/requirements.txt | 10 + desktop/tests/test_modem_manager.py | 372 ++++++++++++++++ 6 files changed, 1538 insertions(+) create mode 100644 desktop/README.md create mode 100644 desktop/modem_manager.py create mode 100644 desktop/modem_utils.py create mode 100644 desktop/requirements.txt create mode 100644 desktop/tests/test_modem_manager.py diff --git a/.gitignore b/.gitignore index 60c554b..ac8295f 100644 --- a/.gitignore +++ b/.gitignore @@ -47,3 +47,8 @@ GoogleService-Info.plist # E2E test reports (generated artifacts) reports/ +__pycache__ +desktop/__pycache__/ +desktop/tests/__pycache__/ +desktop/dist/ +desktop/build/ diff --git a/desktop/README.md b/desktop/README.md new file mode 100644 index 0000000..59a26b1 --- /dev/null +++ b/desktop/README.md @@ -0,0 +1,145 @@ +# Modem RF Parameter Manager — Desktop Application + +A standalone, Windows-friendly desktop GUI for monitoring and managing wireless +modem RF parameters via HTTP. + +--- + +## Requirements + +| Requirement | Version | +|---|---| +| Python | 3.8 or later | +| tkinter | Included with Python (no extra install) | +| Operating System | Windows 10/11, macOS, Linux | + +No third-party packages are required — the application uses only Python's +standard library (`tkinter`, `urllib`, `xml.etree.ElementTree`, `json`, +`threading`, `queue`). + +--- + +## Quick Start + +```bash +# Clone the repo (if you haven't already) +git clone https://github.com/deilert00/Goald.git +cd Goald/desktop + +# Run the application +python modem_manager.py +``` + +On Windows you can also double-click `modem_manager.py` if Python is +associated with `.py` files, or run it from PowerShell: + +```powershell +python .\modem_manager.py +``` + +--- + +## Supported API Types + +### Huawei HiLink (default) + +Most Huawei USB modems (E3372, E3531, …) and home routers (B525, B535, …) +expose the HiLink REST API. The application sends `GET` requests to: + +| Data | Endpoint | +|---|---| +| Signal quality | `/api/device/signal` | +| Device info | `/api/device/information` | +| Band lock | `/api/net/net-mode` (POST) | + +**Default modem URL:** `http://192.168.8.1` + +### Generic JSON + +If your modem or gateway exposes an HTTP endpoint that returns a JSON object +with signal fields, select **Generic JSON** from the API drop-down and set the +URL to your modem's base address. The application will fetch +`/signal` and attempt to map common field names automatically. + +Supported field name variants: + +| Parameter | Recognised keys | +|---|---| +| RSSI | `rssi`, `RSSI`, `signal_strength` | +| RSRP | `rsrp`, `RSRP` | +| RSRQ | `rsrq`, `RSRQ` | +| SINR | `sinr`, `SINR`, `snr`, `SNR` | +| Band | `band`, `Band`, `lte_band`, `cell_band` | +| Channel | `earfcn`, `EARFCN`, `channel`, `arfcn` | + +--- + +## Features + +| Feature | Description | +|---|---| +| **Live signal monitoring** | Polls the modem at a configurable interval (2 s – 60 s) | +| **Signal quality colouring** | RSRP and SINR values are colour-coded (Excellent / Good / Fair / Poor) | +| **Band & Frequency view** | Shows active band, EARFCN, bandwidth, Cell ID and PCI | +| **Band locking** | Apply an LTE band lock by entering comma-separated band numbers (HiLink only) | +| **History table** | Keeps a scrollable log of all readings during the session | +| **JSON export** | Export the full history to a JSON file | +| **Raw response view** | Shows the last parsed response in JSON format with copy-to-clipboard | +| **Windows DPI awareness** | Calls `SetProcessDpiAwareness(1)` on Windows for crisp rendering | + +--- + +## Signal Quality Reference + +### RSRP thresholds + +| Range (dBm) | Quality | +|---|---| +| ≥ −80 | Excellent | +| −80 to −90 | Good | +| −90 to −100 | Fair | +| −100 to −110 | Poor | +| < −110 | Very Poor | + +### SINR thresholds + +| Range (dB) | Quality | +|---|---| +| ≥ 20 | Excellent | +| 13 to 20 | Good | +| 0 to 13 | Fair | +| −3 to 0 | Poor | +| < −3 | Very Poor | + +--- + +## Running the Tests + +```bash +cd desktop +python -m unittest tests/test_modem_manager.py -v +``` + +--- + +## Extending the Application + +To add support for a new modem API: + +1. Create a `fetch__signal(base_url: str) -> RFSignal` function in + `modem_utils.py` following the existing pattern (e.g. `fetch_hilink_signal`). +2. Add a new entry to the `API_MODES` list in `modem_manager.py`. +3. Handle the new mode in `ModemManagerApp._make_fetch_fn()` in `modem_manager.py`. + +--- + +## Packaging as a Windows Executable (optional) + +```bash +pip install pyinstaller +pyinstaller --onefile --windowed modem_manager.py +# Output: dist/modem_manager.exe +``` + +The resulting `modem_manager.exe` is fully self-contained and requires no +Python installation on the target machine. diff --git a/desktop/modem_manager.py b/desktop/modem_manager.py new file mode 100644 index 0000000..41db0dd --- /dev/null +++ b/desktop/modem_manager.py @@ -0,0 +1,637 @@ +#!/usr/bin/env python3 +""" +Modem RF Parameter Manager +=========================== +A standalone desktop GUI application for monitoring and managing +wireless modem RF parameters via HTTP. + +Supports: + - Huawei HiLink API (most USB and home LTE modems) + - Generic JSON REST API + +Requirements: Python 3.8+ with tkinter (included with Python on Windows) +Usage: python modem_manager.py +""" + +import json +import queue +import threading +import tkinter as tk +from tkinter import messagebox, scrolledtext, ttk +from typing import Dict, List, Optional, Tuple + +from modem_utils import ( + APP_VERSION, + DeviceInfo, + PollingThread, + RFSignal, + RSRP_LEVELS, + SINR_LEVELS, + fetch_hilink_device_info, + fetch_hilink_signal, + fetch_json_signal, + quality_label, + set_hilink_band_lock, +) + +# --------------------------------------------------------------------------- +# Application constants +# --------------------------------------------------------------------------- + +APP_TITLE = f"Modem RF Parameter Manager v{APP_VERSION}" + +REFRESH_INTERVALS = {"2 s": 2, "5 s": 5, "10 s": 10, "30 s": 30, "60 s": 60} + +API_MODES = ["Huawei HiLink (XML)", "Generic JSON"] + + +# --------------------------------------------------------------------------- +# Main application window +# --------------------------------------------------------------------------- + + +class ModemManagerApp(tk.Tk): + """ + Standalone desktop application for monitoring and managing + wireless modem RF parameters. + """ + + def __init__(self) -> None: + super().__init__() + self.title(APP_TITLE) + self.minsize(820, 580) + self.geometry("960x660") + + # Try to enable Windows DPI awareness for crisp rendering + try: + from ctypes import windll # type: ignore[import] + windll.shcore.SetProcessDpiAwareness(1) + except (AttributeError, OSError): + pass + + self._setup_style() + self._build_ui() + + # State + self._poll_thread: Optional[PollingThread] = None + self._result_queue: queue.Queue = queue.Queue() + self._connected = False + self._last_signal: Optional[RFSignal] = None + self._history: List[RFSignal] = [] + + # Start the result-processing loop + self._process_queue() + + # ------------------------------------------------------------------ + # Style + # ------------------------------------------------------------------ + + def _setup_style(self) -> None: + style = ttk.Style(self) + # Prefer native-looking themes; 'vista' on Windows, 'clam' elsewhere + preferred = ["vista", "winnative", "clam", "alt", "default"] + for theme in preferred: + if theme in style.theme_names(): + style.theme_use(theme) + break + + style.configure("Header.TLabel", font=("Segoe UI", 11, "bold")) + style.configure("Value.TLabel", font=("Consolas", 12)) + style.configure("Good.Value.TLabel", font=("Consolas", 12), foreground="#27ae60") + style.configure("Fair.Value.TLabel", font=("Consolas", 12), foreground="#f39c12") + style.configure("Poor.Value.TLabel", font=("Consolas", 12), foreground="#e74c3c") + style.configure("Status.TLabel", font=("Segoe UI", 9)) + + # ------------------------------------------------------------------ + # UI construction + # ------------------------------------------------------------------ + + def _build_ui(self) -> None: + self._build_toolbar() + self._build_notebook() + self._build_statusbar() + + def _build_toolbar(self) -> None: + toolbar = ttk.Frame(self, padding=(8, 6)) + toolbar.pack(fill=tk.X, side=tk.TOP) + + ttk.Label(toolbar, text="Modem URL:").pack(side=tk.LEFT, padx=(0, 4)) + + self._url_var = tk.StringVar(value="http://192.168.8.1") + url_entry = ttk.Entry(toolbar, textvariable=self._url_var, width=32) + url_entry.pack(side=tk.LEFT, padx=(0, 8)) + url_entry.bind("", lambda _e: self._toggle_connection()) + + ttk.Label(toolbar, text="API:").pack(side=tk.LEFT, padx=(0, 4)) + self._api_mode_var = tk.StringVar(value=API_MODES[0]) + api_combo = ttk.Combobox( + toolbar, + textvariable=self._api_mode_var, + values=API_MODES, + state="readonly", + width=22, + ) + api_combo.pack(side=tk.LEFT, padx=(0, 8)) + + ttk.Label(toolbar, text="Refresh:").pack(side=tk.LEFT, padx=(0, 4)) + self._refresh_var = tk.StringVar(value="5 s") + refresh_combo = ttk.Combobox( + toolbar, + textvariable=self._refresh_var, + values=list(REFRESH_INTERVALS.keys()), + state="readonly", + width=7, + ) + refresh_combo.pack(side=tk.LEFT, padx=(0, 12)) + + self._connect_btn = ttk.Button( + toolbar, + text="Connect", + command=self._toggle_connection, + width=12, + ) + self._connect_btn.pack(side=tk.LEFT, padx=(0, 4)) + + self._refresh_now_btn = ttk.Button( + toolbar, + text="Refresh Now", + command=self._refresh_now, + state=tk.DISABLED, + width=12, + ) + self._refresh_now_btn.pack(side=tk.LEFT) + + def _build_notebook(self) -> None: + self._nb = ttk.Notebook(self) + self._nb.pack(fill=tk.BOTH, expand=True, padx=8, pady=(4, 0)) + + self._tab_overview = ttk.Frame(self._nb, padding=12) + self._tab_signal = ttk.Frame(self._nb, padding=12) + self._tab_band = ttk.Frame(self._nb, padding=12) + self._tab_history = ttk.Frame(self._nb, padding=12) + self._tab_raw = ttk.Frame(self._nb, padding=12) + + self._nb.add(self._tab_overview, text=" Overview ") + self._nb.add(self._tab_signal, text=" Signal Quality ") + self._nb.add(self._tab_band, text=" Band & Frequency ") + self._nb.add(self._tab_history, text=" History ") + self._nb.add(self._tab_raw, text=" Raw Response ") + + self._build_overview_tab() + self._build_signal_tab() + self._build_band_tab() + self._build_history_tab() + self._build_raw_tab() + + def _build_statusbar(self) -> None: + bar = ttk.Frame(self, relief=tk.SUNKEN) + bar.pack(fill=tk.X, side=tk.BOTTOM) + + self._status_var = tk.StringVar(value="Not connected.") + ttk.Label(bar, textvariable=self._status_var, style="Status.TLabel", padding=(8, 3)).pack( + side=tk.LEFT + ) + + self._ts_var = tk.StringVar(value="") + ttk.Label(bar, textvariable=self._ts_var, style="Status.TLabel", padding=(8, 3)).pack( + side=tk.RIGHT + ) + + # ------------------------------------------------------------------ + # Overview tab + # ------------------------------------------------------------------ + + def _build_overview_tab(self) -> None: + tab = self._tab_overview + + dev_lf = ttk.LabelFrame(tab, text="Device Information", padding=10) + dev_lf.pack(fill=tk.X, pady=(0, 10)) + + self._dev_labels: Dict[str, tk.StringVar] = {} + dev_fields = [ + ("Device", "device_name"), + ("Firmware", "software_version"), + ("IMEI", "imei"), + ("WAN IP", "wan_ip"), + ] + for row_idx, (label_text, field_key) in enumerate(dev_fields): + ttk.Label(dev_lf, text=label_text + ":", width=12, anchor=tk.E).grid( + row=row_idx, column=0, sticky=tk.E, padx=(0, 6), pady=2 + ) + var = tk.StringVar(value="—") + self._dev_labels[field_key] = var + ttk.Label(dev_lf, textvariable=var, style="Value.TLabel").grid( + row=row_idx, column=1, sticky=tk.W, pady=2 + ) + + sig_lf = ttk.LabelFrame(tab, text="Signal Summary", padding=10) + sig_lf.pack(fill=tk.X) + + self._summary_vars: Dict[str, Tuple[tk.StringVar, tk.StringVar]] = {} + summary_fields = [ + ("RSRP", "rsrp"), + ("SINR", "sinr"), + ("Band", "band"), + ("Mode", "mode"), + ] + for col_idx, (label_text, key) in enumerate(summary_fields): + frame = ttk.Frame(sig_lf) + frame.grid(row=0, column=col_idx, padx=16, pady=4, sticky=tk.N) + ttk.Label(frame, text=label_text, style="Header.TLabel").pack() + val_var = tk.StringVar(value="—") + qual_var = tk.StringVar(value="") + ttk.Label(frame, textvariable=val_var, font=("Consolas", 20, "bold")).pack() + ttk.Label(frame, textvariable=qual_var, foreground="#7f8c8d").pack() + self._summary_vars[key] = (val_var, qual_var) + + # ------------------------------------------------------------------ + # Signal Quality tab + # ------------------------------------------------------------------ + + def _build_signal_tab(self) -> None: + tab = self._tab_signal + self._sig_vars: Dict[str, Tuple[tk.StringVar, tk.StringVar]] = {} + + fields = [ + ("RSSI (dBm)", "rssi", "Received Signal Strength — total power including noise"), + ("RSRP (dBm)", "rsrp", "Reference Signal Received Power — signal quality indicator"), + ("RSRQ (dB)", "rsrq", "Reference Signal Received Quality — channel quality"), + ("SINR (dB)", "sinr", "Signal to Interference + Noise Ratio — link quality"), + ] + for row, (label_text, key, tooltip) in enumerate(fields): + ttk.Label(tab, text=label_text, width=14, anchor=tk.E).grid( + row=row, column=0, sticky=tk.E, padx=(0, 10), pady=8 + ) + val_var = tk.StringVar(value="—") + qual_var = tk.StringVar(value="") + val_lbl = ttk.Label(tab, textvariable=val_var, style="Value.TLabel", width=14) + val_lbl.grid(row=row, column=1, sticky=tk.W, pady=8) + qual_lbl = ttk.Label(tab, textvariable=qual_var, width=12) + qual_lbl.grid(row=row, column=2, sticky=tk.W, pady=8) + ttk.Label(tab, text=tooltip, foreground="#7f8c8d").grid( + row=row, column=3, sticky=tk.W, padx=16, pady=8 + ) + self._sig_vars[key] = (val_var, qual_var) + + # ------------------------------------------------------------------ + # Band & Frequency tab + # ------------------------------------------------------------------ + + def _build_band_tab(self) -> None: + tab = self._tab_band + + info_lf = ttk.LabelFrame(tab, text="Current Band & Frequency", padding=10) + info_lf.pack(fill=tk.X, pady=(0, 12)) + + self._band_vars: Dict[str, tk.StringVar] = {} + band_fields = [ + ("Band", "band"), + ("Channel (EARFCN)", "channel"), + ("Bandwidth", "bandwidth"), + ("Cell ID", "cell_id"), + ("PCI", "pci"), + ] + for row_idx, (label_text, key) in enumerate(band_fields): + ttk.Label(info_lf, text=label_text + ":", width=18, anchor=tk.E).grid( + row=row_idx, column=0, sticky=tk.E, padx=(0, 8), pady=3 + ) + var = tk.StringVar(value="—") + self._band_vars[key] = var + ttk.Label(info_lf, textvariable=var, style="Value.TLabel").grid( + row=row_idx, column=1, sticky=tk.W, pady=3 + ) + + lock_lf = ttk.LabelFrame(tab, text="Band Lock (Huawei HiLink only)", padding=10) + lock_lf.pack(fill=tk.X) + + ttk.Label(lock_lf, text="Bands to lock (comma-separated, e.g. 3,7,20):").pack( + anchor=tk.W + ) + self._band_lock_var = tk.StringVar() + band_lock_entry = ttk.Entry(lock_lf, textvariable=self._band_lock_var, width=32) + band_lock_entry.pack(anchor=tk.W, pady=(4, 8)) + + ttk.Button( + lock_lf, + text="Apply Band Lock", + command=self._apply_band_lock, + ).pack(anchor=tk.W) + + ttk.Label( + lock_lf, + text="\u26a0 Applying a band lock will disconnect and reconnect the modem.", + foreground="#c0392b", + ).pack(anchor=tk.W, pady=(8, 0)) + + # ------------------------------------------------------------------ + # History tab + # ------------------------------------------------------------------ + + def _build_history_tab(self) -> None: + tab = self._tab_history + + cols = ("timestamp", "rsrp", "sinr", "band", "mode") + self._hist_tree = ttk.Treeview(tab, columns=cols, show="headings", height=16) + headings = { + "timestamp": ("Timestamp", 160), + "rsrp": ("RSRP (dBm)", 100), + "sinr": ("SINR (dB)", 100), + "band": ("Band", 80), + "mode": ("Mode", 80), + } + for col, (heading, width) in headings.items(): + self._hist_tree.heading(col, text=heading) + self._hist_tree.column(col, width=width, anchor=tk.CENTER) + + scrollbar = ttk.Scrollbar(tab, orient=tk.VERTICAL, command=self._hist_tree.yview) + self._hist_tree.configure(yscrollcommand=scrollbar.set) + self._hist_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) + scrollbar.pack(side=tk.RIGHT, fill=tk.Y) + + btn_frame = ttk.Frame(tab) + btn_frame.pack(side=tk.BOTTOM, fill=tk.X, pady=(6, 0)) + ttk.Button(btn_frame, text="Clear History", command=self._clear_history).pack(side=tk.LEFT) + ttk.Button( + btn_frame, text="Export JSON\u2026", command=self._export_history_json + ).pack(side=tk.LEFT, padx=6) + + # ------------------------------------------------------------------ + # Raw response tab + # ------------------------------------------------------------------ + + def _build_raw_tab(self) -> None: + tab = self._tab_raw + + self._raw_text = scrolledtext.ScrolledText( + tab, + wrap=tk.NONE, + font=("Consolas", 9), + state=tk.DISABLED, + ) + self._raw_text.pack(fill=tk.BOTH, expand=True) + + ttk.Button(tab, text="Copy to Clipboard", command=self._copy_raw).pack( + anchor=tk.E, pady=(4, 0) + ) + + # ------------------------------------------------------------------ + # Connection management + # ------------------------------------------------------------------ + + def _toggle_connection(self) -> None: + if self._connected: + self._disconnect() + else: + self._connect() + + def _make_fetch_fn(self, base_url: str, api_mode: str): # type: ignore[return] + """Return a zero-argument callable that fetches signal data from the modem.""" + if api_mode == API_MODES[0]: # HiLink + + def _fetch_hilink() -> RFSignal: + return fetch_hilink_signal(base_url) + + return _fetch_hilink + else: + endpoint = base_url.rstrip("/") + "/signal" + + def _fetch_json() -> RFSignal: + return fetch_json_signal(endpoint) + + return _fetch_json + + def _connect(self) -> None: + base_url = self._url_var.get().strip() + if not base_url: + messagebox.showerror("Error", "Please enter a modem URL.") + return + + if not base_url.startswith(("http://", "https://")): + base_url = "http://" + base_url + self._url_var.set(base_url) + + api_mode = self._api_mode_var.get() + interval = REFRESH_INTERVALS.get(self._refresh_var.get(), 5) + fetch_fn = self._make_fetch_fn(base_url, api_mode) + + self._fetch_device_info_async(base_url, api_mode) + + self._poll_thread = PollingThread( + result_queue=self._result_queue, + fetch_fn=fetch_fn, + interval_s=interval, + ) + self._poll_thread.start() + self._connected = True + self._connect_btn.configure(text="Disconnect") + self._refresh_now_btn.configure(state=tk.NORMAL) + self._status_var.set(f"Connecting to {base_url} \u2026") + + def _disconnect(self) -> None: + if self._poll_thread: + self._poll_thread.stop() + self._poll_thread = None + self._connected = False + self._connect_btn.configure(text="Connect") + self._refresh_now_btn.configure(state=tk.DISABLED) + self._status_var.set("Disconnected.") + self._ts_var.set("") + + def _refresh_now(self) -> None: + """Trigger an immediate fetch outside the scheduled interval.""" + if self._poll_thread is None: + return + base_url = self._url_var.get().strip() + api_mode = self._api_mode_var.get() + fetch_fn = self._make_fetch_fn(base_url, api_mode) + + def _run() -> None: + try: + result = fetch_fn() + self._result_queue.put(("ok", result)) + except (OSError, ValueError) as exc: + self._result_queue.put(("error", str(exc))) + + threading.Thread(target=_run, daemon=True).start() + + def _fetch_device_info_async(self, base_url: str, api_mode: str) -> None: + def _run() -> None: + try: + if api_mode == API_MODES[0]: + info = fetch_hilink_device_info(base_url) + self._result_queue.put(("device_info", info)) + except (OSError, ValueError): + pass # Device info is optional — silently ignore on any network/parse error + + threading.Thread(target=_run, daemon=True).start() + + # ------------------------------------------------------------------ + # Result queue processing (called from main thread every 200 ms) + # ------------------------------------------------------------------ + + def _process_queue(self) -> None: + try: + while True: + event, payload = self._result_queue.get_nowait() + if event == "ok": + self._update_signal(payload) + elif event == "error": + self._status_var.set(f"Error: {payload}") + elif event == "device_info": + self._update_device_info(payload) + except queue.Empty: + pass + self.after(200, self._process_queue) + + # ------------------------------------------------------------------ + # UI update helpers + # ------------------------------------------------------------------ + + def _update_signal(self, signal: RFSignal) -> None: + self._last_signal = signal + self._history.append(signal) + + self._status_var.set("Connected \u2014 receiving data.") + self._ts_var.set(f"Last update: {signal.timestamp or '\u2014'}") + + # Overview summary + for key, (val_var, qual_var) in self._summary_vars.items(): + raw_val = getattr(signal, key, None) or "\u2014" + val_var.set(raw_val) + if key == "rsrp": + label, _colour = quality_label(raw_val, RSRP_LEVELS) + qual_var.set(label) + elif key == "sinr": + label, _colour = quality_label(raw_val, SINR_LEVELS) + qual_var.set(label) + else: + qual_var.set("") + + # Signal quality tab + for key, (val_var, qual_var) in self._sig_vars.items(): + raw_val = getattr(signal, key, None) or "\u2014" + val_var.set(raw_val) + if key == "rsrp": + label, _ = quality_label(raw_val, RSRP_LEVELS) + qual_var.set(label) + elif key == "sinr": + label, _ = quality_label(raw_val, SINR_LEVELS) + qual_var.set(label) + + # Band & frequency tab + for key, var in self._band_vars.items(): + var.set(getattr(signal, key, None) or "\u2014") + + # History tab + self._hist_tree.insert( + "", + tk.END, + values=( + signal.timestamp or "", + signal.rsrp or "\u2014", + signal.sinr or "\u2014", + signal.band or "\u2014", + signal.mode or "\u2014", + ), + ) + children = self._hist_tree.get_children() + if children: + self._hist_tree.see(children[-1]) + + # Raw tab + raw_content = json.dumps(signal.as_dict(), indent=2) + self._raw_text.configure(state=tk.NORMAL) + self._raw_text.delete("1.0", tk.END) + self._raw_text.insert(tk.END, raw_content) + self._raw_text.configure(state=tk.DISABLED) + + def _update_device_info(self, info: DeviceInfo) -> None: + for key, var in self._dev_labels.items(): + var.set(getattr(info, key, None) or "\u2014") + + # ------------------------------------------------------------------ + # Band lock + # ------------------------------------------------------------------ + + def _apply_band_lock(self) -> None: + raw = self._band_lock_var.get().strip() + if not raw: + messagebox.showerror("Band Lock", "Enter at least one band number.") + return + bands = [b.strip() for b in raw.split(",") if b.strip()] + base_url = self._url_var.get().strip() + + if not messagebox.askyesno( + "Confirm Band Lock", + f"Apply band lock to bands {bands} on {base_url}?\n\n" + "This will cause a brief modem reconnect.", + ): + return + + def _run() -> None: + try: + set_hilink_band_lock(base_url, bands) + self._result_queue.put(("ok_msg", f"Band lock applied: {bands}")) + except (OSError, ValueError) as exc: + self._result_queue.put(("error", f"Band lock failed: {exc}")) + + threading.Thread(target=_run, daemon=True).start() + + # ------------------------------------------------------------------ + # History tab actions + # ------------------------------------------------------------------ + + def _clear_history(self) -> None: + for item in self._hist_tree.get_children(): + self._hist_tree.delete(item) + self._history.clear() + + def _export_history_json(self) -> None: + if not self._history: + messagebox.showinfo("Export", "No history to export.") + return + from tkinter import filedialog + + path = filedialog.asksaveasfilename( + defaultextension=".json", + filetypes=[("JSON files", "*.json"), ("All files", "*.*")], + initialfile="rf_history.json", + ) + if not path: + return + data = [s.as_dict() for s in self._history] + with open(path, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=2) + messagebox.showinfo("Export", f"Exported {len(data)} record(s) to:\n{path}") + + # ------------------------------------------------------------------ + # Raw tab actions + # ------------------------------------------------------------------ + + def _copy_raw(self) -> None: + content = self._raw_text.get("1.0", tk.END) + self.clipboard_clear() + self.clipboard_append(content) + self._status_var.set("Raw content copied to clipboard.") + + # ------------------------------------------------------------------ + # Window close + # ------------------------------------------------------------------ + + def destroy(self) -> None: + self._disconnect() + super().destroy() + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +def main() -> None: + app = ModemManagerApp() + app.mainloop() + + +if __name__ == "__main__": + main() diff --git a/desktop/modem_utils.py b/desktop/modem_utils.py new file mode 100644 index 0000000..62c0a57 --- /dev/null +++ b/desktop/modem_utils.py @@ -0,0 +1,369 @@ +""" +modem_utils.py — Pure-function utilities for Modem RF Parameter Manager. + +Contains all logic that does NOT depend on tkinter: + - Data models (RFSignal, DeviceInfo) + - HTTP helpers (_http_get, _http_post) + - Huawei HiLink API adapter + - Generic JSON API adapter + - Signal quality interpretation helpers + - Background polling thread +""" + +import json +import queue +import threading +import urllib.error +import urllib.parse +import urllib.request +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Tuple + +# --------------------------------------------------------------------------- +# Version +# --------------------------------------------------------------------------- +APP_VERSION = "1.0.0" + +# --------------------------------------------------------------------------- +# Data models +# --------------------------------------------------------------------------- + + +@dataclass +class RFSignal: + """Snapshot of current modem RF signal parameters.""" + + rssi: Optional[str] = None # Received Signal Strength Indicator (dBm) + rsrp: Optional[str] = None # Reference Signal Received Power (dBm) + rsrq: Optional[str] = None # Reference Signal Received Quality (dB) + sinr: Optional[str] = None # Signal-to-Interference-plus-Noise Ratio (dB) + band: Optional[str] = None # Active LTE / NR band (e.g. "B3") + channel: Optional[str] = None # EARFCN / ARFCN + bandwidth: Optional[str] = None # Channel bandwidth (MHz) + mode: Optional[str] = None # Connection mode (LTE, NR, WCDMA…) + cell_id: Optional[str] = None # Cell identifier + pci: Optional[str] = None # Physical Cell ID + timestamp: Optional[str] = None # ISO-8601 timestamp of the reading + + def as_dict(self) -> Dict[str, Optional[str]]: + return { + "rssi": self.rssi, + "rsrp": self.rsrp, + "rsrq": self.rsrq, + "sinr": self.sinr, + "band": self.band, + "channel": self.channel, + "bandwidth": self.bandwidth, + "mode": self.mode, + "cell_id": self.cell_id, + "pci": self.pci, + "timestamp": self.timestamp, + } + + +@dataclass +class DeviceInfo: + """Static device information returned by the modem.""" + + device_name: Optional[str] = None + hardware_version: Optional[str] = None + software_version: Optional[str] = None + imei: Optional[str] = None + imsi: Optional[str] = None + iccid: Optional[str] = None + msisdn: Optional[str] = None + wan_ip: Optional[str] = None + + +# --------------------------------------------------------------------------- +# API client helpers +# --------------------------------------------------------------------------- + +REQUEST_TIMEOUT = 8 # seconds + + +def _http_get(url: str, timeout: int = REQUEST_TIMEOUT) -> Tuple[int, str]: + """ + Perform an HTTP GET and return (status_code, body_text). + Raises urllib.error.URLError / urllib.error.HTTPError on failure. + """ + req = urllib.request.Request( + url, + headers={ + "Accept": "application/json, application/xml, text/xml, */*", + "User-Agent": f"ModemRFManager/{APP_VERSION}", + }, + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + body = response.read().decode("utf-8", errors="replace") + return response.status, body + + +def _http_post( + url: str, + data: Any, + timeout: int = REQUEST_TIMEOUT, + content_type: str = "application/json", +) -> Tuple[int, str]: + """ + Perform an HTTP POST with a JSON (or XML) body. + Returns (status_code, body_text). + """ + if content_type == "application/json": + raw = json.dumps(data).encode("utf-8") + else: + raw = data.encode("utf-8") if isinstance(data, str) else str(data).encode("utf-8") + + req = urllib.request.Request( + url, + data=raw, + headers={ + "Content-Type": content_type, + "Accept": "application/json, application/xml, */*", + "User-Agent": f"ModemRFManager/{APP_VERSION}", + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + body = response.read().decode("utf-8", errors="replace") + return response.status, body + + +# --------------------------------------------------------------------------- +# Huawei HiLink API adapter +# --------------------------------------------------------------------------- + +HILINK_PATHS = { + "device_info": "/api/device/information", + "signal": "/api/device/signal", + "status": "/api/monitoring/status", + "token": "/api/webserver/SesTokInfo", +} + + +def _xml_text(element: Optional[ET.Element], tag: str, default: str = "—") -> str: + """Return text of *tag* child, or *default* if absent/empty.""" + if element is None: + return default + child = element.find(tag) + return child.text.strip() if child is not None and child.text else default + + +def fetch_hilink_signal(base_url: str) -> RFSignal: + """ + Fetch signal parameters from a Huawei HiLink modem at *base_url*. + + Example: fetch_hilink_signal("http://192.168.8.1") + """ + import datetime + + url = base_url.rstrip("/") + HILINK_PATHS["signal"] + _status, body = _http_get(url) + root = ET.fromstring(body) + + return RFSignal( + rssi=_xml_text(root, "rssi"), + rsrp=_xml_text(root, "rsrp"), + rsrq=_xml_text(root, "rsrq"), + sinr=_xml_text(root, "sinr"), + band=_xml_text(root, "band"), + channel=_xml_text(root, "earfcn"), + bandwidth=_xml_text(root, "bandwidth"), + mode=_xml_text(root, "mode"), + cell_id=_xml_text(root, "cell_id"), + pci=_xml_text(root, "pci"), + timestamp=datetime.datetime.now().isoformat(timespec="seconds"), + ) + + +def fetch_hilink_device_info(base_url: str) -> DeviceInfo: + """Fetch static device information from a Huawei HiLink modem.""" + url = base_url.rstrip("/") + HILINK_PATHS["device_info"] + _status, body = _http_get(url) + root = ET.fromstring(body) + return DeviceInfo( + device_name=_xml_text(root, "DeviceName"), + hardware_version=_xml_text(root, "HardwareVersion"), + software_version=_xml_text(root, "SoftwareVersion"), + imei=_xml_text(root, "Imei"), + imsi=_xml_text(root, "Imsi"), + iccid=_xml_text(root, "Iccid"), + msisdn=_xml_text(root, "Msisdn"), + wan_ip=_xml_text(root, "WanIPAddress"), + ) + + +def set_hilink_band_lock(base_url: str, bands: List[str]) -> bool: + """ + Request a band lock on a HiLink modem. + + *bands* is a list of band numbers as strings, e.g. ["3", "7", "20"]. + Returns True on success, False if no valid bands were provided. + """ + bitmask = _bands_to_bitmask(bands) + if bitmask == 0: + return False + + bitmask_hex = format(bitmask, "016X") + xml_body = ( + "" + "" + "03" + f"{bitmask_hex}" + "0" + "" + ) + url = base_url.rstrip("/") + "/api/net/net-mode" + _http_post(url, xml_body, content_type="application/xml") + return True + + +def _bands_to_bitmask(bands: List[str]) -> int: + """ + Convert a list of band number strings to an LTE band bitmask. + + Band n maps to bit (n-1). Valid range: 1–64. + """ + bitmask = 0 + for b in bands: + try: + n = int(b) + if 1 <= n <= 64: + bitmask |= 1 << (n - 1) + except ValueError: + pass + return bitmask + + +# --------------------------------------------------------------------------- +# Generic JSON API adapter +# --------------------------------------------------------------------------- + +JSON_SIGNAL_FIELD_MAP: Dict[str, List[str]] = { + "rssi": ["rssi", "RSSI", "signal_strength"], + "rsrp": ["rsrp", "RSRP"], + "rsrq": ["rsrq", "RSRQ"], + "sinr": ["sinr", "SINR", "snr", "SNR"], + "band": ["band", "Band", "lte_band", "cell_band"], + "channel": ["earfcn", "EARFCN", "channel", "arfcn"], + "bandwidth": ["bandwidth", "Bandwidth", "bw"], + "mode": ["mode", "Mode", "network_type", "rat"], + "cell_id": ["cell_id", "CellId", "cellId", "enodeb_id"], + "pci": ["pci", "PCI", "physical_cell_id"], +} + + +def _find_json_field(data: Dict[str, Any], candidates: List[str]) -> Optional[str]: + """Return the value of the first matching key in *data*, as a string.""" + for key in candidates: + if key in data: + val = data[key] + return str(val) if val is not None else None + return None + + +def fetch_json_signal(endpoint_url: str) -> RFSignal: + """ + Fetch signal parameters from a generic JSON REST endpoint. + Field names are matched using common variations. + """ + import datetime + + _status, body = _http_get(endpoint_url) + data: Dict[str, Any] = json.loads(body) + + # Allow top-level wrapper objects (e.g. {"signal": {...}}) + for key in ("signal", "data", "result", "response"): + if key in data and isinstance(data[key], dict): + data = data[key] + break + + return RFSignal( + rssi=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["rssi"]), + rsrp=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["rsrp"]), + rsrq=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["rsrq"]), + sinr=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["sinr"]), + band=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["band"]), + channel=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["channel"]), + bandwidth=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["bandwidth"]), + mode=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["mode"]), + cell_id=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["cell_id"]), + pci=_find_json_field(data, JSON_SIGNAL_FIELD_MAP["pci"]), + timestamp=datetime.datetime.now().isoformat(timespec="seconds"), + ) + + +# --------------------------------------------------------------------------- +# Signal quality interpretation helpers +# --------------------------------------------------------------------------- + +RSRP_LEVELS = [ + (-80, "Excellent", "#27ae60"), + (-90, "Good", "#2ecc71"), + (-100, "Fair", "#f39c12"), + (-110, "Poor", "#e67e22"), + (None, "Very Poor", "#e74c3c"), +] + +SINR_LEVELS = [ + (20, "Excellent", "#27ae60"), + (13, "Good", "#2ecc71"), + (0, "Fair", "#f39c12"), + (-3, "Poor", "#e67e22"), + (None, "Very Poor", "#e74c3c"), +] + + +def quality_label(value_str: Optional[str], levels: list) -> Tuple[str, str]: + """ + Return (quality_text, colour_hex) for a dBm/dB value string. + *levels* is a list of (threshold, label, colour) tuples, descending. + """ + if not value_str or value_str == "—": + return "N/A", "#95a5a6" + try: + val = float(value_str.replace("dBm", "").replace("dB", "").strip()) + except ValueError: + return "N/A", "#95a5a6" + + for threshold, label, colour in levels: + if threshold is None or val >= threshold: + return label, colour + return "N/A", "#95a5a6" + + +# --------------------------------------------------------------------------- +# Background polling thread +# --------------------------------------------------------------------------- + + +class PollingThread(threading.Thread): + """ + Background thread that fetches RF parameters at a fixed interval and + puts results (or errors) on a thread-safe queue. + """ + + def __init__( + self, + result_queue: "queue.Queue[Tuple[str, Any]]", + fetch_fn: Any, + interval_s: float = 5.0, + ) -> None: + super().__init__(daemon=True) + self._queue = result_queue + self._fetch = fetch_fn + self._interval = interval_s + self._stop_event = threading.Event() + + def run(self) -> None: + while not self._stop_event.is_set(): + try: + result = self._fetch() + self._queue.put(("ok", result)) + except (OSError, ValueError) as exc: + self._queue.put(("error", str(exc))) + self._stop_event.wait(self._interval) + + def stop(self) -> None: + self._stop_event.set() diff --git a/desktop/requirements.txt b/desktop/requirements.txt new file mode 100644 index 0000000..40d33c7 --- /dev/null +++ b/desktop/requirements.txt @@ -0,0 +1,10 @@ +# No third-party packages required. +# The application uses only Python's standard library: +# tkinter — bundled with CPython on Windows, macOS, and most Linux distros +# urllib — stdlib HTTP client +# xml.etree.ElementTree — stdlib XML parser +# json, threading, queue, dataclasses — all stdlib + +# Optional: install PyInstaller to build a standalone Windows executable +# pyinstaller --onefile --windowed modem_manager.py +# pyinstaller>=6.0 diff --git a/desktop/tests/test_modem_manager.py b/desktop/tests/test_modem_manager.py new file mode 100644 index 0000000..0578417 --- /dev/null +++ b/desktop/tests/test_modem_manager.py @@ -0,0 +1,372 @@ +""" +Unit tests for modem_utils utility functions. + +Run with: python -m unittest tests/test_modem_manager.py + or: python -m pytest tests/test_modem_manager.py +""" + +import sys +import os +import unittest +import json +import xml.etree.ElementTree as ET + +# Allow importing modem_utils from the parent directory +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import modem_utils as mm + + +class TestQualityLabel(unittest.TestCase): + """Tests for the quality_label helper function.""" + + def test_excellent_rsrp(self): + label, colour = mm.quality_label("-75", mm.RSRP_LEVELS) + self.assertEqual(label, "Excellent") + self.assertEqual(colour, "#27ae60") + + def test_good_rsrp(self): + label, colour = mm.quality_label("-85", mm.RSRP_LEVELS) + self.assertEqual(label, "Good") + self.assertEqual(colour, "#2ecc71") + + def test_fair_rsrp(self): + label, colour = mm.quality_label("-95", mm.RSRP_LEVELS) + self.assertEqual(label, "Fair") + self.assertEqual(colour, "#f39c12") + + def test_poor_rsrp(self): + label, colour = mm.quality_label("-105", mm.RSRP_LEVELS) + self.assertEqual(label, "Poor") + self.assertEqual(colour, "#e67e22") + + def test_very_poor_rsrp(self): + label, colour = mm.quality_label("-115", mm.RSRP_LEVELS) + self.assertEqual(label, "Very Poor") + self.assertEqual(colour, "#e74c3c") + + def test_none_value(self): + label, colour = mm.quality_label(None, mm.RSRP_LEVELS) + self.assertEqual(label, "N/A") + self.assertEqual(colour, "#95a5a6") + + def test_dash_value(self): + label, colour = mm.quality_label("—", mm.RSRP_LEVELS) + self.assertEqual(label, "N/A") + self.assertEqual(colour, "#95a5a6") + + def test_non_numeric_value(self): + label, colour = mm.quality_label("not_a_number", mm.RSRP_LEVELS) + self.assertEqual(label, "N/A") + + def test_excellent_sinr(self): + label, colour = mm.quality_label("25", mm.SINR_LEVELS) + self.assertEqual(label, "Excellent") + + def test_good_sinr(self): + label, colour = mm.quality_label("15", mm.SINR_LEVELS) + self.assertEqual(label, "Good") + + def test_fair_sinr(self): + label, colour = mm.quality_label("5", mm.SINR_LEVELS) + self.assertEqual(label, "Fair") + + def test_poor_sinr(self): + label, colour = mm.quality_label("-2", mm.SINR_LEVELS) + self.assertEqual(label, "Poor") + + def test_very_poor_sinr(self): + label, colour = mm.quality_label("-10", mm.SINR_LEVELS) + self.assertEqual(label, "Very Poor") + + def test_value_with_dbm_suffix(self): + label, _ = mm.quality_label("-75dBm", mm.RSRP_LEVELS) + self.assertEqual(label, "Excellent") + + def test_value_with_db_suffix(self): + label, _ = mm.quality_label("-75dB", mm.RSRP_LEVELS) + self.assertEqual(label, "Excellent") + + def test_value_with_whitespace(self): + label, _ = mm.quality_label(" -85 ", mm.RSRP_LEVELS) + self.assertEqual(label, "Good") + + +class TestXmlText(unittest.TestCase): + """Tests for the _xml_text helper function.""" + + def _make_element(self, xml_str: str) -> ET.Element: + return ET.fromstring(xml_str) + + def test_finds_existing_tag(self): + root = self._make_element("-90") + result = mm._xml_text(root, "rsrp") + self.assertEqual(result, "-90") + + def test_returns_default_for_missing_tag(self): + root = self._make_element("") + result = mm._xml_text(root, "rsrp") + self.assertEqual(result, "—") + + def test_returns_custom_default_for_missing_tag(self): + root = self._make_element("") + result = mm._xml_text(root, "rsrp", default="N/A") + self.assertEqual(result, "N/A") + + def test_returns_default_for_none_element(self): + result = mm._xml_text(None, "rsrp") + self.assertEqual(result, "—") + + def test_trims_whitespace(self): + root = self._make_element(" B3 ") + result = mm._xml_text(root, "band") + self.assertEqual(result, "B3") + + def test_returns_default_for_empty_tag(self): + root = self._make_element("") + result = mm._xml_text(root, "rsrp") + self.assertEqual(result, "—") + + +class TestFindJsonField(unittest.TestCase): + """Tests for the _find_json_field helper function.""" + + def test_finds_first_candidate(self): + data = {"rssi": "-70", "RSSI": "-71"} + result = mm._find_json_field(data, ["rssi", "RSSI"]) + self.assertEqual(result, "-70") + + def test_falls_back_to_second_candidate(self): + data = {"RSSI": "-71"} + result = mm._find_json_field(data, ["rssi", "RSSI"]) + self.assertEqual(result, "-71") + + def test_returns_none_when_no_match(self): + data = {"unrelated_key": "value"} + result = mm._find_json_field(data, ["rssi", "RSSI"]) + self.assertIsNone(result) + + def test_converts_int_to_str(self): + data = {"rsrp": -90} + result = mm._find_json_field(data, ["rsrp"]) + self.assertEqual(result, "-90") + + def test_returns_none_for_none_value(self): + data = {"rsrp": None} + result = mm._find_json_field(data, ["rsrp"]) + self.assertIsNone(result) + + def test_empty_data(self): + result = mm._find_json_field({}, ["rsrp", "RSRP"]) + self.assertIsNone(result) + + +class TestRFSignalAsDict(unittest.TestCase): + """Tests for RFSignal.as_dict().""" + + def test_all_none_by_default(self): + signal = mm.RFSignal() + d = signal.as_dict() + self.assertIn("rssi", d) + self.assertIn("rsrp", d) + self.assertIn("sinr", d) + self.assertIn("band", d) + self.assertIn("timestamp", d) + for val in d.values(): + self.assertIsNone(val) + + def test_populated_values_round_trip(self): + signal = mm.RFSignal( + rssi="-75", + rsrp="-85", + rsrq="-10", + sinr="20", + band="B3", + channel="1300", + timestamp="2026-01-01T00:00:00", + ) + d = signal.as_dict() + self.assertEqual(d["rssi"], "-75") + self.assertEqual(d["rsrp"], "-85") + self.assertEqual(d["band"], "B3") + self.assertEqual(d["timestamp"], "2026-01-01T00:00:00") + + def test_as_dict_is_json_serialisable(self): + signal = mm.RFSignal(rsrp="-90", sinr="15", band="B7") + serialised = json.dumps(signal.as_dict()) + restored = json.loads(serialised) + self.assertEqual(restored["rsrp"], "-90") + self.assertEqual(restored["band"], "B7") + + +class TestHilinkBandLockBitmask(unittest.TestCase): + """ + Test the band-lock bitmask logic inside set_hilink_band_lock. + + We test the bitmask calculation logic directly (without making HTTP calls). + """ + + def _compute_bitmask(self, bands): + """Delegate to the public modem_utils helper.""" + return mm._bands_to_bitmask(bands) + + def test_band_1(self): + bitmask = self._compute_bitmask(["1"]) + self.assertEqual(bitmask, 0b01) + + def test_band_3(self): + bitmask = self._compute_bitmask(["3"]) + self.assertEqual(bitmask, 0b100) + + def test_band_7_and_20(self): + bitmask = self._compute_bitmask(["7", "20"]) + expected = (1 << 6) | (1 << 19) + self.assertEqual(bitmask, expected) + + def test_all_bands(self): + bitmask = self._compute_bitmask(["3", "7", "20"]) + expected = (1 << 2) | (1 << 6) | (1 << 19) + self.assertEqual(bitmask, expected) + + def test_invalid_band_ignored(self): + bitmask = self._compute_bitmask(["abc", "3"]) + self.assertEqual(bitmask, 1 << 2) + + def test_out_of_range_band_ignored(self): + bitmask = self._compute_bitmask(["0", "65", "3"]) + self.assertEqual(bitmask, 1 << 2) + + def test_empty_bands_zero(self): + bitmask = self._compute_bitmask([]) + self.assertEqual(bitmask, 0) + + def test_bitmask_hex_format(self): + bitmask = self._compute_bitmask(["3"]) + hex_str = format(bitmask, "016X") + self.assertEqual(len(hex_str), 16) + self.assertEqual(hex_str, "0000000000000004") + + +class TestFetchJsonSignalParsing(unittest.TestCase): + """ + Tests for fetch_json_signal field-mapping logic, using a mock HTTP call. + """ + + def _parse_json_body(self, body: str) -> mm.RFSignal: + """ + Replicate the parsing logic of fetch_json_signal without HTTP. + """ + import datetime + + data = json.loads(body) + for key in ("signal", "data", "result", "response"): + if key in data and isinstance(data[key], dict): + data = data[key] + break + + return mm.RFSignal( + rssi=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["rssi"]), + rsrp=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["rsrp"]), + rsrq=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["rsrq"]), + sinr=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["sinr"]), + band=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["band"]), + channel=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["channel"]), + bandwidth=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["bandwidth"]), + mode=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["mode"]), + cell_id=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["cell_id"]), + pci=mm._find_json_field(data, mm.JSON_SIGNAL_FIELD_MAP["pci"]), + timestamp=datetime.datetime.now().isoformat(timespec="seconds"), + ) + + def test_flat_json_lowercase(self): + body = json.dumps({"rsrp": "-88", "sinr": "12", "band": "B3"}) + signal = self._parse_json_body(body) + self.assertEqual(signal.rsrp, "-88") + self.assertEqual(signal.sinr, "12") + self.assertEqual(signal.band, "B3") + + def test_flat_json_uppercase(self): + body = json.dumps({"RSRP": "-88", "SINR": "12"}) + signal = self._parse_json_body(body) + self.assertEqual(signal.rsrp, "-88") + self.assertEqual(signal.sinr, "12") + + def test_wrapped_in_signal_key(self): + body = json.dumps({"signal": {"rsrp": "-92", "band": "B7"}}) + signal = self._parse_json_body(body) + self.assertEqual(signal.rsrp, "-92") + self.assertEqual(signal.band, "B7") + + def test_wrapped_in_data_key(self): + body = json.dumps({"data": {"rsrp": "-95", "earfcn": "2850"}}) + signal = self._parse_json_body(body) + self.assertEqual(signal.rsrp, "-95") + self.assertEqual(signal.channel, "2850") + + def test_missing_fields_are_none(self): + body = json.dumps({"rsrp": "-88"}) + signal = self._parse_json_body(body) + self.assertIsNone(signal.rssi) + self.assertIsNone(signal.band) + + +class TestHilinkXmlParsing(unittest.TestCase): + """Tests for Huawei HiLink XML signal response parsing.""" + + def _parse_xml_response(self, xml_str: str) -> mm.RFSignal: + """Replicate fetch_hilink_signal parsing without HTTP.""" + import datetime + + root = ET.fromstring(xml_str) + return mm.RFSignal( + rssi=mm._xml_text(root, "rssi"), + rsrp=mm._xml_text(root, "rsrp"), + rsrq=mm._xml_text(root, "rsrq"), + sinr=mm._xml_text(root, "sinr"), + band=mm._xml_text(root, "band"), + channel=mm._xml_text(root, "earfcn"), + bandwidth=mm._xml_text(root, "bandwidth"), + mode=mm._xml_text(root, "mode"), + cell_id=mm._xml_text(root, "cell_id"), + pci=mm._xml_text(root, "pci"), + timestamp=datetime.datetime.now().isoformat(timespec="seconds"), + ) + + def test_typical_hilink_response(self): + xml = ( + "" + "" + "-75" + "-90" + "-11" + "18" + "B3" + "1300" + "20" + "LTE" + "12345" + "123" + "" + ) + signal = self._parse_xml_response(xml) + self.assertEqual(signal.rssi, "-75") + self.assertEqual(signal.rsrp, "-90") + self.assertEqual(signal.rsrq, "-11") + self.assertEqual(signal.sinr, "18") + self.assertEqual(signal.band, "B3") + self.assertEqual(signal.channel, "1300") + self.assertEqual(signal.bandwidth, "20") + self.assertEqual(signal.mode, "LTE") + self.assertEqual(signal.cell_id, "12345") + self.assertEqual(signal.pci, "123") + + def test_partial_response_uses_dash_default(self): + xml = "-88" + signal = self._parse_xml_response(xml) + self.assertEqual(signal.rsrp, "-88") + self.assertEqual(signal.rssi, "—") + self.assertEqual(signal.band, "—") + + +if __name__ == "__main__": + unittest.main()