MythUtil-Channel-HDHR-channelCheck

#!/usr/bin/python3

import argparse
import sys
import json
import re
import natsort
import requests
import requests.auth

class MythTVServices():
    def __init__(self, host=None, port=None, username=None, password=None):
        if host is None:
            host = 'localhost'
        self.host = host
        if port is None:
            port = 6544
        self.port = port
        self.session = requests.Session()
        if username and password:
            self.session.auth = requests.auth.HTTPDigestAuth(username, password)
            self.request(service='Myth', api='version')

    def request(self, service=None, api=None, data={}, method=None, stream=False):
        version = '0.28'
        headers = {'User-Agent':'{} Python Services API Client'.format(version),
                   'Accept':'application/json',
                   'Accept-Encoding':'gzip,deflate'}
        if api is None:
            raise ValueError('api must be specified')
        url = 'http://{}:{}/{}/{}'.format(self.host, self.port, service, api)
        if method is None:
            if bool(data):
                method = 'post'
            else:
                method = 'get'
        if method == 'get':
            response = self.session.get(url, headers=headers, params=data, stream=stream)
        elif method == 'post':
            response = self.session.post(url, headers=headers, data=data, stream=stream)
        else:
            raise ValueError('method is not post or get: {}'.format(method))
        response.raise_for_status()
        if stream:
            response.raw.decode_content = True
            return response.raw
        else:
            return response.json()

    def Capture(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Capture', api=api, data=data, method=method, stream=stream)
    def Channel(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Channel', api=api, data=data, method=method, stream=stream)
    def Content(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Content', api=api, data=data, method=method, stream=stream)
    def Dvr(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Dvr', api=api, data=data, method=method, stream=stream)
    def Frontend(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Frontend', api=api, data=data, method=method, stream=stream)
    def Guide(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Guide', api=api, data=data, method=method, stream=stream)
    def Myth(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Myth', api=api, data=data, method=method, stream=stream)
    def Video(self, api=None, data={}, method=None, stream=False):
        return self.request(service='Video', api=api, data=data, method=method, stream=stream)

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 '{}'.format(int(m0.group(1)))
    elif m1:
        return '{}.{}'.format(int(m1.group(1)), int(m1.group(2)))
    elif m2:
        return '{}.{}'.format(int(m2.group(1)), int(m2.group(2)))
    elif m3:
        return '{}.{}'.format(int(m3.group(1)), int(m3.group(2)))
    raise TypeError('Invalid channel: {}'.format(channel))

def channelCheck(channel):
    try:
        return channelNormalize(channel)
    except Exception:
        raise argparse.ArgumentTypeError('{} is not a valid channel'.format(channel))

def unsignedInt(v):
    try:
        v = int(v)
    except ValueError:
        raise argparse.ArgumentTypeError("{} is not a valid positive integer".format(v))
    if v < 0:
        raise argparse.ArgumentTypeError("{} is not a valid positive integer".format(v))
    return v

def versionTuple(v):
    return tuple(map(int, (v.split("."))))

def transformAPIElementNames(APIname, existingDict):
    transforms = \
        {
            'VideoSource':
                {
                    'Id':                     'SourceId',
                },
            'Channel':
                {
                    'SourceId':               'SourceID',
                    'MplexId':                'MplexID',
                    'ChanId':                 'ChannelID',
                    'ChanNum':                'ChannelNumber',
                    'ServiceId':              'ServiceID',
                    'ATSCMajorChan':          'ATSCMajorChannel',
                    'ATSCMinorChan':          'ATSCMinorChannel',
                    'Visible':                'visible',
                    'FrequencyId':            'FrequencyID',
                    'DefaultAuth':            'DefaultAuthority'
                }
        }
    transform = transforms.get(APIname, {})
    newDict = {}
    for key, value in existingDict.items():
        newKey = transform.get(key, key)
        if newKey is not None:
            newDict[newKey] = value
    return newDict

if __name__ == '__main__':

    parser = argparse.ArgumentParser()
    parser.add_argument('--hdhr', action='store', type=str, required=True,
                        help='the HDHomeRun from which to retrieve the lineup')
    parser.add_argument('--backend', '--host', action='store', type=str, default='localhost',
                        help='the host (backend) to access.  The default is localhost.')
    parser.add_argument('--port', action='store', type=int, default=6544,
                        help='the port to connect to on on the host.  The default is 6544')
    parser.add_argument('--username', action='store', type=str, default=None,
                        help='the username to use for host authentication')
    parser.add_argument('--password', action='store', type=str, default=None,
                        help='the password to use for host authentication')
    sourcegroup = parser.add_mutually_exclusive_group(required=True)
    sourcegroup.add_argument('--videosource-name', action='store', type=str, dest='sourceName',
                             help='the video source name')
    sourcegroup.add_argument('--videosource-id', action='store', type=unsignedInt, dest='sourceId',
                             help='the video source id')
    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('--include-demo-channels', action='store_true', default=False, dest='includeDemo',
                        help='include demo channels on HDHR')
    args = parser.parse_args()

    s = MythTVServices(args.backend, args.port, args.username, args.password)

    try:
        hostname = s.Myth('GetHostName')['String']
    except Exception:
        print('Unable to obtain hostname from host {}:{}'.format(args.backend, args.port))
        sys.exit(1)

    ChannelServiceVersion = versionTuple(s.Channel('version')['String'])

    # Validate sourceid/name
    mythsl = s.Channel('GetVideoSourceList')['VideoSourceList']['VideoSources']
    sourceId = None
    sourceName = None
    for source in mythsl:
        if int(source['Id']) == args.sourceId or source['SourceName'] == args.sourceName:
            sourceId = int(source['Id'])
            sourceName = source['SourceName']
            break
    if sourceId is None:
        print('Video source not found')
        sys.exit(1)

    # Get channel list from hdhr
    try:
        hdhrLineup = requests.get('http://{}/lineup.json?show=all&tuning'.format(args.hdhr)).json()
    except Exception:
        print('Unable to obtain lineup from {}'.format(args.hdhr))
        sys.exit(1)
    hdhrChannelList = {}
    for hdhrChannel in hdhrLineup:
        if 'GuideNumber' not in hdhrChannel:
            continue
        if 'Demo' in hdhrChannel:
            if bool(hdhrChannel['Demo']) and not args.includeDemo:
                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

    # Get channel list for source
    mythChannelInfo = s.Channel('GetChannelInfoList', {'SourceID': sourceId, 'Details': True})['ChannelInfoList']['ChannelInfos']
    mythChannelList = []
    mythChannelFreqList = []
    for mythChannel in mythChannelInfo:
        if 'ChanNum' not in mythChannel:
            continue
        c = channelNormalize(mythChannel['ChanNum'])
        if args.channelInclude is not None:
            if c not in args.channelInclude:
                continue
        if args.channelExclude is not None:
            if c in args.channelExclude:
                continue
        mythChannelList.append(mythChannel)
        f = channelNormalize(mythChannel['FrequencyId'])
        mythChannelFreqList.append(f)

    # Report on channels that (likely) should be added to MythTV
    for guidenumber in natsort.natsorted(hdhrChannelList.keys()):
        if guidenumber not in mythChannelFreqList:
            hdhrChannel = hdhrChannelList[guidenumber]
            guidename = ''
            if 'GuideName' in hdhrChannel:
                guidename = hdhrChannel['GuideName']
            print('Tuner channel {} ({}) is not in the MythTV channel database'.format(guidenumber, guidename))

    # Report on channels that (likely) should be deleted from MythTV
    mythChannelList = natsort.natsorted(mythChannelList, key=lambda c: c['ChanNum'])
    for mythChannel in mythChannelList:
        if mythChannel['FrequencyId']  not in hdhrChannelList:
            print('MythTV channel {} ({}) specifies a tuner channel {} which is not available'.format(mythChannel['ChanNum'], mythChannel['CallSign'], mythChannel['FrequencyId']))