HDHRUtil-Tuner-waitOnline

#!/usr/bin/env python3

"""

HDHRUtil-Tuner-waitOnline

HDHRUtil-Tuner-waitOnline is a utility to wait for
a HDHR tuner to come online.  While typically it
is expected that its usage will be in a systemd
unit (either via an ExecPre= in a server.service
which needs the tuner(s) or a oneshot with an After=
in the server.service file.  There may be other
uses.

Copyright (c) 2024 by Gary Buhrmaster <gary.buhrmaster@gmail.com>

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

"""


import sys
import json
import re
import argparse
import socket
import requests
import time


def HDHRdiscover():

    #
    # Try a few different ways to identify eligible HDHRs
    #
    #
    # First, if --use-cloud-discovery is specified, try to obtain
    # the list from api.hdhomerun.com (SD provided service)
    #
    # Second, try to get a list of IP addresses that appear to
    # be tuners (via a hand constructed packet (we just collect
    # the IP addresses, and then perform a discovery)
    #

    discoveredTuners = {}

    if args.usecloud:
        SDdiscover = []
        try:
            r = requests.get('https://api.hdhomerun.com/discover', timeout=(4, 1))
            r.raise_for_status()
            SDdiscover = r.json()
            if not isinstance(SDdiscover, list):
                SDdiscover = []
        except (requests.exceptions.RequestException, json.decoder.JSONDecodeError):
            SDdiscover = []

        for device in SDdiscover:
            if not isinstance(device, dict):
                continue
            Legacy = bool(device.get('Legacy', False))
            DeviceID = device.get('DeviceID', None)
            DiscoverURL = device.get('DiscoverURL', None)
            LocalIP = device.get('LocalIP', None)

            if (LocalIP is None) or (DeviceID is None) or (DiscoverURL is None):
                continue

            discoveredTuners[LocalIP] = DiscoverURL

    discovery_udp_port = 65001
    # Hand constructed discovery message (device type = tuner, device id = wildcard)
    discovery_udp_msg = bytearray.fromhex('00 02 00 0c 01 04 00 00 00 01 02 04 ff ff ff ff 4e 50 7f 35')
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    sock.settimeout(.2)
    for _ in range(2):
        sock.sendto(discovery_udp_msg, ('<broadcast>', discovery_udp_port))
        while True:
            try:
                (buf, addr) = sock.recvfrom(2048)
            except socket.timeout:
                break
            if addr is None:
                continue
            if buf is None:
                continue

            DiscoverURL = 'http://' + addr[0] + ':80/discover.json'
            discoveredTuners[addr[0]] = DiscoverURL

    eligibleTuners = []

    for device in discoveredTuners:
        discoverResponse = {}
        try:
            r = requests.get(discoveredTuners[device], timeout=(4, 1))
            r.raise_for_status()
            discoverResponse = r.json()
            if not isinstance(discoverResponse, dict):
                discoverResponse = {}
        except (requests.exceptions.RequestException, json.decoder.JSONDecodeError):
            discoverResponse = {}

        discoverResponse['LocalIP'] = device
        eligibleTuners.append(discoverResponse)

    return eligibleTuners


if __name__ == '__main__':

    # Parse our args
    parser = argparse.ArgumentParser()
    parser.add_argument('--hdhr', action='store', type=str, dest='hdhr', required=True,
                        help='the HDHomeRun to wait for')
    parser.add_argument('--use-cloud-discovery', action='store_true', default=False, dest='usecloud',
                        help='use the SiliconDust Cloud API services to discover local tuners')
    parser.add_argument('--timeout', action='store', type=int, default=30, dest='timeout',
                        help='timeout for waiting')
    args = parser.parse_args()

    # Since the network, or DNS services, may not
    # be available, we have to keep asking for
    # the entire set of HDHRs

    startTime = time.time()

    hdhrFound = False

    while (not hdhrFound):

        discoveredHDHRs = HDHRdiscover()

        if re.match(r'^[0-9A-Z]{8}$', args.hdhr.upper()):   # deviceid?
            for d in discoveredHDHRs:
                if d['DeviceID'] == args.hdhr.upper():
                    # Found a match
                    hdhrFound = True
                    break
        else:                                               # possible IP or dns
            try:
                ip = socket.getaddrinfo(args.hdhr, None)[0][4][0]
                for d in discoveredHDHRs:
                    if d['LocalIP'] == ip:
                        # Found a match
                        hdhrFound = True
                        break
            except socket.error:
                pass

        if (time.time() - startTime) > args.timeout:
            print('timeout while waiting for tuner')
            sys.exit(1)
        time.sleep(0.1)

    sys.exit(0)




# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4