HDHRUtil-Tuner-channelSetting

#!/usr/bin/env python3

"""

HDHRUtil-Tuner-channelSetting

HDHRUtil-Tuner-channelSetting is a utility for configuring the
channel selection on SiliconDust HDHR devices.

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

def channelCheck(channel):
    try:
        return channelNormalize(channel)
    except TypeError:
        raise argparse.ArgumentTypeError('{} is not a valid channel'.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('--channel', '--channels', '--include-channel', '--include-channels',
                        nargs='+', type=channelCheck, dest='channelInclude',
                        help='list of channels to consider.  The default is all')
    parser.add_argument('--exclude-channel', '--exclude-channels', '--no-channel', '--no-channels',
                        nargs='+', type=channelCheck, dest='channelExclude',
                        help='list of channels to exclude.  The default is none')
    parser.add_argument('--set', action='store', type=str, dest='setValue',
                        choices=['enabled', 'disabled', 'favorite'],
                        help='channel setting (enabled, disabled, favorite).  The default is just to report the status.')
    parser.add_argument('--dry-run', action='store_true', dest='dryrun', default=False,
                        help='dry run only (do not update)')
    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 select 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("HDHR device not found")
        sys.exit(1)

    # Get channel list from hdhr
    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)
    hdhrChannelList = {}
    for hdhrChannel in hdhrLineup:
        subscribed = True
        if 'Subscribed' in hdhrChannel:
            subscribed = bool(hdhrChannel['Subscribed'])
        if not subscribed:
            continue
        if 'GuideNumber' not in hdhrChannel:
            continue
        guidenumber = channelNormalize(hdhrChannel['GuideNumber'])
        if args.channelInclude is not None:
            if guidenumber not in args.channelInclude:
                continue
        if args.channelExclude is not None:
            if guidenumber in args.channelExclude:
                continue
        hdhrChannelList[guidenumber] = hdhrChannel

    for channel, hdhrChannel in natsort.natsorted(hdhrChannelList.items()):
        enabled = True
        favorite = False
        if 'Enabled' in hdhrChannel:
            enabled = bool(hdhrChannel['Enabled'])
        if 'Favorite' in hdhrChannel:
            favorite = bool(hdhrChannel['Favorite'])
        if 'GuideName' in hdhrChannel:
            name = hdhrChannel['GuideName']
        else:
            name = 'Unknown'
        currentValue = 'disabled'
        if favorite:
            currentValue = 'favorite'
        elif enabled:
            currentValue = 'enabled'
        print('Channel ' + channel + ' (' + name + ') is currently set as ' + currentValue)
        if args.setValue is not None:
            newValue = None
            if args.setValue == 'favorite':
                if not favorite:
                    newValue = '+'
                    print('  Channel ' + channel + ' to be marked as a favorite')
            elif args.setValue == 'disabled':
                if enabled:
                    newValue = 'x'
                    print('  Channel ' + channel + ' to be disabled')
            else:   # args.setValue == 'enabled'
                if (not enabled) or favorite:
                    newValue = '-'
                    print('  Channel ' + channel + ' to be enabled')
            if (newValue is not None) and (args.dryrun is not True):
                try:
                    r = requests.post('http://' + HDHRip + '/lineup.post?favorite=' + newValue + channel)
                    r.raise_for_status()
                    print('    Channel ' + channel + ' updated')
                except requests.exceptions.RequestException:
                    print('    Channel ' + channel + ' update failed')




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