From fdf617b0327a831da4fdfe7807b012ef365c5955 Mon Sep 17 00:00:00 2001 From: ChaoticByte Date: Thu, 27 Feb 2025 21:29:42 +0100 Subject: [PATCH] Move ping functionality into a PingableMixin, add WakeOnLanMixin and PingableWOLSystem --- dashboard/mixins.py | 44 +++++++++++++++++++++++++++++++++++++ dashboard/system.py | 53 ++++++++++++++++++++++++++++++--------------- dashboard/ui.py | 1 + 3 files changed, 80 insertions(+), 18 deletions(-) create mode 100644 dashboard/mixins.py diff --git a/dashboard/mixins.py b/dashboard/mixins.py new file mode 100644 index 0000000..baa677d --- /dev/null +++ b/dashboard/mixins.py @@ -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})") diff --git a/dashboard/system.py b/dashboard/system.py index 38c6a55..9ed6c3e 100644 --- a/dashboard/system.py +++ b/dashboard/system.py @@ -3,18 +3,15 @@ # additional libraries -import platform + import re -import subprocess import time 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 @@ -37,6 +34,9 @@ class SystemState(Enum): UNKNOWN = 2 +# base System + + class System: def __init__(self, name: str, description: str): @@ -60,27 +60,17 @@ class System: self.state_verbose = "" -# some basic systems +# Pingable System 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): super().__init__(name, description) 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): try: ok, stdout, stderr = self.ping() @@ -99,6 +89,33 @@ class PingableSystem(System): 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): def __init__(self, name, description, url: str, expected_status_code: int = 200, allow_self_signed_cert: bool = False): diff --git a/dashboard/ui.py b/dashboard/ui.py index df17b77..e870a64 100644 --- a/dashboard/ui.py +++ b/dashboard/ui.py @@ -1,5 +1,6 @@ # Copyright (c) 2025, Julian Müller (ChaoticByte) + import asyncio import datetime