Skip to content
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.

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

1.3.0 #418

Merged
merged 18 commits into from
Dec 3, 2023
Merged

1.3.0 #418

Show file tree
Hide file tree
Changes from all commits
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
20 changes: 11 additions & 9 deletions asusrouter/asusrouter.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,9 @@ async def _async_handle_reboot(self) -> None:
if led_state and led_state.data:
_LOGGER.debug("Restoring LED state")
await keep_state(
self.async_run_service, led_state.data["state"], self._identity
callback=self.async_run_service,
states=led_state.data["state"],
identity=self._identity,
)

# Reset the reboot flag
Expand Down Expand Up @@ -688,8 +690,8 @@ async def _async_check_state_dependency(self, state: AsusState) -> None:
async def async_set_state(
self,
state: AsusState,
arguments: Optional[dict[str, Any]] = None,
expect_modify: bool = False,
**kwargs: Any,
) -> bool:
"""Set the state."""

Expand All @@ -698,20 +700,20 @@ async def async_set_state(
_LOGGER.debug(
"Setting state `%s` with arguments `%s`. Expecting modify: `%s`",
state,
arguments,
kwargs,
expect_modify,
)

# Check dependencies
await self._async_check_state_dependency(state)

result = await set_state(
self.async_run_service,
state,
arguments,
expect_modify,
self._state,
self._identity,
callback=self.async_run_service,
state=state,
expect_modify=expect_modify,
router_state=self._state,
identity=self._identity,
**kwargs,
)

if result is True:
Expand Down
105 changes: 56 additions & 49 deletions asusrouter/modules/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
get_connection_type,
get_internet_mode,
)
from asusrouter.modules.const import MapValueType
from asusrouter.modules.ip_address import IPAddressType, read_ip_address_type
from asusrouter.modules.vendor import replace_vendor
from asusrouter.tools.converters import (
Expand Down Expand Up @@ -70,7 +71,7 @@ class AsusClientDescription:
vendor: Optional[str] = None


CLIENT_MAP = {
CLIENT_MAP: dict[str, list[MapValueType]] = {
"connected_since": [
("wlConnectTime", safe_time_from_delta),
],
Expand Down Expand Up @@ -120,7 +121,7 @@ class AsusClientDescription:
],
}

CLIENT_MAP_DESCRIPTION = {
CLIENT_MAP_DESCRIPTION: dict[str, list[MapValueType]] = {
"name": [
("nickName"),
("name"),
Expand All @@ -134,7 +135,7 @@ class AsusClientDescription:
],
}

CLIENT_MAP_CONNECTION = {
CLIENT_MAP_CONNECTION: dict[str, list[MapValueType]] = {
"type": [
("connection_type", get_connection_type),
("isWL", get_connection_type),
Expand All @@ -160,7 +161,7 @@ class AsusClientDescription:
],
}

CLIENT_MAP_CONNECTION_WLAN = {
CLIENT_MAP_CONNECTION_WLAN: dict[str, list[MapValueType]] = {
"guest_id": [
("guest"),
("isGN", safe_int),
Expand All @@ -183,12 +184,23 @@ class AsusClientDescription:
def process_client(
data: dict[str, Any], history: Optional[AsusClient] = None
) -> AsusClient:
"""Process client data."""
"""
Process client data.

description = process_client_description(data)
connection = process_client_connection(data)
Parameters:
data (dict[str, Any]): The client data to process.
history (Optional[AsusClient]): The previous state of the client, if any.

state = process_client_state(connection)
Returns:
AsusClient: The processed client data.
"""

description: AsusClientDescription = process_client_description(data)
connection: AsusClientConnection | AsusClientConnectionWlan = (
process_client_connection(data)
)

state: ConnectionState = process_client_state(connection)

# Clean disconnected client
if state != ConnectionState.CONNECTED:
Expand All @@ -201,42 +213,46 @@ def process_client(
return AsusClient(state=state, description=description, connection=connection)


def process_client_description(data: dict[str, Any]) -> AsusClientDescription:
"""Process client description data."""

description = AsusClientDescription()
def process_data(data: dict[str, Any], mapping: dict[str, Any], obj: Any) -> Any:
"""Process data based on a mapping and set attributes on an object."""

for key, value in CLIENT_MAP_DESCRIPTION.items():
# Go through all keys in mapping
for key, value in mapping.items():
for pair in value:
key_to_find, converter = safe_unpack_key(pair)
# Get the search key and converter(s)
key_to_find, converters = safe_unpack_key(pair)
# Get the value from the data
item = data.get(key_to_find)

# Process the value if it's an actual value
if item is not None and item != str():
setattr(
description,
key,
converter(item) if converter else item,
)
# Apply converters one by one if there are multiple
if isinstance(converters, list):
for converter in converters:
item = converter(item)
# Apply single converter
elif converters:
item = converters(item)

# Set the attribute on the object
setattr(obj, key, item)

# We found the value, no need to continue searching
break

return description
return obj


def process_client_description(data: dict[str, Any]) -> AsusClientDescription:
"""Process client description data."""

return process_data(data, CLIENT_MAP_DESCRIPTION, AsusClientDescription())


def process_client_connection(data: dict[str, Any]) -> AsusClientConnection:
"""Process client connection data."""

connection = AsusClientConnection()

for key, value in CLIENT_MAP_CONNECTION.items():
for pair in value:
key_to_find, converter = safe_unpack_key(pair)
item = data.get(key_to_find)
if item and item != str():
setattr(
connection,
key,
converter(item) if converter else item,
)
break
connection = process_data(data, CLIENT_MAP_CONNECTION, AsusClientConnection())

if not connection.type in (ConnectionType.WIRED, ConnectionType.DISCONNECTED):
connection = process_client_connection_wlan(data, connection)
Expand All @@ -245,28 +261,19 @@ def process_client_connection(data: dict[str, Any]) -> AsusClientConnection:


def process_client_connection_wlan(
data: dict[str, Any], connection: AsusClientConnection
data: dict[str, Any], base_connection: AsusClientConnection
) -> AsusClientConnectionWlan:
"""Process WLAN client connection data."""

connection = AsusClientConnectionWlan(**connection.__dict__)

for key, value in CLIENT_MAP_CONNECTION_WLAN.items():
for pair in value:
key_to_find, converter = safe_unpack_key(pair)
item = data.get(key_to_find)
if item and item != str():
setattr(
connection,
key,
converter(item) if converter else item,
)
break
wlan_connection = AsusClientConnectionWlan(**base_connection.__dict__)
wlan_connection = process_data(data, CLIENT_MAP_CONNECTION_WLAN, wlan_connection)

# Mark `guest` attribute if `guest_id` is non-zero
connection.guest = connection.guest_id != 0 and connection.guest_id is not None
wlan_connection.guest = (
wlan_connection.guest_id != 0 and wlan_connection.guest_id is not None
)

return connection
return wlan_connection


def process_client_state(
Expand Down
14 changes: 14 additions & 0 deletions asusrouter/modules/const.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
"""Constants for modules."""

from __future__ import annotations

from typing import Any, Callable

# A converter type
MapConverterType = Callable[..., Any] | list[Callable[..., Any]]

# A map value type - this value should be found and converted
MapValueType = str | tuple[str] | tuple[str, MapConverterType]

# A map replace type - this value should be found, renamed and converted
MapReplaceType = str | tuple[str] | tuple[str, str] | tuple[str, str, MapConverterType]
2 changes: 2 additions & 0 deletions asusrouter/modules/data_finder.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
MAP_WIREGUARD_SERVER,
)
from asusrouter.modules.parental_control import (
KEY_PARENTAL_CONTROL_BLOCK_ALL,
KEY_PARENTAL_CONTROL_MAC,
KEY_PARENTAL_CONTROL_NAME,
KEY_PARENTAL_CONTROL_STATE,
Expand Down Expand Up @@ -110,6 +111,7 @@ def __init__(
for key, _, _ in [converters.safe_unpack_keys(element)]
],
"parental_control": [
KEY_PARENTAL_CONTROL_BLOCK_ALL,
KEY_PARENTAL_CONTROL_MAC,
KEY_PARENTAL_CONTROL_NAME,
KEY_PARENTAL_CONTROL_STATE,
Expand Down
11 changes: 11 additions & 0 deletions asusrouter/modules/data_transform.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,17 @@
"rx_speed": 0.0,
"tx_speed": 0.0,
}

# Check if we have 5GHz2 available in the network data
if "5ghz2" in network:

Check warning on line 50 in asusrouter/modules/data_transform.py

View check run for this annotation

Codecov / codecov/patch

asusrouter/modules/data_transform.py#L50

Added line #L50 was not covered by tests
# Check interfaces for 5Ghz2/6Ghz
support_5ghz2 = "5G-2" in services
support_6ghz = "wifi6e" in services

Check warning on line 53 in asusrouter/modules/data_transform.py

View check run for this annotation

Codecov / codecov/patch

asusrouter/modules/data_transform.py#L52-L53

Added lines #L52 - L53 were not covered by tests

if support_5ghz2 is False and support_6ghz is True:

Check warning on line 55 in asusrouter/modules/data_transform.py

View check run for this annotation

Codecov / codecov/patch

asusrouter/modules/data_transform.py#L55

Added line #L55 was not covered by tests
# Rename 5Ghz2 to 6Ghz
network["6ghz"] = network.pop("5ghz2")

Check warning on line 57 in asusrouter/modules/data_transform.py

View check run for this annotation

Codecov / codecov/patch

asusrouter/modules/data_transform.py#L57

Added line #L57 was not covered by tests

return network


Expand Down
15 changes: 15 additions & 0 deletions asusrouter/modules/endpoint/hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,13 @@
from asusrouter.modules.endpoint.error import AccessError
from asusrouter.modules.led import AsusLED
from asusrouter.modules.parental_control import (
KEY_PARENTAL_CONTROL_BLOCK_ALL,
KEY_PARENTAL_CONTROL_MAC,
KEY_PARENTAL_CONTROL_STATE,
KEY_PARENTAL_CONTROL_TYPE,
MAP_PARENTAL_CONTROL_ITEM,
MAP_PARENTAL_CONTROL_TYPE,
AsusBlockAll,
AsusParentalControl,
ParentalControlRule,
)
Expand Down Expand Up @@ -335,6 +337,11 @@ def process_parental_control(data: dict[str, Any]) -> dict[str, Any]:
safe_int(data.get(KEY_PARENTAL_CONTROL_STATE), default=-999)
)

# Block all
parental_control["block_all"] = AsusBlockAll(
safe_int(data.get(KEY_PARENTAL_CONTROL_BLOCK_ALL), default=-999)
)

# Rules
rules = {}

Expand Down Expand Up @@ -534,8 +541,12 @@ def process_vpnc(data: dict[str, Any]) -> Tuple[dict[AsusVPNType, dict[int, Any]

# Re-sort the data by VPN type / id
vpn: dict[AsusVPNType, dict[int, Any]] = {
AsusVPNType.L2TP: {},
AsusVPNType.OPENVPN: {},
AsusVPNType.PPTP: {},
AsusVPNType.SURFSHARK: {},
AsusVPNType.WIREGUARD: {},
AsusVPNType.UNKNOWN: {},
}

for vpnc_id, info in vpnc.items():
Expand All @@ -561,6 +572,10 @@ def process_vpnc(data: dict[str, Any]) -> Tuple[dict[AsusVPNType, dict[int, Any]
"error": AccessError.NO_ERROR,
}

# Remove UNKNOWN VPN type if it's empty
if not vpn[AsusVPNType.UNKNOWN]:
vpn.pop(AsusVPNType.UNKNOWN, None)

return vpn, vpnc_clientlist


Expand Down
17 changes: 8 additions & 9 deletions asusrouter/modules/led.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,9 @@
from __future__ import annotations

from enum import IntEnum
from typing import Any, Awaitable, Callable, Optional
from typing import Any, Awaitable, Callable

from asusrouter.modules.endpoint import Endpoint
from asusrouter.modules.identity import AsusDevice


class AsusLED(IntEnum):
Expand All @@ -20,9 +19,7 @@ class AsusLED(IntEnum):
async def set_state(
callback: Callable[..., Awaitable[bool]],
state: AsusLED,
arguments: Optional[dict[str, Any]] = None,
expect_modify: bool = False,
_: Optional[dict[Any, Any]] = None,
**kwargs: Any,
) -> bool:
"""Set the LED state."""

Expand All @@ -34,17 +31,19 @@ async def set_state(
service="start_ctrl_led",
arguments=arguments,
apply=True,
expect_modify=expect_modify,
expect_modify=kwargs.get("expect_modify", False),
)


async def keep_state(
callback: Callable[..., Awaitable[bool]],
state: AsusLED = AsusLED.ON,
identity: Optional[AsusDevice] = None,
**kwargs: Any,
) -> bool:
"""Keep the LED state."""

identity = kwargs.get("identity")

# Check if identity is available and if endpoints are defined
if identity is None or not identity.endpoints:
return False
Expand All @@ -57,7 +56,7 @@ async def keep_state(
return False

# Toggle the LED
await set_state(callback, AsusLED.ON)
await set_state(callback, AsusLED.OFF)
await set_state(callback, AsusLED.ON, **kwargs)
await set_state(callback, AsusLED.OFF, **kwargs)

return True
Loading