Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 0 additions & 10 deletions airos/airos8.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@ def __init__(
self._status_cgi_url = f"{self.base_url}/status.cgi" # AirOS 8
self._stakick_cgi_url = f"{self.base_url}/stakick.cgi" # AirOS 8
self.current_csrf_token = None
self.warnings_cache = []

self._use_json_for_login_post = False

Expand Down Expand Up @@ -214,15 +213,6 @@ async def status(self) -> AirOSData:
logger.exception("Failed to deserialize AirOS data")
raise KeyDataMissingError from err

# Show new enums detected, once after (each) startup
if airos_data.warnings:
for field_name, messages in airos_data.warnings.items():
for msg in messages:
log = f"AirOS data warning for field '{field_name}': {msg}"
if log not in self.warnings_cache:
self.warnings_cache.append(log)
logger.warning(log)

return airos_data
except json.JSONDecodeError:
logger.exception(
Expand Down
108 changes: 50 additions & 58 deletions airos/airos8data.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,35 @@
"""Provide mashumaro data object for AirOSData."""

from dataclasses import dataclass, field
from dataclasses import dataclass
from enum import Enum
import logging
from typing import Any

from mashumaro import DataClassDictMixin

logger = logging.getLogger(__name__)


def _check_and_log_unknown_enum_value(
data_dict: dict[str, Any],
key: str,
enum_class: type[Enum],
dataclass_name: str,
field_name: str,
) -> None:
"""Clean unsupported parameters with logging."""
value = data_dict.get(key)
if value is not None and isinstance(value, str):
if value not in [e.value for e in enum_class]:
logging.warning(
"Unknown value '%s' for %s.%s. Please report at "
"https://github.com/CoMPaTech/python-airos/issues so we can add support.",
value,
dataclass_name,
field_name,
)
del data_dict[key]


class IeeeMode(Enum):
"""Enum definition."""
Expand Down Expand Up @@ -58,14 +82,20 @@ class Host:
timestamp: int
fwversion: str
devmodel: str
netrole: NetRole | str
netrole: NetRole
loadavg: float
totalram: int
freeram: int
temperature: int
cpuload: float
height: int

@classmethod
def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
"""Pre-deserialize hook for Host."""
_check_and_log_unknown_enum_value(d, "netrole", NetRole, "Host", "netrole")
return d


@dataclass
class Services:
Expand Down Expand Up @@ -193,7 +223,7 @@ class Remote:
totalram: int
freeram: int
netrole: str
mode: WirelessMode | str # Allow non-breaking future expansion
mode: WirelessMode
sys_id: str
tx_throughput: int
rx_throughput: int
Expand Down Expand Up @@ -266,8 +296,8 @@ class Wireless:
"""Leaf definition."""

essid: str
mode: WirelessMode | str # Allow non-breaking expansion
ieeemode: IeeeMode | str # Allow non-breaking expansion
mode: WirelessMode
ieeemode: IeeeMode
band: int
compat_11n: int
hide_essid: int
Expand All @@ -277,7 +307,7 @@ class Wireless:
center1_freq: int
dfs: int
distance: int
security: Security | str # Allow non-breaking expansion
security: Security
noisef: int
txpower: int
aprepeater: bool
Expand All @@ -300,6 +330,18 @@ class Wireless:
sta: list[Station]
sta_disconnected: list[Any]

@classmethod
def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
"""Pre-deserialize hook for Wireless."""
_check_and_log_unknown_enum_value(d, "mode", WirelessMode, "Wireless", "mode")
_check_and_log_unknown_enum_value(
d, "ieeemode", IeeeMode, "Wireless", "ieeemode"
)
_check_and_log_unknown_enum_value(
d, "security", Security, "Wireless", "security"
)
return d


@dataclass
class InterfaceStatus:
Expand Down Expand Up @@ -375,57 +417,7 @@ class AirOSData(DataClassDictMixin):
portfw: bool
wireless: Wireless
interfaces: list[Interface]
provmode: (
ProvisioningMode | str | dict[str, Any] | list[Any] | Any
) # If it can be populated, define its fields
ntpclient: (
NtpClient | str | dict[str, Any] | list[Any] | Any
) # If it can be populated, define its fields
provmode: Any
ntpclient: Any
unms: UnmsStatus
gps: GPSMain
warnings: dict[str, list[str]] = field(default_factory=dict, init=False)

@classmethod
def __post_deserialize__(cls, airos_object: "AirOSData") -> "AirOSData":
"""Validate after deserialization."""
airos_object.check_for_warnings()
return airos_object

def check_for_warnings(self):
"""Validate unions for unknown fields."""
# Check wireless mode
if isinstance(self.wireless.mode, str):
self.add_warning(
"wireless", f"Unknown (new) wireless mode: '{self.wireless.mode}'"
)

# Check host netrole
if isinstance(self.host.netrole, str):
self.add_warning(
"host", f"Unknown (new) network role: '{self.host.netrole}'"
)

# Check wireless IEEE mode
if isinstance(self.wireless.ieeemode, str):
self.add_warning(
"wireless", f"Unknown (new) IEEE mode: '{self.wireless.ieeemode}'"
)

# Check wireless security
if isinstance(self.wireless.security, str):
self.add_warning(
"wireless", f"Unknown (new) security type: '{self.wireless.security}'"
)
# Check station remote modes
for i, station in enumerate(self.wireless.sta):
if hasattr(station.remote, "mode") and isinstance(station.remote.mode, str):
self.add_warning(
f"wireless.sta[{i}].remote",
f"Unknown (new) remote mode: '{station.remote.mode}', please report to the CODEOWNERS for inclusion",
)

def add_warning(self, field_name: str, message: str):
"""Insert warnings into the dictionary on unknown field data."""
if field_name not in self.warnings:
self.warnings[field_name] = []
self.warnings[field_name].append(message)