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

Test and refactor client module #401

Merged
merged 1 commit into from
Nov 25, 2023
Merged
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
109 changes: 59 additions & 50 deletions asusrouter/modules/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

from dataclasses import dataclass
from datetime import datetime
from typing import Any, Optional
from typing import Any, Callable, Optional

from asusrouter.modules.connection import (
ConnectionState,
Expand Down Expand Up @@ -70,7 +70,10 @@ class AsusClientDescription:
vendor: Optional[str] = None


CLIENT_MAP = {
MapConverterType = Callable[..., Any] | list[Callable[..., Any]]
MapValueType = str | tuple[str] | tuple[str, MapConverterType]

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

CLIENT_MAP_DESCRIPTION = {
CLIENT_MAP_DESCRIPTION: dict[str, list[MapValueType]] = {
"name": [
("nickName"),
("name"),
Expand All @@ -134,7 +137,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 +163,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 +186,23 @@ class AsusClientDescription:
def process_client(
data: dict[str, Any], history: Optional[AsusClient] = None
) -> AsusClient:
"""Process client data."""
"""
Process client data.

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

Returns:
AsusClient: The processed client data.
"""

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

state = process_client_state(connection)
state: ConnectionState = process_client_state(connection)

# Clean disconnected client
if state != ConnectionState.CONNECTED:
Expand All @@ -201,42 +215,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 +263,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
Loading