diff --git a/check_tvheadend.py b/check_tvheadend.py index 5d4773b..b821bd3 100644 --- a/check_tvheadend.py +++ b/check_tvheadend.py @@ -15,54 +15,184 @@ def parse_args(): 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("--timeout", type=int, default=10, help="Request timeout") - parser.add_argument("--debug", action="store_true", help="Enable debug mode") + 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 check_api_status(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: - # First check basic API accessibility - status_url = f"{base_url}/api/status" if args.debug: - print(f"DEBUG: Checking API status at {status_url}") - - response = requests.get(status_url, auth=auth, timeout=args.timeout) - if response.status_code == 401: - raise Exception("Authentication failed - check credentials") - response.raise_for_status() - - if args.debug: - print(f"DEBUG: API status response: {response.json()}") - - return True - except Exception as e: - if args.debug: - print(f"DEBUG: API check failed: {str(e)}") - raise Exception(f"Cannot access TVHeadend API: {str(e)}") + print("DEBUG: Starting TVHeadend check...") + print(f"DEBUG: Base URL: {base_url}") -def main(): - args = parse_args() - - try: - # Verify API is accessible - check_api_status(args) + # Discover available endpoints + endpoints = discover_api_endpoints(base_url, auth, args.timeout, args.debug) - # If we got here, API is accessible but we couldn't find specific endpoints - msg = "TVHeadend API is accessible but required endpoints not found" + # Get adapter status + adapters = tvheadend_api_call( + base_url + endpoints['adapters'], + auth, + args.timeout, + args.debug + ) + if args.debug: - print(f"DEBUG: {msg}") - print("DEBUG: Try checking available endpoints with:") - print(f"DEBUG: curl -u '{args.user}:{args.password}' http://{args.host}:{args.port}/api/help") - - print(msg) - sys.exit(UNKNOWN) - + 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: - print(str(e)) - sys.exit(CRITICAL) + if args.debug: + print(f"DEBUG: Error occurred: {str(e)}") + return CRITICAL, str(e), [] if __name__ == "__main__": - main() \ No newline at end of file + args = parse_args() + status, message, perfdata = check_tuner_status(args) + + if perfdata: + message += " | " + " ".join(perfdata) + + print(message) + sys.exit(status) \ No newline at end of file