File: data_transform.py

package info (click to toggle)
python-asusrouter 1.21.3-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 1,856 kB
  • sloc: python: 20,497; makefile: 3
file content (143 lines) | stat: -rw-r--r-- 3,704 bytes parent folder | download
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