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
|
# ========================================================================== #
# #
# KVMD - The main PiKVM daemon. #
# #
# Copyright (C) 2018-2023 Maxim Devaev <[email protected]> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
import socket
import functools
from typing import Callable
from typing import Any
from ...logging import get_logger
from ... import aiotools
from ...yamlconf import Option
from ...validators.net import valid_ip
from ...validators.net import valid_port
from ...validators.net import valid_mac
from . import GpioDriverOfflineError
from . import BaseUserGpioDriver
# =====
class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attributes
def __init__( # pylint: disable=super-init-not-called
self,
instance_name: str,
notifier: aiotools.AioNotifier,
ip: str,
port: int,
mac: str,
) -> None:
super().__init__(instance_name, notifier)
self.__ip = ip
self.__port = port
self.__mac = mac
@classmethod
def get_plugin_options(cls) -> dict:
return {
"ip": Option("255.255.255.255", type=functools.partial(valid_ip, v6=False)),
"port": Option(9, type=valid_port),
"mac": Option("", type=valid_mac, if_empty=""),
}
@classmethod
def get_pin_validator(cls) -> Callable[[Any], Any]:
return str
async def read(self, pin: str) -> bool:
_ = pin
return False
async def write(self, pin: str, state: bool) -> None:
_ = pin
if not state:
return
sock: (socket.socket | None) = None
try:
# TODO: IPv6 support: http://lists.cluenet.de/pipermail/ipv6-ops/2014-September/010139.html
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.connect((self.__ip, self.__port))
sock.send(bytes.fromhex("FF" * 6 + self.__mac.replace(":", "") * 16))
except Exception:
get_logger(0).exception("Can't send Wake-on-LAN packet via %s to %s", self, self.__mac)
raise GpioDriverOfflineError(self)
finally:
if sock:
try:
sock.close()
except Exception:
pass
def __str__(self) -> str:
return f"WakeOnLan({self._instance_name})"
__repr__ = __str__
|
