# simclient/simclient.py from logging.handlers import RotatingFileHandler import time import uuid import json import socket import hashlib import paho.mqtt.client as mqtt import os import shutil import re import platform import logging from dotenv import load_dotenv import requests import base64 from datetime import datetime import threading from urllib.parse import urlsplit, urlunsplit, unquote # ENV laden - support both container and native development env_paths = [ "/workspace/simclient/.env", # Container path os.path.join(os.path.dirname(__file__), ".env"), # Same directory os.path.join(os.path.expanduser("~"), "infoscreen-dev", ".env"), # Development path ] for env_path in env_paths: if os.path.exists(env_path): load_dotenv(env_path) break def _env_int(name, default): """Parse an int from environment variable, tolerating inline comments. Examples: - "10 # seconds" -> 10 - " 300ms" -> 300 - invalid or empty -> default """ raw = os.getenv(name) if raw is None or str(raw).strip() == "": return default try: # Remove inline comments sanitized = str(raw).split('#', 1)[0].strip() # Extract first integer occurrence m = re.search(r"-?\d+", sanitized) if m: return int(m.group(0)) except Exception: pass return default def _env_bool(name, default=False): raw = os.getenv(name) if raw is None: return default return str(raw).strip().lower() in ("1", "true", "yes", "on") def _env_host(name, default): """Parse a hostname/IP from env, stripping inline comments and whitespace. Example: "192.168.1.10 # comment" -> "192.168.1.10" """ raw = os.getenv(name) if raw is None: return default # Remove inline comments and extra spaces sanitized = str(raw).split('#', 1)[0].strip() # If any whitespace remains, take the first token as host if not sanitized: return default return sanitized.split()[0] def _env_str_clean(name, default=""): """Parse a generic string from env, removing inline comments and trimming. Returns the first whitespace-delimited token to avoid accidental comment tails. """ raw = os.getenv(name) if raw is None: return default sanitized = str(raw).split('#', 1)[0].strip() if not sanitized: return default return sanitized.split()[0] # Konfiguration aus ENV ENV = os.getenv("ENV", "development") HEARTBEAT_INTERVAL = _env_int("HEARTBEAT_INTERVAL", 5 if ENV == "development" else 60) SCREENSHOT_INTERVAL = _env_int("SCREENSHOT_INTERVAL", 30 if ENV == "development" else 300) LOG_LEVEL = os.getenv("LOG_LEVEL", "DEBUG" if ENV == "development" else "INFO") # Default to localhost in development, 'mqtt' (Docker compose service) otherwise MQTT_BROKER = _env_host("MQTT_BROKER", "localhost" if ENV == "development" else "mqtt") MQTT_PORT = _env_int("MQTT_PORT", 1883) DEBUG_MODE = _env_bool("DEBUG_MODE", ENV == "development") MQTT_BROKER_FALLBACKS = [] _fallbacks_raw = os.getenv("MQTT_BROKER_FALLBACKS", "") if _fallbacks_raw: for item in _fallbacks_raw.split(","): host = item.split('#', 1)[0].strip() if host: # Only take the first whitespace-delimited token MQTT_BROKER_FALLBACKS.append(host.split()[0]) # File server/API configuration # Defaults: use same host as MQTT broker, port 8000, http scheme FILE_SERVER_BASE_URL = _env_str_clean("FILE_SERVER_BASE_URL", "") _scheme_raw = _env_str_clean("FILE_SERVER_SCHEME", "http").lower() FILE_SERVER_SCHEME = _scheme_raw if _scheme_raw in ("http", "https") else "http" FILE_SERVER_HOST = _env_host("FILE_SERVER_HOST", MQTT_BROKER) FILE_SERVER_PORT = _env_int("FILE_SERVER_PORT", 8000) # Logging-Konfiguration LOG_PATH = os.path.join(os.path.dirname(__file__), "simclient.log") os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True) log_handlers = [] log_handlers.append(RotatingFileHandler( LOG_PATH, maxBytes=2*1024*1024, backupCount=5, encoding="utf-8")) if DEBUG_MODE: log_handlers.append(logging.StreamHandler()) logging.basicConfig( level=getattr(logging, LOG_LEVEL.upper(), logging.INFO), format="%(asctime)s [%(levelname)s] %(message)s", handlers=log_handlers ) discovered = False def save_event_to_json(event_data): """Speichert eine Event-Nachricht in der Datei current_event.json This function preserves ALL fields from the incoming event data, including scheduler-specific fields like: - page_progress: Current page/slide progress tracking - auto_progress: Auto-progression state - And any other fields sent by the scheduler """ try: json_path = os.path.join(os.path.dirname(__file__), "current_event.json") with open(json_path, "w", encoding="utf-8") as f: json.dump(event_data, f, ensure_ascii=False, indent=2) logging.info(f"Event message saved to {json_path}") # Log if scheduler-specific fields are present if isinstance(event_data, list): for idx, event in enumerate(event_data): if isinstance(event, dict): if 'page_progress' in event: logging.debug(f"Event {idx}: page_progress = {event['page_progress']}") if 'auto_progress' in event: logging.debug(f"Event {idx}: auto_progress = {event['auto_progress']}") elif isinstance(event_data, dict): if 'page_progress' in event_data: logging.debug(f"Event page_progress = {event_data['page_progress']}") if 'auto_progress' in event_data: logging.debug(f"Event auto_progress = {event_data['auto_progress']}") except Exception as e: logging.error(f"Error saving event message: {e}") def delete_event_file(): """Löscht die current_event.json Datei wenn kein Event aktiv ist""" try: json_path = os.path.join(os.path.dirname(__file__), "current_event.json") if os.path.exists(json_path): # Copy to last_event.json first so we keep a record of the last event try: last_path = os.path.join(os.path.dirname(__file__), "last_event.json") # Use atomic replace: write to temp then replace tmp_path = last_path + ".tmp" shutil.copyfile(json_path, tmp_path) os.replace(tmp_path, last_path) logging.info(f"Copied {json_path} to {last_path} (last event)") except Exception as e: logging.warning(f"Could not copy current_event.json to last_event.json: {e}") os.remove(json_path) logging.info(f"Event file {json_path} deleted - no active event") except Exception as e: logging.error(f"Error deleting event file: {e}") def is_empty_event(event_data): """Prüft ob eine Event-Nachricht bedeutet, dass kein Event aktiv ist""" if event_data is None: return True # Verschiedene Möglichkeiten für "kein Event": # 1. Leeres Dictionary if not event_data: return True # 2. Explizite "null" oder "empty" Werte if isinstance(event_data, dict): # Event ist null/None if event_data.get("event") is None or event_data.get("event") == "null": return True # Event ist explizit als "empty" oder "none" markiert if str(event_data.get("event", "")).lower() in ["empty", "none", ""]: return True # Status zeigt an dass kein Event aktiv ist status = str(event_data.get("status", "")).lower() if status in ["inactive", "none", "empty", "cleared"]: return True # 3. String-basierte Events if isinstance(event_data, str) and event_data.lower() in ["null", "none", "empty", ""]: return True return False def on_message(client, userdata, msg, properties=None): global discovered logging.info(f"Received: {msg.topic} {msg.payload.decode()}") if msg.topic.startswith("infoscreen/events/"): event_payload = msg.payload.decode() logging.info(f"Event message from scheduler received: {event_payload}") try: event_data = json.loads(event_payload) if is_empty_event(event_data): logging.info("No active event - deleting event file") delete_event_file() else: save_event_to_json(event_data) # Check if event_data is a list or a dictionary if isinstance(event_data, list): for event in event_data: presentation_files = event.get("presentation", {}).get("files", []) for file in presentation_files: file_url = file.get("url") if file_url: download_presentation_file(file_url) elif isinstance(event_data, dict): presentation_files = event_data.get("presentation", {}).get("files", []) for file in presentation_files: file_url = file.get("url") if file_url: download_presentation_file(file_url) except json.JSONDecodeError as e: logging.error(f"Invalid JSON in event message: {e}") if event_payload.strip().lower() in ["null", "none", "empty", ""]: logging.info("Empty event message received - deleting event file") delete_event_file() else: event_data = {"raw_message": event_payload, "error": "Invalid JSON format"} save_event_to_json(event_data) if msg.topic.endswith("/discovery_ack"): discovered = True logging.info("Discovery ACK received. Starting heartbeat.") def get_mac_addresses(): macs = set() try: for root, dirs, files in os.walk('/sys/class/net/'): for iface in dirs: try: with open(f'/sys/class/net/{iface}/address') as f: mac = f.read().strip() if mac and mac != '00:00:00:00:00:00': macs.add(mac) except Exception: continue break except Exception: pass return sorted(macs) def get_board_serial(): # Raspberry Pi: /proc/cpuinfo, andere: /sys/class/dmi/id/product_serial serial = None try: with open('/proc/cpuinfo') as f: for line in f: if line.lower().startswith('serial'): serial = line.split(':')[1].strip() break except Exception: pass if not serial: try: with open('/sys/class/dmi/id/product_serial') as f: serial = f.read().strip() except Exception: pass return serial or "unknown" def get_ip(): # Versucht, die lokale IP zu ermitteln (nicht 127.0.0.1) try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return "unknown" def get_hardware_token(): serial = get_board_serial() macs = get_mac_addresses() token_raw = serial + "_" + "_".join(macs) # Hashen für Datenschutz token_hash = hashlib.sha256(token_raw.encode()).hexdigest() return token_hash def get_model(): # Versucht, das Modell auszulesen (z.B. Raspberry Pi, PC, etc.) try: if os.path.exists('/proc/device-tree/model'): with open('/proc/device-tree/model') as f: return f.read().strip() elif os.path.exists('/sys/class/dmi/id/product_name'): with open('/sys/class/dmi/id/product_name') as f: return f.read().strip() except Exception: pass return "unknown" SOFTWARE_VERSION = "1.0.0" # Optional: Anpassen bei neuen Releases def send_discovery(client, client_id, hardware_token, ip_addr): macs = get_mac_addresses() discovery_msg = { "uuid": client_id, "hardware_token": hardware_token, "ip": ip_addr, "type": "infoscreen", "hostname": socket.gethostname(), "os_version": platform.platform(), "software_version": SOFTWARE_VERSION, "macs": macs, "model": get_model(), } client.publish("infoscreen/discovery", json.dumps(discovery_msg)) logging.info(f"Discovery message sent: {discovery_msg}") def get_persistent_uuid(uuid_path=None): if uuid_path is None: uuid_path = os.path.join(os.path.dirname(__file__), "config", "client_uuid.txt") # Prüfe, ob die Datei existiert if os.path.exists(uuid_path): with open(uuid_path, "r") as f: return f.read().strip() # Generiere neue UUID und speichere sie new_uuid = str(uuid.uuid4()) os.makedirs(os.path.dirname(uuid_path), exist_ok=True) with open(uuid_path, "w") as f: f.write(new_uuid) return new_uuid def load_last_group_id(path): try: with open(path, 'r') as f: return f.read().strip() except Exception: return None def save_last_group_id(path, group_id): try: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, 'w') as f: f.write(str(group_id)) except Exception as e: logging.error(f"Error saving group_id: {e}") def download_presentation_file(url): """Downloads the presentation file from the given URL.""" try: # Resolve URL to correct API host (same IP as MQTT broker by default) resolved_url = resolve_file_url(url) # Create the presentation directory if it doesn't exist presentation_dir = os.path.join(os.path.dirname(__file__), "presentation") os.makedirs(presentation_dir, exist_ok=True) # Extract the filename from the (possibly encoded) URL filename = unquote(urlsplit(resolved_url).path.split("/")[-1]) or "downloaded_file" file_path = os.path.join(presentation_dir, filename) # Check if the file already exists if os.path.exists(file_path): logging.info(f"File already exists: {file_path}") return # Download the file logging.info(f"Downloading file from: {resolved_url}") response = requests.get(resolved_url, timeout=20) response.raise_for_status() # Raise an error for bad responses # Save the file with open(file_path, "wb") as f: f.write(response.content) logging.info(f"File downloaded successfully: {file_path}") except Exception as e: logging.error(f"Error downloading file: {e}") def resolve_file_url(original_url: str) -> str: """Resolve/normalize a file URL to point to the configured file server. Rules: - If FILE_SERVER_BASE_URL is set, force scheme/host/port from it. - Else default to FILE_SERVER_HOST (defaults to MQTT_BROKER) and FILE_SERVER_PORT (8000). - Only rewrite host when incoming URL host is missing or equals 'server'. - Preserve path and query. """ try: parts = urlsplit(original_url) # Determine target base target_scheme = FILE_SERVER_SCHEME target_host = FILE_SERVER_HOST target_port = FILE_SERVER_PORT if FILE_SERVER_BASE_URL: base = urlsplit(FILE_SERVER_BASE_URL) # Only assign if present to allow partial base definitions if base.scheme: target_scheme = base.scheme if base.hostname: target_host = base.hostname if base.port: target_port = base.port # Decide whether to rewrite incoming_host = parts.hostname should_rewrite = (incoming_host is None) or (incoming_host.lower() == "server") if should_rewrite: # Build netloc with port (always include port to be explicit) netloc = f"{target_host}:{target_port}" if target_port else target_host new_parts = ( target_scheme, netloc, parts.path or "/", parts.query, parts.fragment, ) return urlunsplit(new_parts) else: # Keep original if it's already a proper absolute URL return original_url except Exception as e: logging.warning(f"Could not resolve URL, using original: {original_url} (error: {e})") return original_url def get_latest_screenshot(): """Get the latest screenshot from the host OS shared folder""" try: screenshot_dir = os.path.join(os.path.dirname(__file__), "screenshots") if not os.path.exists(screenshot_dir): return None # Prefer 'latest.jpg' symlink/copy if present (written by display_manager) preferred_path = os.path.join(screenshot_dir, "latest.jpg") if os.path.exists(preferred_path): try: with open(preferred_path, "rb") as f: screenshot_data = base64.b64encode(f.read()).decode('utf-8') file_stats = os.stat(preferred_path) logging.debug(f"Using preferred latest.jpg for screenshot ({file_stats.st_size} bytes)") return { "filename": os.path.basename(preferred_path), "data": screenshot_data, "timestamp": datetime.fromtimestamp(file_stats.st_mtime).isoformat(), "size": file_stats.st_size } except Exception as e: logging.debug(f"Could not read latest.jpg, falling back to newest file: {e}") # Find the most recent screenshot file screenshot_files = [f for f in os.listdir(screenshot_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))] if not screenshot_files: return None # Get the most recent file latest_file = max(screenshot_files, key=lambda f: os.path.getmtime(os.path.join(screenshot_dir, f))) screenshot_path = os.path.join(screenshot_dir, latest_file) # Read and encode screenshot with open(screenshot_path, "rb") as f: screenshot_data = base64.b64encode(f.read()).decode('utf-8') # Get file info file_stats = os.stat(screenshot_path) info = { "filename": latest_file, "data": screenshot_data, "timestamp": datetime.fromtimestamp(file_stats.st_mtime).isoformat(), "size": file_stats.st_size } logging.debug(f"Selected latest screenshot: {latest_file} ({file_stats.st_size} bytes)") return info except Exception as e: logging.error(f"Error reading screenshot: {e}") return None def send_screenshot_heartbeat(client, client_id): """Send heartbeat with screenshot to server for dashboard monitoring""" try: screenshot_info = get_latest_screenshot() heartbeat_data = { "timestamp": datetime.now().isoformat(), "client_id": client_id, "status": "alive", "screenshot": screenshot_info, "system_info": { "hostname": socket.gethostname(), "ip": get_ip(), "uptime": time.time() # Could be replaced with actual uptime } } # Send to dashboard monitoring topic dashboard_topic = f"infoscreen/{client_id}/dashboard" payload = json.dumps(heartbeat_data) res = client.publish(dashboard_topic, payload, qos=0) if res.rc == mqtt.MQTT_ERR_SUCCESS: if screenshot_info: logging.info(f"Dashboard heartbeat sent with screenshot: {screenshot_info['filename']} ({screenshot_info['size']} bytes)") else: logging.info("Dashboard heartbeat sent (no screenshot available)") elif res.rc == mqtt.MQTT_ERR_NO_CONN: logging.warning("Dashboard heartbeat publish returned NO_CONN; will retry on next interval") else: logging.warning(f"Dashboard heartbeat publish failed with code: {res.rc}") except Exception as e: logging.error(f"Error sending screenshot heartbeat: {e}") def screenshot_service_thread(client, client_id): """Background thread for screenshot monitoring and transmission""" logging.info(f"Screenshot service started with {SCREENSHOT_INTERVAL}s interval") while True: try: send_screenshot_heartbeat(client, client_id) time.sleep(SCREENSHOT_INTERVAL) except Exception as e: logging.error(f"Screenshot service error: {e}") time.sleep(60) # Wait a minute before retrying def main(): global discovered print(f"[{datetime.now().isoformat()}] simclient.py: program started") logging.info("Client starting - deleting old event file if present") delete_event_file() client_id = get_persistent_uuid() hardware_token = get_hardware_token() ip_addr = get_ip() # Persistenz für group_id (needed in on_connect) group_id_path = os.path.join(os.path.dirname(__file__), "config", "last_group_id.txt") current_group_id = load_last_group_id(group_id_path) event_topic = None # paho-mqtt v2: opt into latest callback API to avoid deprecation warnings. client_kwargs = {"protocol": mqtt.MQTTv311} try: # Use enum when available (paho-mqtt >= 2.0) if hasattr(mqtt, "CallbackAPIVersion"): client_kwargs["callback_api_version"] = mqtt.CallbackAPIVersion.VERSION2 except Exception: pass client = mqtt.Client(**client_kwargs) client.on_message = on_message # Enable automatic reconnection client.reconnect_delay_set(min_delay=1, max_delay=120) # Connection state tracking connection_state = {"connected": False, "last_disconnect": None} # Optional: Enable MQTT debug logging in DEBUG_MODE if DEBUG_MODE: def on_log(client, userdata, level, buf): logging.debug(f"MQTT: {buf}") client.on_log = on_log # Define subscribe_event_topic BEFORE on_connect so it can be called from the callback def subscribe_event_topic(new_group_id): nonlocal event_topic, current_group_id # Check if group actually changed to handle cleanup group_changed = new_group_id != current_group_id if group_changed: if current_group_id is not None: logging.info(f"Group change from {current_group_id} to {new_group_id} - deleting old event file") delete_event_file() if event_topic: client.unsubscribe(event_topic) logging.info(f"Unsubscribed from event topic: {event_topic}") # Always ensure the event topic is subscribed new_event_topic = f"infoscreen/events/{new_group_id}" # Only subscribe if we don't already have this topic subscribed if event_topic != new_event_topic: if event_topic: client.unsubscribe(event_topic) logging.info(f"Unsubscribed from event topic: {event_topic}") event_topic = new_event_topic client.subscribe(event_topic) logging.info(f"Subscribing to event topic: {event_topic} for group_id: {new_group_id}") else: logging.info(f"Event topic already subscribed: {event_topic}") # Update current group_id and save it if group_changed: current_group_id = new_group_id save_last_group_id(group_id_path, new_group_id) # on_connect callback: Subscribe to all topics after connection is established def on_connect(client, userdata, flags, rc, properties=None): if rc == 0: connection_state["connected"] = True connection_state["last_disconnect"] = None # Check if this is a reconnection # paho-mqtt v2 provides ConnectFlags with attribute 'session_present' # Older versions may provide dict-like flags; default to False. is_reconnect = False try: if hasattr(flags, "session_present"): is_reconnect = bool(getattr(flags, "session_present")) elif isinstance(flags, dict): is_reconnect = bool(flags.get("session present", False)) except Exception: is_reconnect = False if is_reconnect: logging.info("MQTT reconnected successfully - resubscribing to all topics...") else: logging.info("MQTT connected successfully - subscribing to topics...") # Discovery-ACK-Topic abonnieren ack_topic = f"infoscreen/{client_id}/discovery_ack" client.subscribe(ack_topic) logging.info(f"Subscribed to: {ack_topic}") # Config topic client.subscribe(f"infoscreen/{client_id}/config") logging.info(f"Subscribed to: infoscreen/{client_id}/config") # group_id Topic abonnieren (retained) group_id_topic = f"infoscreen/{client_id}/group_id" client.subscribe(group_id_topic) logging.info(f"Subscribed to: {group_id_topic}") # Wenn beim Start eine group_id vorhanden ist, sofort Event-Topic abonnieren if current_group_id: logging.info(f"Subscribing to event topic for saved group_id: {current_group_id}") subscribe_event_topic(current_group_id) # Send discovery message after reconnection to re-register with server if is_reconnect: logging.info("Sending discovery after reconnection to re-register with server") send_discovery(client, client_id, hardware_token, ip_addr) else: connection_state["connected"] = False logging.error(f"MQTT connection failed with code: {rc}") # on_disconnect callback (Paho v2 signature) def on_disconnect(client, userdata, disconnect_flags, rc, properties=None): connection_state["connected"] = False connection_state["last_disconnect"] = time.time() if rc == 0: logging.info("MQTT disconnected cleanly") else: logging.warning(f"MQTT disconnected unexpectedly with code: {rc}") logging.info("Automatic reconnection will be attempted...") client.on_connect = on_connect client.on_disconnect = on_disconnect # Robust MQTT connect with fallbacks and retries broker_candidates = [MQTT_BROKER] # Add environment-provided fallbacks broker_candidates.extend([b for b in MQTT_BROKER_FALLBACKS if b not in broker_candidates]) # Add common local fallbacks for alt in ("127.0.0.1", "localhost", "mqtt"): if alt not in broker_candidates: broker_candidates.append(alt) connect_ok = False last_error = None for attempt in range(1, 6): # up to 5 attempts for host in broker_candidates: try: logging.info(f"Connecting to MQTT broker {host}:{MQTT_PORT} (attempt {attempt}/5)...") client.connect(host, MQTT_PORT) connect_ok = True MQTT_HOST_USED = host # noqa: N816 local doc variable break except Exception as e: last_error = e logging.warning(f"MQTT connection to {host}:{MQTT_PORT} failed: {e}") if connect_ok: break backoff = min(5 * attempt, 20) logging.info(f"Retrying connection in {backoff}s...") time.sleep(backoff) if not connect_ok: logging.error(f"MQTT connection failed after multiple attempts: {last_error}") raise last_error # Start the network loop early to begin connection process client.loop_start() logging.info("MQTT network loop started - establishing connection...") # Wait for connection to complete and on_connect callback to fire logging.info("Waiting for initial connection and subscription setup...") connection_timeout = 30 # seconds start_wait = time.time() while not connection_state["connected"] and (time.time() - start_wait) < connection_timeout: time.sleep(0.5) if not connection_state["connected"]: logging.error(f"Failed to establish initial MQTT connection within {connection_timeout}s") raise Exception("MQTT connection timeout") logging.info("Initial connection established, subscription setup complete") # group_id message callback group_id_topic = f"infoscreen/{client_id}/group_id" def on_group_id_message(client, userdata, msg, properties=None): payload = msg.payload.decode().strip() new_group_id = None # Versuche, group_id aus JSON zu extrahieren, sonst als String verwenden try: data = json.loads(payload) if isinstance(data, dict) and "group_id" in data: new_group_id = str(data["group_id"]) else: new_group_id = str(data) except Exception: new_group_id = payload new_group_id = new_group_id.strip() if new_group_id: if new_group_id != current_group_id: logging.info(f"New group_id received: {new_group_id}") else: logging.info(f"group_id unchanged: {new_group_id}, ensuring event topic is subscribed") # Always call subscribe_event_topic to ensure subscription subscribe_event_topic(new_group_id) else: logging.warning("Empty group_id received!") client.message_callback_add(group_id_topic, on_group_id_message) logging.info(f"Current group_id at start: {current_group_id if current_group_id else 'none'}") # Discovery-Phase: Sende Discovery bis ACK empfangen # The loop is already started, just wait and send discovery messages discovery_attempts = 0 max_discovery_attempts = 20 while not discovered and discovery_attempts < max_discovery_attempts: if connection_state["connected"]: send_discovery(client, client_id, hardware_token, ip_addr) discovery_attempts += 1 # Wait for ACK, checking every second for _ in range(int(HEARTBEAT_INTERVAL)): if discovered: break time.sleep(1) else: logging.info("Waiting for MQTT connection before sending discovery...") time.sleep(2) if discovered: break if not discovered: logging.warning(f"Discovery ACK not received after {max_discovery_attempts} attempts - continuing anyway") # Start screenshot service in background thread screenshot_thread = threading.Thread( target=screenshot_service_thread, args=(client, client_id), daemon=True ) screenshot_thread.start() logging.info("Screenshot service thread started") # Heartbeat-Loop with connection state monitoring last_heartbeat = 0 logging.info("Entering heartbeat loop (network loop already running in background thread)") while True: try: current_time = time.time() # Check connection state and log warnings if disconnected if not connection_state["connected"]: if connection_state["last_disconnect"]: disconnect_duration = current_time - connection_state["last_disconnect"] logging.warning(f"MQTT disconnected for {disconnect_duration:.1f}s - waiting for reconnection...") else: logging.warning("MQTT not connected - waiting for connection...") # Send heartbeat only when connected if current_time - last_heartbeat >= HEARTBEAT_INTERVAL: if client.is_connected(): result = client.publish(f"infoscreen/{client_id}/heartbeat", "alive", qos=0) if result.rc == mqtt.MQTT_ERR_SUCCESS: logging.info("Heartbeat sent.") elif result.rc == mqtt.MQTT_ERR_NO_CONN: logging.debug("Heartbeat publish returned NO_CONN; retrying in 2s...") time.sleep(2) if client.is_connected(): retry = client.publish(f"infoscreen/{client_id}/heartbeat", "alive", qos=0) if retry.rc == mqtt.MQTT_ERR_SUCCESS: logging.info("Heartbeat sent after retry.") else: logging.warning(f"Heartbeat publish failed after retry with code: {retry.rc}") else: logging.warning("Skipping heartbeat retry - MQTT still not connected") else: logging.warning(f"Heartbeat publish failed with code: {result.rc}") else: logging.debug("Skipping heartbeat - MQTT not connected (is_connected=False)") last_heartbeat = current_time time.sleep(5) except KeyboardInterrupt: logging.info("Shutting down gracefully...") client.loop_stop() client.disconnect() break except Exception as e: logging.error(f"Error in main loop: {e}") time.sleep(5) if __name__ == "__main__": main()