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
|
"""WireGuard module."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from enum import IntEnum
import logging
from typing import Any
_LOGGER = logging.getLogger(__name__)
class AsusWireGuardClient(IntEnum):
"""Asus WireGuard client state."""
UNKNOWN = -999
OFF = 0
ON = 1
class AsusWireGuardServer(IntEnum):
"""Asus WireGuard server state."""
UNKNOWN = -999
OFF = 0
ON = 1
def _get_arguments(**kwargs: Any) -> int | None:
"""Get the arguments from kwargs."""
arguments = kwargs.get("arguments", {})
# Get the id from arguments
wlan_id = arguments.get("id") if arguments else kwargs.get("id")
if wlan_id is None:
wlan_id = 1
_LOGGER.debug("Using default id 1")
return wlan_id
async def set_state(
callback: Callable[..., Awaitable[bool]],
state: AsusWireGuardClient | AsusWireGuardServer,
**kwargs: Any,
) -> bool:
"""Set the WireGuard state."""
# Get the arguments
wlan_id = _get_arguments(**kwargs)
# WireGuard unit type (server or client)
wg_unit = "wgs" if isinstance(state, AsusWireGuardServer) else "wgc"
# Callback arguments
callback_arguments = {
"id": wlan_id,
f"{wg_unit}_enable": 1
if state in (AsusWireGuardClient.ON, AsusWireGuardServer.ON)
else 0,
f"{wg_unit}_unit": wlan_id,
}
# Get the expect_modify argument
expect_modify = kwargs.get("expect_modify", False)
# Get the correct service call
service_map: dict[Any, str] = {
(AsusWireGuardClient, AsusWireGuardClient.ON): f"start_wgc {wlan_id}",
(AsusWireGuardClient, AsusWireGuardClient.OFF): f"stop_wgc {wlan_id}",
(
AsusWireGuardServer,
AsusWireGuardServer.ON,
): "restart_wgs;restart_dnsmasq",
(
AsusWireGuardServer,
AsusWireGuardServer.OFF,
): "restart_wgs;restart_dnsmasq",
}
service = service_map.get((type(state), state))
if not service:
_LOGGER.debug("Unknown state %s", state)
return False
# Call the service
return await callback(
service=service,
arguments=callback_arguments,
apply=True,
expect_modify=expect_modify,
)
|
