#!/usr/bin/python3 ################################################################################ # Start a MythTV recording. ################################################################################ import argparse import datetime import lxml import MythTV import os import sys from dateutil.relativedelta import relativedelta from enum import IntEnum from pprint import pprint import MythTV.services_api.send as api import MythTV.services_api.utilities as util ############################################################################## # Configuration # # These settings are intended to be changed by users of this program to # customise it to their needs. ############################################################################## # 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 # 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' # Default chanid of the channel to be recorded. CHANID: int = 1001 # Name of the template rule to be used as the basis for creating the new # recording rule. TEMPLATE: str = 'Default' # Default title for the recording. TITLE: str = 'Dive Cam' # Default duration of the recording (Units: minutes) DURATION: int = 360 ############################################################################## # Version: 0.1 2021-12-28 ############################################################################## VERSION = '0.1' ############################################################################## # Startup ############################################################################## program_name = os.path.basename(sys.argv[0]) ############################################################################## # Program exit codes. ############################################################################## class ExitCode(IntEnum): OK = 0 MessageAndExit = 1 MissingPackageOrProgram = 2 DryRun = 3 Bad = 4 # Failed to create recording rule. ############################################################################## # Debug output. ############################################################################## if DEBUG_OUTPUT: global sr_debug sr_debug = None global previous_end previous_end = '\n' def dprint_init() -> None: global sr_debug if sr_debug == None: sr_debug = sys.stdout if DEBUG_FILE: for debug_dir in DEBUG_DIRS: try: sr_debug = 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 sr_debug global previous_end if sr_debug != None: if previous_end == '\n': print(datetime.datetime.now().strftime('%H:%M:%S.%f')[:-3] + ' ', *args, **kwargs, file=sr_debug) else: print(*args, end=end, file=sr_debug, **kwargs) previous_end = end def dpprint(*args, **kwargs) -> None: pprint(*args, **kwargs, stream=sr_debug) else: def dprint_init() -> None: pass def dprint(*args, end='\n', **kwargs) -> None: pass def dpprint(*args, **kwargs) -> None: pass dprint_init() ############################################################################## # Check for an IPv6 address. ############################################################################## def is_ipv6_address(addr) -> bool: return ':' in addr ############################################################################## # 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') ############################################################################## # 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: dprint('config_xml="' + config_xml + '"') self.config_xml = config_xml if self.config_xml != '': self.remote_config = ConfigXml() 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 if is_ipv6_address(self.host): self.host = '[' + self.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 create_new_recording(self): try: chaninfo = self.backend.send(endpoint='Channel/GetChannelInfo?ChanID=' + str(args.chanid)) except: dprint(self.host + ' create_new_recording(): Exception from Channel/GetChannelInfo') if DEBUG_OUTPUT: raise return None try: rec_rule = self.backend.send(endpoint='Dvr/GetRecordSchedule?Template=' + TEMPLATE) except: dprint(self.host + ' create_new_recording(): Exception from Dvr/GetRecordSchedule') if DEBUG_OUTPUT: raise return None dpprint(rec_rule) postdata=rec_rule['RecRule'] postdata['ChanId'] = args.chanid postdata['Title'] = args.title + ' (Manual Record)' now = datetime.datetime.utcnow() postdata['StartTime'] = to_mythtv_datetime(now) postdata['EndTime'] = to_mythtv_datetime(now + relativedelta(minutes=360)) postdata['StartOffset'] = 0 postdata['EndOffset'] = 0 postdata['SubTitle'] = args.subtitle postdata['Type'] = 'Single' postdata['SearchType'] = 'Manual Search' postdata['Search'] = 'Manual' postdata['Station'] = chaninfo['ChannelInfo']['CallSign'] postdata['FindTime'] = util.create_find_time(postdata['StartTime']) 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_new_recording(): Exception from Dvr/AddRecordSchedule') if DEBUG_OUTPUT: raise return None dprint(f'{r=}') return r ############################################################################## # Main ############################################################################## parser = argparse.ArgumentParser( description='Start recording (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('-C', '--chanid', type=int, action='store', default=CHANID, help='Chanid of the channel to be recorded (default: ' + str(CHANID) + ')') parser.add_argument('-d', '--duration', type=int, action='store', default=DURATION, help='Duration of the recording in minutes (default: ' + str(DURATION) + ')') parser.add_argument('-s', '--subtitle', type=str, action='store', default='', help='Subtitle for the recording (default: empty string)') parser.add_argument('-t', '--title', type=str, action='store', default=TITLE, help='Title for the recording (default: ' + TITLE + ')') parser.add_argument('-n', '--dryrun', action='store_true', default=False, help='Dry run only - do not start the recording, but do everything else up to that point.') args = parser.parse_args() if args.version: print('Version ' + VERSION) exit(ExitCode.MessageAndExit) be = Backend(args.config) result = be.create_new_recording() if result == None: print('Failed to start recording') exit(ExitCode.Bad) else: if not 'uint' in result: print(f'Error: recording may not have started, bad result: {result}') exit(ExitCode.Bad) print(f"Started recording, recordid={result['uint']}") exit(ExitCode.OK)