#!/usr/bin/env python3 import requests from requests.auth import HTTPDigestAuth import argparse import sys OK = 0 WARNING = 1 CRITICAL = 2 UNKNOWN = 3 def parse_args(): parser = argparse.ArgumentParser(description="TVHeadend DVB Tuner Check") parser.add_argument("--host", default="localhost", help="TVHeadend host") parser.add_argument("--port", default="9981", help="TVHeadend port") parser.add_argument("--user", required=True, help="API username") parser.add_argument("--password", required=True, help="API password") parser.add_argument("--min_snr", type=int, default=20, help="Minimum SNR threshold (dB)") parser.add_argument("--min_signal", type=int, default=-60, help="Minimum signal strength (dBm)") parser.add_argument("--timeout", type=int, default=10, help="Request timeout in seconds") parser.add_argument("--debug", action="store_true", help="Enable debug output") return parser.parse_args() def tvheadend_api_call(url, auth, timeout, debug=False): headers = {"Accept": "application/json"} try: if debug: print(f"DEBUG: Calling API endpoint: {url}") response = requests.get(url, auth=auth, headers=headers, timeout=timeout) if debug: print(f"DEBUG: Response status: {response.status_code}") if response.status_code == 404: return None response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: if debug: print(f"DEBUG: Request failed: {str(e)}") raise Exception(f"API request failed: {str(e)}") def discover_api_endpoints(base_url, auth, timeout, debug): endpoints = { 'adapters': None, 'muxes': None } # Common possible endpoints for adapters possible_adapter_endpoints = [ '/api/dvb/adapters', # Older versions '/api/hardware/tree', # Newer versions '/api/adapter' # Some variants ] # Common possible endpoints for muxes possible_mux_endpoints = [ '/api/mux/status', '/api/dvb/mux/status', '/api/mux' ] # Test adapter endpoints for endpoint in possible_adapter_endpoints: if debug: print(f"DEBUG: Trying adapter endpoint: {endpoint}") result = tvheadend_api_call(base_url + endpoint, auth, timeout, debug) if result is not None: endpoints['adapters'] = endpoint break # Test mux endpoints for endpoint in possible_mux_endpoints: if debug: print(f"DEBUG: Trying mux endpoint: {endpoint}") result = tvheadend_api_call(base_url + endpoint, auth, timeout, debug) if result is not None: endpoints['muxes'] = endpoint break if debug: print(f"DEBUG: Discovered endpoints: {endpoints}") if not endpoints['adapters'] or not endpoints['muxes']: raise Exception("Could not discover required API endpoints. Check TVHeadend version.") return endpoints def check_tuner_status(args): base_url = f"http://{args.host}:{args.port}" auth = HTTPDigestAuth(args.user, args.password) try: if args.debug: print("DEBUG: Starting TVHeadend check...") print(f"DEBUG: Base URL: {base_url}") # Discover available endpoints endpoints = discover_api_endpoints(base_url, auth, args.timeout, args.debug) # Get adapter status adapters = tvheadend_api_call( base_url + endpoints['adapters'], auth, args.timeout, args.debug ) if args.debug: print(f"DEBUG: Adapters response: {adapters}") # Find DVB-T/T2 adapters dvbt_adapters = [] for adapter in adapters.get("entries", []): if args.debug: print(f"DEBUG: Checking adapter: {adapter}") # Different versions use different field names adapter_type = adapter.get("type") or adapter.get("adapter_type") or "" if "DVB-T" in adapter_type.upper(): dvbt_adapters.append(adapter) if not dvbt_adapters: return CRITICAL, "No active DVB-T/T2 adapters found", [] if args.debug: print(f"DEBUG: Found {len(dvbt_adapters)} DVB-T/T2 adapters") # Get muxes (signal information) muxes = tvheadend_api_call( base_url + endpoints['muxes'], auth, args.timeout, args.debug ) if args.debug: print(f"DEBUG: Muxes response: {muxes}") status = OK messages = [] perfdata = [] for mux in muxes.get("entries", []): # Different versions use different status indicators enabled = mux.get("enabled", False) status_field = mux.get("status", "active") or mux.get("state", "active") if enabled and status_field.lower() == "active": signal = mux.get("signal", 0) snr = mux.get("snr", 0) ber = mux.get("ber", 0) mux_id = mux.get("uuid", mux.get("id", "unknown")) perfdata.extend([ f"'{mux_id}_signal'={signal}dBm", f"'{mux_id}_snr'={snr}dB", f"'{mux_id}_ber'={ber}" ]) if args.debug: print(f"DEBUG: MUX {mux_id} - Signal: {signal}dBm, SNR: {snr}dB, BER: {ber}") if snr < args.min_snr or signal < args.min_signal: msg = f"MUX {mux_id}: Low signal (SNR={snr}dB, Signal={signal}dBm)" if snr < (args.min_snr / 2) or signal < (args.min_signal - 10): status = max(status, CRITICAL) messages.append(f"CRITICAL: {msg}") else: status = max(status, WARNING) messages.append(f"WARNING: {msg}") if status == OK: active_muxes = len([m for m in muxes.get("entries", []) if m.get("enabled", False) and (m.get("status", "active") or m.get("state", "active")).lower() == "active"]) message = f"OK: {active_muxes} active MUXes with good signal" if args.debug: print(f"DEBUG: {message}") return OK, message, perfdata else: return status, " ".join(messages), perfdata except Exception as e: if args.debug: print(f"DEBUG: Error occurred: {str(e)}") return CRITICAL, str(e), [] if __name__ == "__main__": args = parse_args() status, message, perfdata = check_tuner_status(args) if perfdata: message += " | " + " ".join(perfdata) print(message) sys.exit(status)