Move ping functionality into a PingableMixin, add WakeOnLanMixin and PingableWOLSystem
This commit is contained in:
parent
60886b4d0f
commit
fdf617b032
3 changed files with 80 additions and 18 deletions
44
dashboard/mixins.py
Normal file
44
dashboard/mixins.py
Normal file
|
@ -0,0 +1,44 @@
|
||||||
|
# Copyright (c) 2025, Julian Müller (ChaoticByte)
|
||||||
|
|
||||||
|
|
||||||
|
import platform
|
||||||
|
import socket
|
||||||
|
import subprocess
|
||||||
|
|
||||||
|
from typing import Tuple, List
|
||||||
|
|
||||||
|
from nicegui import ui
|
||||||
|
|
||||||
|
|
||||||
|
class PingableMixin:
|
||||||
|
|
||||||
|
def ping(self) -> Tuple[bool, str, str]:
|
||||||
|
''' requires the following attributes:
|
||||||
|
- self.host: str Host ip address
|
||||||
|
'''
|
||||||
|
if platform.system().lower() == "windows": p = "-n"
|
||||||
|
else: p = "-c"
|
||||||
|
s = subprocess.run(
|
||||||
|
["ping", p, '1', self.host],
|
||||||
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||||
|
env={"LC_ALL": "C"} # don't translate
|
||||||
|
)
|
||||||
|
return s.returncode == 0, s.stdout.decode(), s.stderr.decode()
|
||||||
|
|
||||||
|
|
||||||
|
class WakeOnLanMixin:
|
||||||
|
|
||||||
|
def wakeonlan(self):
|
||||||
|
''' requires the following attributes:
|
||||||
|
- self.name: str System.name
|
||||||
|
- self.host_mac: str host mac address
|
||||||
|
'''
|
||||||
|
assert hasattr(self, "host_mac")
|
||||||
|
assert hasattr(self, "name")
|
||||||
|
host_mac_bin = bytes.fromhex(self.host_mac.replace(':', '').replace('-', '').lower())
|
||||||
|
magic_packet = (b'\xff' * 6) + (host_mac_bin * 16)
|
||||||
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
|
||||||
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
||||||
|
for port in [7, 9]: # we send to both port 7 and 9 to be compatible with most of the systems
|
||||||
|
s.sendto(magic_packet, ("255.255.255.255", port))
|
||||||
|
ui.notify(f"Magic packet sent to wake up '{self.name}' ({self.host_mac})")
|
|
@ -3,18 +3,15 @@
|
||||||
|
|
||||||
# additional libraries
|
# additional libraries
|
||||||
|
|
||||||
import platform
|
|
||||||
import re
|
import re
|
||||||
import subprocess
|
|
||||||
import time
|
import time
|
||||||
|
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Tuple, List
|
from typing import List
|
||||||
|
|
||||||
import requests, urllib3
|
from .mixins import PingableMixin, WakeOnLanMixin
|
||||||
|
|
||||||
# don't need the warning, ssl verification needs to be disabled explicitly
|
|
||||||
urllib3.disable_warnings(category=urllib3.connectionpool.InsecureRequestWarning)
|
|
||||||
|
|
||||||
# base classes and types and stuff
|
# base classes and types and stuff
|
||||||
|
|
||||||
|
@ -37,6 +34,9 @@ class SystemState(Enum):
|
||||||
UNKNOWN = 2
|
UNKNOWN = 2
|
||||||
|
|
||||||
|
|
||||||
|
# base System
|
||||||
|
|
||||||
|
|
||||||
class System:
|
class System:
|
||||||
|
|
||||||
def __init__(self, name: str, description: str):
|
def __init__(self, name: str, description: str):
|
||||||
|
@ -60,27 +60,17 @@ class System:
|
||||||
self.state_verbose = ""
|
self.state_verbose = ""
|
||||||
|
|
||||||
|
|
||||||
# some basic systems
|
# Pingable System
|
||||||
|
|
||||||
|
|
||||||
ping_time_regex = re.compile(r".*ttl=\d+ time=((?:\d+\.)?\d+ ms).*")
|
ping_time_regex = re.compile(r".*ttl=\d+ time=((?:\d+\.)?\d+ ms).*")
|
||||||
|
|
||||||
class PingableSystem(System):
|
class PingableSystem(PingableMixin, System):
|
||||||
|
|
||||||
def __init__(self, name, description, host: str):
|
def __init__(self, name, description, host: str):
|
||||||
super().__init__(name, description)
|
super().__init__(name, description)
|
||||||
self.host = host
|
self.host = host
|
||||||
|
|
||||||
def ping(self) -> Tuple[bool, str, str]:
|
|
||||||
if platform.system().lower() == "windows": p = "-n"
|
|
||||||
else: p = "-c"
|
|
||||||
s = subprocess.run(
|
|
||||||
["ping", p, '1', self.host],
|
|
||||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
||||||
env={"LC_ALL": "C"} # don't translate
|
|
||||||
)
|
|
||||||
return s.returncode == 0, s.stdout.decode(), s.stderr.decode()
|
|
||||||
|
|
||||||
def update_state(self):
|
def update_state(self):
|
||||||
try:
|
try:
|
||||||
ok, stdout, stderr = self.ping()
|
ok, stdout, stderr = self.ping()
|
||||||
|
@ -99,6 +89,33 @@ class PingableSystem(System):
|
||||||
self.state_verbose = f"Exception: {str(e)}"
|
self.state_verbose = f"Exception: {str(e)}"
|
||||||
|
|
||||||
|
|
||||||
|
# Pingable + WakeOnLan System
|
||||||
|
|
||||||
|
|
||||||
|
class PingableWOLSystem(WakeOnLanMixin, PingableSystem):
|
||||||
|
|
||||||
|
'''Pingable System with Wake On LAN'''
|
||||||
|
|
||||||
|
def __init__(self, name, description, host_ip: str, host_mac: str):
|
||||||
|
super().__init__(name, description, host_ip)
|
||||||
|
self.host_mac = host_mac
|
||||||
|
|
||||||
|
def get_actions(self) -> List[Action]:
|
||||||
|
actions = []
|
||||||
|
if self.state != SystemState.OK:
|
||||||
|
actions.append(Action("Wake On LAN", self.wakeonlan))
|
||||||
|
return actions
|
||||||
|
|
||||||
|
|
||||||
|
# HTTP Server System
|
||||||
|
|
||||||
|
|
||||||
|
import requests, urllib3
|
||||||
|
|
||||||
|
# don't need the warning, bc. ssl verification needs to be disabled explicitly
|
||||||
|
urllib3.disable_warnings(category=urllib3.connectionpool.InsecureRequestWarning)
|
||||||
|
|
||||||
|
|
||||||
class HTTPServer(System):
|
class HTTPServer(System):
|
||||||
|
|
||||||
def __init__(self, name, description, url: str, expected_status_code: int = 200, allow_self_signed_cert: bool = False):
|
def __init__(self, name, description, url: str, expected_status_code: int = 200, allow_self_signed_cert: bool = False):
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
# Copyright (c) 2025, Julian Müller (ChaoticByte)
|
# Copyright (c) 2025, Julian Müller (ChaoticByte)
|
||||||
|
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import datetime
|
import datetime
|
||||||
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue