#!/usr/bin/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 import time 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 1.0 2024-02-18 # First working version # Version 1.1 2024-02-25 # Fix finding matching ANY rules. # Version 1.2 2024-11-20 # Fix for MythTV v34 - integers now returned as type int instead of type # str. ############################################################################## VERSION:str = '1.2' ############################################################################## # Configuration ############################################################################## # Set this to True to enable debug output. DEBUG_OUTPUT: bool = True #DEBUG_OUTPUT: bool = False # Set this to True to for debug output to be on by default. Set to False # for debug output to be off by default. Toggle the state of debug_output # using the command line -T option. #debug_output = DEBUG_OUTPUT #debug_output = True debug_output = False # Set this to send debug output to a separate debug log file. #DEBUG_FILE: bool = True DEBUG_FILE: bool = False # List of locations to try to write the debug log file to. These locations # are tried in order, and if they all fail, no debug log file is written - the # debug output will be sent to sysout. DEBUG_DIRS = ['/var/log/mythtv', '/var/log', '/tmp', '.'] # 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: 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: if debug_output: 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 DryRun = 3 ############################################################################## # Convert a datetime to a MythTV format ISO UTC time string such as this: # 2021-12-27T08:22:16Z ############################################################################## def to_mythtv_datetime(dt: datetime) -> str: return dt.strftime('%Y-%m-%dT%H:%M:%SZ') ############################################################################## # Convert a MythTV format ISO UTC time string such as this: # 2021-12-27T08:22:16Z # to a datetime. ############################################################################## def to_datetime(mythtime: str) -> datetime: if mythtime[19] == 'Z': mythtime = mythtime[:19] return datetime.datetime.fromisoformat(mythtime) ############################################################################## # 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(): ############################################################################## # Init ############################################################################## 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() ############################################################################## # Refresh the stored recording rules. ############################################################################## def refresh_recording_rules(self): 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)) exit(ExitCode.CanNotTalkToBackend) else: #dpprint(str(resp_dict)) self.all_rules = resp_dict['RecRuleList'] #rules_count = all_rules['Count'] ############################################################################## # Get all scheduled recording rules for a channel. ############################################################################## def get_recording_rules(self, chanid: int) -> List[str]: if self.all_rules == []: self.refresh_recording_rules() rules = [] for rule in self.all_rules['RecRules']: if int(rule['ChanId']) == chanid: rules.append(rule) #if (rule['Title'] == 'Ghosts' or rule['Title'] == 'Tracker' or rule['Title'] == 'The Price Is Right') and rule['Type'] != 'Override Recording': # print("\n".join(f"{k}\t{v}" for k, v in rule.items())) return rules ############################################################################## # Get the upcoming recordings list for a channel. ############################################################################## def get_upcoming_list(self, chanid: int): try: resp_dict = self.backend.send(endpoint='Dvr/GetUpcomingList') except: dprint(self.host + ' get_upcoming_list(): Exception') dprint(f'{resp_dict=}') return None if list(resp_dict.keys())[0] in ['Abort', 'Warning']: dprint(self.host + ' get_upcoming_list(): Abort or Warning') dprint(str(resp_dict)) exit(ExitCode.CanNotTalkToBackend) else: upcoming = [] for program in resp_dict['ProgramList']['Programs']: if int(program['Channel']['ChanId']) == chanid: upcoming.append(program) return upcoming ############################################################################## # Get a channel's callsign. ############################################################################## 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'] ############################################################################## # Create an override rule from an existing rule. ############################################################################## def create_override(self, chanid: int, callsign: str, id: int, startts: str, endts: str) -> None: try: override_rule = self.backend.send(endpoint='Dvr/GetRecordSchedule?RecordId=' + str(id) + '&MakeOverride=True') except: dprint(self.host + ' create_override(): Exception from Dvr/GetRecordSchedule') if DEBUG_OUTPUT: raise return None # dprint(f'{override_rule=}') # dprint(f"diff={set(rec_rule) ^ set(override_rule['RecRule'])}") dprint(f'{override_rule=}') postdata=override_rule['RecRule'] postdata['Type'] = 'Override Recording' postdata['ChanId'] = str(chanid) postdata['ParentId'] = postdata['Id'] postdata['StartTime'] = to_mythtv_datetime(to_datetime(startts) + datetime.timedelta(minutes=int(override_rule['RecRule']['StartOffset']))) postdata['EndTime'] = to_mythtv_datetime(to_datetime(endts) - datetime.timedelta(minutes=int(override_rule['RecRule']['EndOffset']))) #print(f"{postdata['NextRecording']=}") #next_recording = to_datetime(override_rule['RecRule']['NextRecording']) + \ # datetime.timedelta(minutes=int(override_rule['RecRule']['StartOffset'])) #postdata['EndTime'] = to_mythtv_datetime( # next_recording + # (to_datetime(override_rule['RecRule']['EndTime']) - to_datetime(override_rule['RecRule']['StartTime'])) # ) #postdata['StartTime'] = next_recording postdata['Station'] = callsign params_not_sent = ( 'Description', 'SubTitle', 'Season', 'Episode', 'CallSign', 'AverageDelay', 'Id', 'LastDeleted', 'LastRecorded', ) for param in params_not_sent: del postdata[param] try: r = self.backend.send(endpoint='Dvr/AddRecordSchedule', postdata=postdata, opts={'debug': True, 'wrmi': not args.dry_run}) except RuntimeWarning as w: dpprint(f'{w=}') dpprint(f'{w.args=}') if w.args[0] == 'wrmi=False': print('Dry run completed OK, override rule not created') else: dprint(self.host + ' create_override(): Exception from Dvr/AddRecordSchedule') if DEBUG_OUTPUT: raise else: print(f"Override: {chanid=}, parentid={postdata['ParentId']}, starttime={postdata['StartTime']}, title={postdata['Title']}") return ############################################################################## # Delete all override rules for a channel. ############################################################################## def delete_override_rules(self, rules) -> int: deleted_count = 0 for rule in rules: if rule['Type'] == 'Override Recording' and rule['Title'][-11:] != ' (Template)': dprint(f'{rule=}') postdata = {'RecordId': rule['Id']} opts = {'noetag': False, 'nogzip': False, 'usexml': False, 'timeout': 2, 'wsdl': False, 'wrmi': not args.dry_run} dprint(f'{opts=}') rule_display = f"ChanId: {rule['ChanId']}, ParentId: {rule['ParentId']}, RecordId: {rule['Id']}, Title: {rule['Title']}, Type: {rule['Type']}" try: self.backend.send(endpoint='Dvr/RemoveRecordSchedule', postdata=postdata, opts=opts) except RuntimeWarning: print(f'Dry run, would have deleted: ' + rule_display) else: print(f'Deleted: ' + rule_display) deleted_count += 1 return deleted_count ############################################################################## # 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) ############################################################################## # List all the override rules for a chanid. ############################################################################## def list_override_rules(rules): for rule in rules: if rule['Type'] == 'Override Recording': dprint(rule) print(f"Override rule: ChanId: {rule['ChanId']}, ParentId: {rule['ParentId']}, RecordId: {rule['Id']}, StartTime: {rule['StartTime']}, Title: {rule['Title']}") ############################################################################## # 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('-d', '--delete', dest='delete', default=False, action='store_true', help='Delete all override rules from the primary and secondary channels.') parser.add_argument('-T', '--toggle-debug', dest='toggle_debug', default=False, action='store_true', help='Toggle the debug setting - if it defaults to on, turn debug off. If it defaults to off, turn debug on. Ignored if debug is disabled.') parser.add_argument('-n', '--dry_run', dest='dry_run', default=False, action='store_true', help='Do not actually make any changes, just try things out.') parser.add_argument('-l', '--list-overrides', dest='list_overrides', default=False, action='store_true', help='Debug option - list all override recording rules on the primary and secondary channels.') 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.toggle_debug: debug_output = not debug_output 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=}') print('Primary channel:') print(f'chanid={args.primary} callsign={callsign[args.primary]}') print() print('Secondary channel(s):') for chanid in args.secondary: print(f'{chanid=} callsign={callsign[chanid]}') print() if args.delete: be.delete_override_rules(primary_rules) for chanid in args.secondary: #be.refresh_recording_rules() be.delete_override_rules(secondary_rules[chanid]) elif args.list_overrides: print(f'Primary override rules: chanid={args.primary}') list_override_rules(primary_rules) for chanid in args.secondary: print(f'Secondary override rules: {chanid=}') list_override_rules(secondary_rules[chanid]) else: # Where an override does not already exist, create override rules in the primary and secondary channels for all primary rules. upcoming_list = be.get_upcoming_list(args.primary) for upcoming in upcoming_list: id = upcoming['Recording']['RecordId'] if int(upcoming['Channel']['ChanId']) == args.primary: startts = upcoming['Recording']['StartTs'] endts = upcoming['Recording']['EndTs'] be.create_override(args.primary, callsign[args.primary], id, startts, endts) for chanid in args.secondary: be.create_override(chanid, callsign[chanid], id, startts, endts) exit(ExitCode.OK)