#!/usr/bin/env python3 # -*- coding: utf-8 -*- ############################################################################## # Record Simulcasts # # When programmes are available from more than one station, but both stations # are unreliable and can make bad recordings, it is useful to force recordings # of the simulcasts from multiple stations at once. This program will read # the recording rules from one station where the MythTV user creates recording # rules and will automatically create override rules to force recording the # same programme from all the simulcasts. # # Author: J S Worthington # Created: 2024-02-06 ############################################################################## import argparse import datetime import lxml import MythTV import os import sys from enum import IntEnum from pprint import pprint from typing import Dict, List import MythTV.services_api.send as api import MythTV.services_api.utilities as util program_name = os.path.basename(sys.argv[0]) ############################################################################## # # Version: 0.1 2024-02-06 # Initial version ############################################################################## VERSION:str = '0.01' ############################################################################## # Configuration ############################################################################## # Set this to True get debug output. DEBUG_OUTPUT: bool = True #DEBUG_OUTPUT: bool = False # Set this to send debug output to a separate debug log file. #DEBUG_FILE: bool = True DEBUG_FILE: bool = False # Location of default config.xml file. CONFIG_XML: str = '/etc/mythtv/config.xml' # The chanid of the primary station where the MythTV user creates recording rules. PRIMARY:int = 1010 # FreeviewHD DVB-T Sky Open # List of chanid values for the secondary stations where simulcasts occur. SECONDARY:[int] = [10004] # [SkyTV DVB-S2 Sky Open HD] # The IP address of the mythbackend to talk to. Default: 127.0.0.1 HOST: str = '127.0.0.1' ############################################################################## # Debug output. ############################################################################## if DEBUG_OUTPUT: global debug_output_file debug_output_file = None global previous_end previous_end = '\n' def dprint_init() -> None: global debug_output_file if debug_output_file == None: debug_output_file = sys.stdout if DEBUG_FILE: for debug_dir in DEBUG_DIRS: try: debug_output_file = open(debug_dir + '/' + program_name + '-debug.log', 'w', 1) except: continue break dprint(program_name + ' debug output started') def dprint(*args, end='\n', **kwargs) -> None: global debug_output_file global previous_end if debug_output_file != None: if previous_end == '\n': print(datetime.datetime.now().strftime('%H:%M:%S.%f')[:-3] + ' ', *args, **kwargs, file=debug_output_file) else: print(*args, end=end, file=debug_output_file, **kwargs) previous_end = end def dpprint(*args, **kwargs) -> None: pprint(*args, **kwargs, stream=debug_output_file) else: def dprint_init() -> None: pass def dprint(*args, end='\n', **kwargs) -> None: pass def dpprint(*args, **kwargs) -> None: pass dprint_init() ############################################################################## # Program exit codes. ############################################################################## class ExitCode(IntEnum): OK = 0 MessageAndExit = 1 CanNotTalkToBackend = 2 ############################################################################## # Read config.xml file to get database settings only (other settings ignored). ############################################################################## class ConfigXml: DBHostName = 'localhost' DBUsername = 'mythtv' DBPassword = 'mythtv' DBName = 'mythconverg' DBPort = 3306 _conf_trans = { 'Host':'DBHostName', 'UserName':'DBUserName', 'Password':'DBPassword', 'DatabaseName':'DBName', 'Port':'DBPort' } def readXML(self, filename): if not os.access(filename, os.R_OK): dprint('File ' + filename + ' not accessible!') return False try: config = lxml.etree.parse(filename) for child in config.xpath('/Configuration/Database')[0].getchildren(): if child.tag in self._conf_trans: #dprint('child.tag=' + str(child.tag)) setattr(self, self._conf_trans[child.tag], child.text) except Exception as e: dprint(str(e)) dprint('lxml.etree.parse failed') return False return True ############################################################################## # Backend class. Handles communication with a backend and database selected # using a config.xml file. If no config.xml file is provided, the local # database and backend are used. ############################################################################## class Backend(): def __init__(self, config_xml='') -> None: self.all_rules = [] dprint('config_xml="' + config_xml + '"') self.config_xml = config_xml if self.config_xml != '': self.remote_config = ConfigXml() dprint(f'{self.remote_config=}') if not self.remote_config.readXML(self.config_xml): dprint('Failed to read or correctly parse ' + self.config_xml + ' file, using ' + HOST) self.config_xml = '' self.host = HOST # dprint(str(self.remote_config.DBHostName)) # dprint(str(self.remote_config.DBUsername)) # dprint(str(self.remote_config.DBPassword)) # dprint(str(self.remote_config.DBName)) # dprint(str(self.remote_config.DBPort)) self.host = self.remote_config.DBHostName dprint('self.host = ' + self.host) else: self.host = HOST self.backend = api.Send(self.host) # Initialise the UTCOffset so that the conversion to local time works. api.Send(self.host) util.get_utc_offset(self.backend) # Connect to the database. if self.config_xml != '': try: self.db = MythTV.MythDB( DBHostName = self.remote_config.DBHostName, DBUsername = self.remote_config.DBUsername, DBPassword = self.remote_config.DBPassword, DBName = self.remote_config.DBName, DBPort = self.remote_config.DBPort ) except MythTV.exceptions.MythDBError as e: if e.ecode == MythTV.MythError.DB_SCHEMAMISMATCH: dprint('Schema mismatch connecting to ' + self.host + ' database, trying again with schema ' + str(e.remote)) MythTV.MythDB._schema_local = e.remote self.db = MythTV.MythDB( DBHostName = self.remote_config.DBHostName, DBUsername = self.remote_config.DBUsername, DBPassword = self.remote_config.DBPassword, DBName = self.remote_config.DBName, DBPort = self.remote_config.DBPort ) dprint('After retry') else: raise MythTV.MythSchema._schema_local = MythTV.SCHEMA_VERSION else: self.db = MythTV.MythDB() self.dbc = self.db.cursor() def get_recording_rules(self, chanid: int) -> List[str]: if self.all_rules == []: try: resp_dict = self.backend.send(endpoint='Dvr/GetRecordScheduleList') except: dprint(self.host + ' get_recording_rules(): Exception') dprint(f'{resp_dict=}') return [] if list(resp_dict.keys())[0] in ['Abort', 'Warning']: dprint(self.host + ' get_recording_rules(): Abort or Warning') dprint(str(resp_dict)) sys.exit(ExitCode.CanNotTalkToBackend) else: #dpprint(str(resp_dict)) self.all_rules = resp_dict['RecRuleList'] #rules_count = all_rules['Count'] rules = [] for rule in self.all_rules['RecRules']: if int(rule['ChanId']) == chanid: rules.append(rule) return rules def get_callsign(self, chanid: int): try: chaninfo = self.backend.send(endpoint='Channel/GetChannelInfo?ChanID=' + str(chanid)) except: dprint(self.host + ' get_callsign(): Exception from Channel/GetChannelInfo') if DEBUG_OUTPUT: raise return None return chaninfo['ChannelInfo']['CallSign'] def create_override(self, chanid: int, callsign: str, rec_rule: dict): try: rec_rule = self.backend.send(endpoint='Dvr/GetRecordSchedule?Template=' + TEMPLATE) except: dprint(self.host + ' create_override(): Exception from Dvr/GetRecordSchedule') if DEBUG_OUTPUT: raise return None dpprint(rec_rule) postdata=rec_rule['RecRule'] postdata['StartOffset'] = 0 postdata['EndOffset'] = 0 postdata['Station'] = chaninfo['ChannelInfo']['CallSign'] params_not_sent = ( 'AverageDelay', 'CallSign', 'Id', 'LastDeleted', 'LastRecorded', 'NextRecording', 'ParentId' ) for param in params_not_sent: del postdata[param] dpprint(postdata) try: r = self.backend.send(endpoint='Dvr/AddRecordSchedule', postdata=postdata, opts={'debug': True, 'wrmi': not args.dryrun}) except RuntimeWarning as w: dpprint(f'{w=}') dpprint(f'{w.args=}') if w.args[0] == 'wrmi=False': print('Dry run completed OK, recording not started') exit(ExitCode.DryRun) else: dprint(self.host + ' create_override(): Exception from Dvr/AddRecordSchedule') if DEBUG_OUTPUT: raise return None dprint(f'{r=}') return r ############################################################################## # Display all the rule types found for a chanid. ############################################################################## def display_rule_types(rules): rule_types = {} for rule in rules: rule_type = rule['Type'] if rule_type in rule_types: rule_types[rule_type] += 1 else: rule_types[rule_type] = 1 dprint(rule_types) ############################################################################## # Main ############################################################################## parser = argparse.ArgumentParser( description='Record Simulcasts (Version: ' + VERSION + ')', formatter_class=lambda prog: argparse.HelpFormatter(prog, max_help_position=15) ) parser.add_argument('-V', '--version', dest='version', default=False, action='store_true', help='Display the version number and exit') parser.add_argument('-c', '--config', type=str, action='store', default=CONFIG_XML, help='Location of MythTV config.xml file (default: ' + CONFIG_XML + ')') parser.add_argument('primary', nargs='?', type=int, default=PRIMARY, help=f'The chanid of the primary channel, default is {PRIMARY}. The primary channel is the channel where the recording rules are created by the user.') parser.add_argument('secondary', nargs='*', type=int, default=SECONDARY, help=f'The chanids of the secondary channel(s), default is {SECONDARY}. A secondary channel is a channel where the recording rules should only be override rules automatically created by {program_name}.') args = parser.parse_args() if args.version: print('Version '+VERSION) exit(ExitCode.MessageAndExit) be = Backend(args.config) dprint(f'{args.secondary=}') primary_rules = be.get_recording_rules(args.primary) #dpprint(primary_rules) dprint(f'Primary rule types (chanid={args.primary})') display_rule_types(primary_rules) callsign = {} callsign[args.primary] = be.get_callsign(args.primary) secondary_rules = {} for chanid in args.secondary: callsign[chanid] = be.get_callsign(chanid) secondary_rules[chanid] = be.get_recording_rules(chanid) dprint(f'Secondary rule types ({chanid=})') display_rule_types(secondary_rules[chanid]) dprint(f'{callsign=}') exit(ExitCode.OK)