HDHRUtil-Tuner-channelSettingSave

#!/usr/bin/env python3

"""

HDHRUtil-Tuner-channelSettingSave

HDHRUtil-Tuner-channelSettingSave is a utility for saving
the channel selection on SiliconDust HDHR devices.

Copyright (c) 2022 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.



===== NOTE ===== NOTE ===== NOTE ===== NOTE ===== NOTE =====

This utility currently uses a reverse engineered
interface to the HDHR, which, as it is not documented
as a stable API, could change at any time.

===== NOTE ===== NOTE ===== NOTE ===== NOTE ===== NOTE =====


"""


import sys
import json
import re
import argparse
import socket
import natsort
import requests


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 (Legacy) or (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 = {}

        Legacy = bool(discoverResponse.get('Legacy', False))
        DeviceID = discoverResponse.get('DeviceID', None)
        DiscoverURL = discoverResponse.get('DiscoverURL', None)
        LineupURL = discoverResponse.get('LineupURL', None)
        LocalIP = discoverResponse.get('LocalIP', None)

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

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

    return eligibleTuners

def channelNormalize(channel):
    m0 = re.match(r'^(\d+)$', channel)
    m1 = re.match(r'^(\d+)\.(\d+)$', channel)
    m2 = re.match(r'^(\d+)_(\d+)$', channel)
    m3 = re.match(r'^(\d+)-(\d+)$', channel)
    if m0:
        return '{0}'.format(int(m0.group(1)))
    elif m1:
        return '{0}.{1}'.format(int(m1.group(1)), int(m1.group(2)))
    elif m2:
        return '{0}.{1}'.format(int(m2.group(1)), int(m2.group(2)))
    elif m3:
        return '{0}.{1}'.format(int(m3.group(1)), int(m3.group(2)))
    raise TypeError('Invalid channel: {0}'.format(channel))


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 manage')
    parser.add_argument('--outfile', '--output-file', action='store', type=str, dest='outfile', required=False,
                        help='the file in which to store the HDHR settings')
    parser.add_argument('--use-cloud-discovery', action='store_true', default=False, dest='usecloud',
                        help='use the SiliconDust Cloud API services to discover local tuners')
    args = parser.parse_args()

    # Discover HDHRs
    discoveredHDHRs = HDHRdiscover()

    # Try to match the selected HDHR

    HDHRip = None

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

    if HDHRip is None:
        print("The source HDHR device not found")
        sys.exit(1)

    # Get channel list from source hdhr
    hdhrLineup = {}
    try:
        hdhrLineup = requests.get('http://{}/lineup.json?show=all'.format(HDHRip)).json()
    except (requests.exceptions.RequestException, json.decoder.JSONDecodeError):
        print('Unable to obtain lineup from {}'.format(HDHRip))
        sys.exit(1)

    # Create minimal list for save/restore
    lineup = []
    for hdhrChannel in hdhrLineup:
        if 'GuideNumber' not in hdhrChannel:
            continue
        guidenumber = channelNormalize(hdhrChannel['GuideNumber'])
        subscribed = True
        if 'Subscribed' in hdhrChannel:
            subscribed = bool(hdhrChannel['Subscribed'])
        if not subscribed:
            continue
        enabled = True
        if 'Enabled' in hdhrChannel:
            enabled = bool(hdhrChannel['Enabled'])
        favorite = False
        if 'Favorite' in hdhrChannel:
            favorite = bool(hdhrChannel['Favorite'])
        lineup.append({'GuideNumber': guidenumber, 'Enabled': enabled, 'Favorite': favorite})

    # Output file minimal status
    if args.outfile and args.outfile != '-':
        try:
            fh = open(args.outfile, 'w+')
        except OSError as e:
            print('unable to open ' + args.outfile + ' for writing: ' + str(e))
            sys.exit(1)
    else:
        fh = sys.stdout
    print(json.dumps(lineup, indent=2), file=fh)
    if fh is not sys.stdout:
        fh.close

    sys.exit(1)




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