198 lines
6.8 KiB
Python
198 lines
6.8 KiB
Python
#!/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) |