hdhomerun_emulator.py

import socket
import threading
import time
import logging
import select
import struct
import os
import hashlib

logger = logging.getLogger(__name__)

class HDHomeRunEmulator:
    def __init__(self, http_port=5005, config_items=None):
        self.http_port = http_port
        self.device_id = self._generate_device_id(config_items)
        # Model number configurable via environment variable
        self.model = os.getenv("HDHR_MODEL", "HDHR3-US")
        # Friendly name configurable via environment variable
        self.friendly_name = os.getenv("HDHR_FRIENDLY_NAME", "IPTV HDHomeRun")
        # Tuner count configurable via environment variable (default 2)
        self.tuner_count = int(os.getenv("HDHR_TUNER_COUNT", "2"))
        self.running = False
        self.thread = None
        self._stop_event = threading.Event()
        # Check environment variable for default state, but allow runtime override
        self._env_disabled = os.getenv("HDHR_DISABLE_SSDP", "0") == "1"
        self.ssdp_disabled = self._env_disabled
    
    def _generate_device_id(self, ip_port_tuple=None):
        """Generate a stable 8-digit hex device ID based on IP and port."""
        # Use provided tuple or detect
        if ip_port_tuple:
            ip, port = ip_port_tuple
        else:
            ip = self.get_host_ip()
            port = self.http_port
        id_source = f"{ip}:{port}"
        hash_obj = hashlib.md5(id_source.encode())
        device_id = hash_obj.hexdigest()[:8].upper()
        logger.info(f"Generated device ID from {id_source}: {device_id}")
        return device_id
    
    def update_device_id(self, ip_port_tuple=None):
        """Update device ID based on current IP/port."""
        new_id = self._generate_device_id(ip_port_tuple)
        if new_id != self.device_id:
            logger.info(f"Device ID changed from {self.device_id} to {new_id}")
            self.device_id = new_id
        
    def is_env_disabled(self):
        """Check if SSDP is disabled via environment variable"""
        return self._env_disabled
    
    def get_host_ip(self):
        """Simple IP detection"""
        try:
            # Use a short timeout so this call can't block for long on networks
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.settimeout(0.5)
            try:
                # The connect here doesn't perform network I/O for UDP, but it
                # can still block on some platforms; setting a timeout keeps us safe.
                s.connect(("8.8.8.8", 80))
                ip = s.getsockname()[0]
            except Exception:
                ip = "127.0.0.1"
            finally:
                try:
                    s.close()
                except Exception:
                    pass
            return ip
        except Exception as e:
            logger.debug(f"get_host_ip failed: {e}")
            return "127.0.0.1"
    
    def create_ssdp_response(self):
        host_ip = self.get_host_ip()
        base_url = f"http://{host_ip}:{self.http_port}"
        
        response = f"""HTTP/1.1 200 OK
CACHE-CONTROL: max-age=1800
EXT:
LOCATION: {base_url}/discover.json
SERVER: HDHomeRun/1.0 UPnP/1.0
ST: upnp:rootdevice
USN: uuid:{self.device_id}::upnp:rootdevice
BOOTID.UPNP.ORG: 1
CONFIGID.UPNP.ORG: 1
DEVICEID.UPNP.ORG: {self.device_id}
HDHomerun-Device: {self.device_id}
HDHomerun-Device-Auth: iptv_emulator
HDHomerun-Features: base

"""
        return response
    
    def handle_ssdp_discovery(self, data, addr, sock):
        if "M-SEARCH" in data:
            if any(st in data for st in ["upnp:rootdevice", "ssdp:all", "urn:schemas-upnp-org:device:MediaRenderer:1"]):
                response = self.create_ssdp_response()
                try:
                    sock.sendto(response.encode('utf-8'), addr)
                except Exception as e:
                    pass
    
    def run_ssdp_server(self):
        self.running = True
        sock = None
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
            
            # Set timeout on the socket BEFORE bind to prevent hanging on macOS
            sock.settimeout(5.0)
            
            try:
                t0 = time.time()
                logger.info("Attempting to bind to SSDP port 1900 (may take a few seconds on macOS)...")
                sock.bind(('0.0.0.0', 1900))
                t1 = time.time()
                logger.info(f"SSDP socket.bind completed in {t1 - t0:.3f}s")
            except socket.timeout:
                logger.error(f"SSDP bind timed out after 5 seconds - HDHomeRun discovery will not work")
                logger.error("This is a known issue on macOS Docker. The app will continue without SSDP.")
                # Don't raise - just log and return
                self.running = False
                return
            except Exception as e:
                logger.error(f"SSDP bind failed: {e}")
                self.running = False
                raise

            mreq = struct.pack("4sl", socket.inet_aton("239.255.255.250"), socket.INADDR_ANY)
            try:
                t0 = time.time()
                sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
                t1 = time.time()
                logger.info(f"SSDP setsockopt ADD_MEMBERSHIP completed in {t1 - t0:.3f}s")
            except Exception as e:
                logger.error(f"SSDP setsockopt ADD_MEMBERSHIP failed: {e}")
                # Not fatal; continue but log

            sock.setblocking(0)

            while self.running:
                try:
                    ready = select.select([sock], [], [], 1.0)
                    if ready[0]:
                        data, addr = sock.recvfrom(1024)
                        self.handle_ssdp_discovery(data.decode('utf-8'), addr, sock)
                except Exception:
                    logger.debug("SSDP listen error")
                    time.sleep(1)
        except Exception as e:
            logger.error(f"SSDP server failed: {e}")
        finally:
            if sock is not None:
                try:
                    sock.close()
                except Exception:
                    pass
    
    def start(self, force=False):
        """Start SSDP server. 
        
        Args:
            force: If True, override environment variable setting and try to start anyway
        """
        if self.ssdp_disabled and not force:
            logger.warning("SSDP discovery is disabled (HDHR_DISABLE_SSDP=1). HDHomeRun features available via HTTP only.")
            logger.info("To enable: Use the 'Enable Discovery' button or set HDHR_DISABLE_SSDP=0 and restart")
            return False
        
        if self._env_disabled and force:
            logger.warning("Attempting to enable SSDP despite HDHR_DISABLE_SSDP=1 environment variable")
            logger.warning("Note: Port 1900/udp must be exposed in docker-compose.yml for this to work")
        
        if self.thread is None or not self.thread.is_alive():
            logger.info("Starting SSDP thread for HDHomeRun emulator")
            self.ssdp_disabled = False
            self.thread = threading.Thread(target=self.run_ssdp_server, daemon=True)
            self.thread.start()
            return True
        
        logger.info("SSDP thread already running")
        return True

    def stop(self, timeout: float = 2.0):
        """Stop the SSDP thread gracefully."""
        try:
            logger.info("Stopping SSDP thread for HDHomeRun emulator")
            self.running = False
            self.ssdp_disabled = True
            if self.thread is not None and self.thread.is_alive():
                self.thread.join(timeout)
            return True
        except Exception as e:
            logger.error(f"Error stopping SSDP thread: {e}")
            return False

    def is_running(self) -> bool:
        """Check if the emulator is running by verifying both the thread state and running flag."""
        return bool(self.running and self.thread and self.thread.is_alive())

hdhomerun_emulator = HDHomeRunEmulator()