From ba1ecaca8984d92c088640c544d290a9b72a55b3 Mon Sep 17 00:00:00 2001 From: David Huss <dh@atoav.com> Date: Mon, 2 Jun 2025 14:19:40 +0200 Subject: [PATCH] Propper handling of network disconnect --- pyproject.toml | 2 +- src/mediactl/devices/ahm16.py | 16 ++++++++++++++++ static/style.css | 13 +++++++++++++ 3 files changed, 30 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 6dfcfa7..267efa7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mediactl" -version = "0.1.0" +version = "0.1.1" description = "Add your description here" dependencies = [ "fastapi[standard]>=0.114.2", diff --git a/src/mediactl/devices/ahm16.py b/src/mediactl/devices/ahm16.py index d55aa61..14f03ca 100644 --- a/src/mediactl/devices/ahm16.py +++ b/src/mediactl/devices/ahm16.py @@ -1,5 +1,6 @@ import asyncio import aioping +import errno from datetime import datetime from typing import Union, List from result import Ok, Err, Result, is_ok, is_err # noqa: F401 @@ -96,6 +97,7 @@ class PowerState(mediactl.EnumState): on = "on" off = "off" unknown = "unknown" + disconnected = "disconnected" def is_on(self) -> bool: return self in [PowerState.on] @@ -103,6 +105,9 @@ class PowerState(mediactl.EnumState): def is_off(self) -> bool: return not self.is_on() + def is_disconnected(self) -> bool: + return self in [PowerState.disconnected] + class Channel(mediactl.WithLogger): def __init__(self, number: int, logger=None): @@ -233,6 +238,17 @@ class Ahm16(mediactl.WithLogger): else str(self.status["last-seen"]) ) self.log_error(f"AHM could not_be_reached. Last seen: {last}") + except OSError as e: + if e.errno == errno.ENETUNREACH: + self.status["power"] = PowerState.disconnected + last = "never" if self.status["last-seen"] is None else str(self.status["last-seen"]) + self.log_error(f"AHM Dante Network is down or AHM unreachable {e.errno}: {e.strerror}. Last seen: {last}") + else: + # Othernetwork‐level failure (no route, interface down, etc.) + self.status["power"] = PowerState.unknown + last = "never" if self.status["last-seen"] is None else str(self.status["last-seen"]) + self.log_error(f"AHM ping OSError {e.errno}: {e.strerror}. Last seen: {last}") + async def get_status(self, name=False): await self.ping() diff --git a/static/style.css b/static/style.css index d196833..a733bc4 100644 --- a/static/style.css +++ b/static/style.css @@ -34,6 +34,15 @@ main { display: none; } +@keyframes blink-red { + 0%, 45% { + background-color: var(--color-red); + } + 60%, 100% { + background-color: transparent; + } +} + header { width: 100%; border-bottom: 1px solid white; @@ -90,6 +99,10 @@ header { .warmup .status_text { color: var(--color-orange); } .off .led { background-color: var(--color-red); } .off .status_text { color: var(--color-red); } + .disconnected .led { + background-color: transparent; + animation: blink-red 1s infinite; + } } main.disconnected{ -- GitLab