Skip to content

refactor: extract modem parser format families #792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
itsDNNS merged 2 commits into from
Aug 14, 2026
Merged

refactor: extract modem parser format families #792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bc98e7a
test: freeze modem parser outputs
itsDNNS Aug 14, 2026
9684b05
refactor: extract modem parser format families
itsDNNS Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 63 additions & 22 deletions ARCHITECTURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,33 @@ create_backup_to_file(data_dir, dest_dir)

## Driver Architecture

Modem drivers live in `app/drivers/` and implement the `ModemDriver` base class:
Modem drivers live in `app/drivers/` and implement the `ModemDriver` base
class. A driver owns device orchestration: network requests, endpoint and
firmware selection, authentication, sessions, retries, TLS, and cryptography.
It fetches a raw modem payload and delegates normalization to one of the pure,
explicit profiles in `app/drivers/formats/`.

The dependency direction is one-way:

```text
collector -> concrete driver -> named format profile -> parser primitives/types
-> transport helpers (driver code only)
```

Format modules never depend on a concrete driver, session, request client,
Flask, clocks, randomness, or cryptography. They return an immutable
`ParseResult(value, diagnostics)`. Diagnostics contain only the finite safe
fields `family`, `profile`, `code`, `direction`, `row`, `index`, and `field`.
Public driver methods unwrap the result and retain the established
`DocsisData`/channel-list contracts.

`app/drivers/format_compat.py` is a finite compatibility boundary for legacy
warning messages. `app/drivers/arris_html.py` remains an import-compatible shim
for the established bonded 8/7-column parser. Existing private parser seams
that are covered by integrations remain one-statement delegations; format
grammar is not implemented in concrete drivers.

The base interface is:

```python
class ModemDriver(ABC):
Expand All @@ -446,27 +472,42 @@ class ModemDriver(ABC):
def get_connection_info(self) -> dict: ...
```

### Supported Drivers

| Driver | Module | Hardware | Auth |
|--------|--------|----------|------|
| `fritzbox` | `fritzbox.py` | AVM FRITZ!Box | SID-based (data.lua) |
| `tc4400` | `tc4400.py` | Technicolor TC4400 | SNMP |
| `ultrahub7` | `ultrahub7.py` | Vodafone Ultra Hub 7 | Session cookie |
| `cm3500` | `cm3500.py` | Arris CM3500B | Form POST (IP-based session) |
| `connectbox` | `connectbox.py` | Unitymedia Connect Box (CH7465) | Session cookie |
| `vodafone_station` | `vodafone_station.py` | CGA6444VF, CGA4322DE, TG3442DE | Auto-detected (see below) |
| `cm1000` | `cm1000.py` | Netgear CM1000 | HTTP Basic or local Genie form |
| `cm3000` | `cm3000.py` | Netgear CM3000 | HTTP Basic Auth |
| `surfboard` | `surfboard.py` | Arris SURFboard S33/S34/SB8200 | HNAP1 HMAC-SHA256 |
| `sb6183` | `sb6183.py` | Arris SB6183 | None (HTTP status pages) |
| `cm8200` | `cm8200.py` | Arris Touchstone CM8200A | Base64 query string |
| `hitron_coda_4680` | `hitron_coda_4680.py` | Hitron CODA-4680 | Form POST (`/1/Device/Users/Login`) |
| `generic` | `generic.py` | Generic Router (no DOCSIS) | None |

### Driver Registry (`app/drivers/__init__.py`)

Drivers are loaded by name via `load_driver(modem_type, url, user, password)`. The registry maps type strings to fully qualified class paths for lazy importing.
### Exact driver-to-profile matrix

Every registered concrete class exposes a non-empty immutable
`FORMAT_FAMILIES` tuple. Registry aliases appear together in the first column;
there are 21 keys, 20 concrete classes, and 22 explicit profiles.

| Registry key(s) | Concrete class | Format profile(s) | Cohesive module / entrypoint |
|---|---|---|---|
| `cgm4981` | `CGM4981Driver` | `cgm4981_columnar_html` | `html_columnar.parse_cgm4981_columnar_html` |
| `ch7465`, `ch7465_play` | `CH7465Driver` | `ch7465_xml` | `xml_payloads.parse_ch7465_xml` |
| `cm1000` | `CM1000Driver` | `cm1000_html_table`, `cm1000_javascript` | `html_rows.parse_cm1000_html_table`; `javascript.parse_cm1000_javascript` |
| `cm3000` | `CM3000Driver` | `cm3000_javascript` | `javascript.parse_cm3000_javascript` |
| `cm3500` | `CM3500Driver` | `cm3500_html` | `html_rows.parse_cm3500_html` |
| `cm8200` | `CM8200Driver` | `arris_html` | `html_rows.parse_arris_html` |
| `f3896lg` | `F3896LGDriver` | `f3896lg_rest_json` | `sagemcom.parse_f3896lg_rest_json` |
| `fritzbox` | `FritzBoxDriver` | `fritzbox_data_lua` | `fritzbox.parse_fritzbox_data_lua` |
| `generic` | `GenericDriver` | `generic_no_docsis` | `boundaries.parse_generic_no_docsis` |
| `hitron` | `HitronDriver` | `hitron_coda56_json` | `hitron.parse_hitron_coda56_json` |
| `hitron_coda_4680` | `HitronCoda4680Driver` | `hitron_coda4680_json` | `hitron.parse_hitron_coda4680_json` |
| `sagemcom` | `SagemcomDriver` | `sagemcom_xmo_json` | `sagemcom.parse_sagemcom_xmo_json` |
| `sb6141` | `SB6141Driver` | `sb6141_transposed_html` | `html_transposed.parse_sb6141_transposed_html` |
| `sb6183` | `SB6183Driver` | `sb6183_html` | `html_rows.parse_sb6183_html` |
| `sb6190` | `SB6190Driver` | `sb6190_html` | `html_rows.parse_sb6190_html` |
| `sercom_dm1000` | `SercomDM1000Driver` | `sercom_dm1000_json` | `sercom.parse_sercom_dm1000_json` |
| `surfboard` | `SurfboardDriver` | `arris_html`, `surfboard_hnap` | `html_rows.parse_arris_html`; `surfboard.parse_surfboard_hnap` |
| `tc4400` | `TC4400Driver` | `tc4400_html` | `html_rows.parse_tc4400_html` |
| `ultrahub7` | `UltraHub7Driver` | `ultrahub7_json` | `vodafone.parse_ultrahub7_json` |
| `vodafone_station` | `VodafoneStationDriver` | `vodafone_station_cga_json`, `vodafone_station_tg_embedded_json` | `vodafone.parse_vodafone_station_cga_json`; `vodafone.parse_vodafone_station_tg_embedded_json` |

### Driver Registry (`app/drivers/registry.py`)

Drivers are loaded by name through `load_driver(modem_type, url, user,
password)`. The registry maps type strings to fully qualified class paths for
lazy importing. `ch7465` and `ch7465_play` intentionally resolve to the same
class; the registry applies the Play firmware selection without creating a
second parser profile.

### Extension module state

Expand Down
242 changes: 26 additions & 216 deletions app/drivers/arris_html.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,229 +1,39 @@
"""Shared Arris HTML channel-table parser for DOCSight.

Parses the ``/cmconnectionstatus.html`` status page used by Arris cable
modems (CM8200A, SB8200 HTML fallback, and similar) into the standard
DOCSight channel data format.

The page contains two HTML tables:
- "Downstream Bonded Channels" (8 columns)
- "Upstream Bonded Channels" (7 columns)

DOCSIS version is inferred from modulation / channel type:
- DS: modulation "Other" = OFDM (3.1), anything else = SC-QAM (3.0)
- US: "OFDM" in type without "SC-QAM" = OFDMA (3.1), else SC-QAM (3.0)
"""
"""Compatibility shim for the established shared Arris bonded-table parser."""

from __future__ import annotations

import logging

from bs4 import BeautifulSoup, Tag

from ..types import DocsisDataFritz, RawChannel

log = logging.getLogger("docsis.arris_html")


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def parse_arris_channel_tables(html: str) -> DocsisDataFritz:
"""Parse Arris modem status page HTML into DOCSight channel format.

Returns::

{"channelDs": {"docsis30": [...], "docsis31": [...]},
"channelUs": {"docsis30": [...], "docsis31": [...]}}
"""
soup = BeautifulSoup(html, "html.parser")
ds_table, us_table = _find_channel_tables(soup)

ds30, ds31 = _parse_downstream(ds_table)
us30, us31 = _parse_upstream(us_table)

return {
"channelDs": {"docsis30": ds30, "docsis31": ds31},
"channelUs": {"docsis30": us30, "docsis31": us31},
}


# ---------------------------------------------------------------------------
# Table discovery
# ---------------------------------------------------------------------------

def _find_channel_tables(soup: BeautifulSoup) -> tuple:
"""Find downstream and upstream channel tables by header text.

Returns ``(ds_table, us_table)`` where either may be ``None``.
"""
ds_table = None
us_table = None

for table in soup.find_all("table"):
header = table.find("tr")
if not header:
continue
text = header.get_text(strip=True).lower()
if "downstream bonded" in text:
ds_table = table
elif "upstream bonded" in text:
us_table = table

return ds_table, us_table


# ---------------------------------------------------------------------------
# Row classification
# ---------------------------------------------------------------------------

def _is_header_row(row: Tag) -> bool:
"""True if *row* is a table title or column-header row (not data)."""
if row.find("th"):
return True
if row.find("strong"):
return True
return False


# ---------------------------------------------------------------------------
# Downstream parser
# ---------------------------------------------------------------------------

def _parse_downstream(table) -> tuple:
"""Parse downstream table into ``(docsis30, docsis31)`` channel lists.

Expected 8 columns per data row:
Channel ID | Lock Status | Modulation | Frequency |
Power | SNR/MER | Corrected | Uncorrectables
"""
ds30: list[dict] = []
ds31: list[dict] = []
if not table:
return ds30, ds31

for row in table.find_all("tr"):
if _is_header_row(row):
continue
cells = [td.get_text(strip=True) for td in row.find_all("td")]
if len(cells) < 8:
continue

lock_status = cells[1]
if lock_status != "Locked":
continue

try:
channel_id = int(cells[0])
modulation = cells[2]
frequency = _parse_freq_hz(cells[3])
power = _parse_value(cells[4])
snr = _parse_value(cells[5])
corrected = int(cells[6])
uncorrectables = int(cells[7])

channel: RawChannel = {
"channelID": channel_id,
"frequency": frequency,
"powerLevel": power,
"modulation": modulation,
"corrErrors": corrected,
"nonCorrErrors": uncorrectables,
}

if modulation == "Other":
# OFDM channel (DOCSIS 3.1)
channel["type"] = "OFDM"
channel["mer"] = snr
channel["mse"] = None
ds31.append(channel)
else:
# SC-QAM channel (DOCSIS 3.0)
channel["mer"] = snr
channel["mse"] = -snr if snr is not None else None
ds30.append(channel)
except (ValueError, TypeError, IndexError) as e:
log.warning("Failed to parse DS row: %s", e)

return ds30, ds31


# ---------------------------------------------------------------------------
# Upstream parser
# ---------------------------------------------------------------------------

def _parse_upstream(table) -> tuple:
"""Parse upstream table into ``(docsis30, docsis31)`` channel lists.

Expected 7 columns per data row:
Channel | Channel ID | Lock Status | US Channel Type |
Frequency | Width | Power
"""
us30: list[dict] = []
us31: list[dict] = []
if not table:
return us30, us31

for row in table.find_all("tr"):
if _is_header_row(row):
continue
cells = [td.get_text(strip=True) for td in row.find_all("td")]
if len(cells) < 7:
continue

lock_status = cells[2]
if lock_status != "Locked":
continue
from .formats.html_rows import (
_arris_tables as _find_channel_tables,
_optional_value as _parse_value,
parse_arris_downstream,
parse_arris_html,
parse_arris_upstream,
)
from .formats.primitives import hz_to_mhz as _parse_freq_hz

try:
channel_id = int(cells[1])
channel_type = cells[3]
frequency = _parse_freq_hz(cells[4])
power = _parse_value(cells[6])

channel: RawChannel = {
"channelID": channel_id,
"frequency": frequency,
"powerLevel": power,
"modulation": channel_type,
}
def parse_arris_channel_tables(html: str):
return parse_arris_html(html).value

if "OFDM" in channel_type and "SC-QAM" not in channel_type:
# OFDMA channel (DOCSIS 3.1)
channel["type"] = "OFDMA"
channel["multiplex"] = ""
us31.append(channel)
else:
# SC-QAM channel (DOCSIS 3.0)
channel["multiplex"] = "SC-QAM"
us30.append(channel)
except (ValueError, TypeError, IndexError) as e:
log.warning("Failed to parse US row: %s", e)

return us30, us31
def _parse_downstream(table):
return parse_arris_downstream(table).value


# ---------------------------------------------------------------------------
# Value helpers
# ---------------------------------------------------------------------------
def _parse_upstream(table):
return parse_arris_upstream(table).value

def _parse_freq_hz(freq_str: str) -> str:
"""Convert ``'795000000 Hz'`` to ``'795 MHz'``."""
from .utils import hz_to_mhz
return hz_to_mhz(freq_str)

def _is_header_row(row):
return bool(row.find("th") or row.find("strong"))

def _parse_value(val_str: str):
"""Parse ``'8.2 dBmV'`` or ``'43.0 dB'`` to float.

Note: Returns None (not 0.0) for empty/unparseable input, unlike
parse_number(). This preserves arris_html's existing behaviour where
None signals "value not present" vs 0.0 for "value is zero".
"""
if not val_str:
return None
parts = val_str.strip().split()
try:
return float(parts[0])
except (ValueError, IndexError):
return None
__all__ = [
"parse_arris_channel_tables",
"_find_channel_tables",
"_is_header_row",
"_parse_downstream",
"_parse_freq_hz",
"_parse_upstream",
"_parse_value",
]
Loading