Update check_tvheadend.py
WIP
This commit is contained in:
@ -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()
|
||||
args = parse_args()
|
||||
status, message, perfdata = check_tuner_status(args)
|
||||
|
||||
if perfdata:
|
||||
message += " | " + " ".join(perfdata)
|
||||
|
||||
print(message)
|
||||
sys.exit(status)
|
||||
Reference in New Issue
Block a user