1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143
|
"""Data transform module."""
from __future__ import annotations
from typing import Any
from asusrouter.modules.client import process_client
from asusrouter.modules.data import AsusDataState
from asusrouter.modules.ports import PortType
from asusrouter.tools.readers import readable_mac
# List of models with 6Ghz support
# and no 5Ghz2 support
MODEL_WITH_6GHZ = [
"RT-AXE95Q",
]
def transform_network(
data: dict[str, Any],
services: list[str] | None,
history: AsusDataState | None,
**kwargs: Any,
) -> dict[str, Any]:
"""Transform network data."""
# Check if the device has dualwan support
if not services:
return data
if "dualwan" not in services:
return data
network = data.copy()
# Add speed if not available - fix first empty value round
for interface in network:
for speed in ("rx_speed", "tx_speed"):
if speed not in network[interface]:
network[interface][speed] = 0.0
# Add usb network if not available
if "usb" not in network:
# Check history
usb_history = (
history.data.get("usb") if history and history.data else None
)
# Revert to history if available
if usb_history:
network["usb"] = usb_history
network["usb"]["rx_speed"] = 0.0
network["usb"]["tx_speed"] = 0.0
else:
network["usb"] = {
"rx": 0,
"tx": 0,
"rx_speed": 0.0,
"tx_speed": 0.0,
}
# Get the model if available
model = kwargs.get("model")
# Check if we have 5GHz2 available in the network data
if "5ghz2" in network:
# Check interfaces for 5Ghz2/6Ghz
support_5ghz2 = "5G-2" in services
support_6ghz = "wifi6e" in services
if (
support_5ghz2 is False and support_6ghz is True
) or model in MODEL_WITH_6GHZ:
# Rename 5Ghz2 to 6Ghz
network["6ghz"] = network.pop("5ghz2")
return network
def transform_clients(
data: dict[str, Any], history: AsusDataState | None, **kwargs: Any
) -> dict[str, Any]:
"""Transform clients data."""
clients = {}
for mac, client in data.items():
if readable_mac(mac):
# Check client history
client_history = (
history.data.get(mac) if history and history.data else None
)
# Process the client
clients[mac] = process_client(client, client_history, **kwargs)
return clients
def transform_cpu(
data: dict[str, dict[str, Any]],
) -> dict[str, Any]:
"""Transform cpu data."""
for info in data.values():
info.setdefault("usage", None)
return data
def transform_ethernet_ports(
data: dict[str, Any],
mac: str | None,
) -> dict[str, dict[str, Any]]:
"""Transform the legacy ethernet ports data to the new format."""
# Check if the first level of the dict is PortType enum
# If any other key is found, return the data as is
for data_key in data:
if not isinstance(data_key, PortType):
return data
# If mac is not available, return the data as is
if not mac:
return data
# Transform the data
return {mac: data}
def transform_wan(
data: dict[str, Any],
services: list[str] | None,
) -> dict[str, Any]:
"""Transform WAN data."""
wan = data.copy()
if not services:
return wan
service_keys = {"dualwan": "dualwan", "wanbonding": "aggregation"}
for service, key in service_keys.items():
if service not in services:
wan.pop(key, None)
return wan
|