MythUtil-Channel-visibility

#!/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

def extendedVisibility(arg):
    ua = str(arg).lower()
    if ua in ['true'[:len(ua)], 'yes'[:len(ua)], '1', 'visible'[:len(ua)]]:
        return 'Visible'
    elif ua in ['false'[:len(ua)], 'no'[:len(ua)], '0', 'invisible'[:len(ua)], 'notvisible'[:len(ua)], 'not visible'[:len(ua)]]:
        return 'Not Visible'
    elif ua in ['always'[:len(ua)], '2', 'always visible'[:len(ua)]]:
        return 'Always Visible'
    elif ua in ['never'[:len(ua)], '-1', 'never visible'[:len(ua)]]:
        return 'Never Visible'
    raise argparse.ArgumentTypeError('{} is not a valid visible/true, not visible/false, always visible, never visible value'.format(arg))

if __name__ == '__main__':

    parser = argparse.ArgumentParser()
    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')
    visiblegroup = parser.add_mutually_exclusive_group(required=False)
    visiblegroup.add_argument('--force-visible', action='store_true', default=False, dest='forcevisible',
                              help='force the channel(s) to be set to visible')
    visiblegroup.add_argument('--force-invisible', action='store_true', default=False, dest='forceinvisible',
                              help='force the channel(s) to be set to invisible')
    visiblegroup.add_argument('--visibility', action='store', type=extendedVisibility, dest='extendedvisible',
                              help='specify the channel(s) visibility')
    parser.add_argument('--update', action='store_true', default=False,
                        help='update the channels to correct any detected problem.  The default is just report')
    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'])

    if (ChannelServiceVersion < versionTuple("1.9")) and (args.extendedvisible in ['Never Visible', 'Always Visible']):
        print('Warning: backend does not supported extended visibility settings')

    # 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 for source
    mythChannelInfo = s.Channel('GetChannelInfoList', {'SourceID': sourceId, 'Details': True})['ChannelInfoList']['ChannelInfos']
    mythChannelList = []
    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)

    # Report visibility changes
    mythChannelList = natsort.natsorted(mythChannelList, key=lambda c: c['ChanNum'])
    for mythChannel in mythChannelList:
        chanNum = mythChannel['ChanNum']
        if ChannelServiceVersion >= versionTuple("1.9"):
            mvisible = mythChannel['ExtendedVisible']
            desiredvisible = mvisible
            if args.extendedvisible:
                desiredvisible = args.extendedvisible
            elif (args.forcevisible) and (mvisible in ['Visible', 'Not Visible']):
                desiredvisible = 'Visible'
            elif (args.forceinvisible) and (mvisible in ['Visible', 'Not Visible']):
                desiredvisible = 'Not Visible'
        else:
            mvisible = bool(mythChannel['Visible'] == 'true')
            desiredvisible = mvisible
            if args.extendedvisible:
                desiredvisible = bool(args.extendedvisible in ['Visible', 'Always Visible'])
            elif args.forcevisible:
                desiredvisible = True
            elif args.forceinvisible:
                desiredvisible = False
        if mvisible != desiredvisible:
            print('Channel {} visibility should be: {}'.format(chanNum, desiredvisible))
            if ChannelServiceVersion >= versionTuple("1.9"):
                data = {}
                data['ChanId'] = mythChannel['ChanId']
                data['ExtendedVisible'] = desiredvisible
            else:
                data = mythChannel.copy()
                data['Visible'] = desiredvisible
            if args.update:
                if bool(s.Channel('UpdateDBChannel', transformAPIElementNames('Channel', data))['bool']):
                    print('  Channel {} updated'.format(chanNum))
                else:
                    print('  Channel {} update failed'.format(chanNum))
            else:
                print('   Channel {} update not requested'.format(chanNum))
        else:
            print('Channel {} visibility currently set to {}'.format(chanNum, mvisible))

    if args.update:
        try:
            s.Dvr('RescheduleRecordings')
        except Exception:
            pass