diff --git a/python_apps/media-monitor/MediaMonitor.py b/python_apps/media-monitor/MediaMonitor.py index 172af7f6a..909371f75 100644 --- a/python_apps/media-monitor/MediaMonitor.py +++ b/python_apps/media-monitor/MediaMonitor.py @@ -3,20 +3,22 @@ import time import logging import logging.config import sys +import os import signal -from multiprocessing import Process +from multiprocessing import Process, Queue as mpQueue from airtimefilemonitor.airtimenotifier import AirtimeNotifier from airtimefilemonitor.airtimeprocessevent import AirtimeProcessEvent from airtimefilemonitor.mediaconfig import AirtimeMediaConfig +from airtimefilemonitor.airtimemediamonitorbootstrap import AirtimeMediaMonitorBootstrap def handleSigTERM(signum, frame): logger = logging.getLogger() - logger.info("Main Process Shutdown, TERM signal caught. %d") + logger.info("Main Process Shutdown, TERM signal caught.") for p in processes: - p.terminate() logger.info("Killed process. %d", p.pid) + p.terminate() sys.exit(0) @@ -26,49 +28,52 @@ try: logging.config.fileConfig("logging.cfg") except Exception, e: print 'Error configuring logging: ', e - sys.exit() + sys.exit(1) logger = logging.getLogger() processes = [] try: - config = AirtimeMediaConfig() + config = AirtimeMediaConfig(logger) + + multi_queue = mpQueue() logger.info("Initializing event processor") - pe = AirtimeProcessEvent(airtime_config=config) + pe = AirtimeProcessEvent(queue=multi_queue, airtime_config=config) - notifier = AirtimeNotifier(pe.wm, pe, read_freq=0.1, timeout=0.1, airtime_config=config) + notifier = AirtimeNotifier(pe.wm, pe, read_freq=1, timeout=0, airtime_config=config) notifier.coalesce_events() - - #create 5 worker processes - for i in range(5): - p = Process(target=notifier.process_file_events, args=(pe.multi_queue,)) - processes.append(p) - p.start() - - signal.signal(signal.SIGTERM, handleSigTERM) - + logger.info("Setting up monitor") response = None while response is None: response = notifier.api_client.setup_media_monitor() time.sleep(5) - + storage_directory = response["stor"].encode('utf-8') logger.info("Storage Directory is: %s", storage_directory) config.storage_directory = storage_directory + config.imported_directory = storage_directory + '/imported' + + bootstrap = AirtimeMediaMonitorBootstrap(logger, multi_queue, pe) + bootstrap.scan() + #create 5 worker processes + for i in range(5): + p = Process(target=notifier.process_file_events, args=(multi_queue,)) + processes.append(p) + p.start() + wdd = pe.watch_directory(storage_directory) logger.info("Added watch to %s", storage_directory) logger.info("wdd result %s", wdd[storage_directory]) + #register signal before process forks and exits. + signal.signal(signal.SIGTERM, handleSigTERM) notifier.loop(callback=pe.notifier_loop_callback) - - for p in processes: - p.join() - + except KeyboardInterrupt: notifier.stop() + logger.info("Keyboard Interrupt") except Exception, e: - notifier.stop() + #notifier.stop() logger.error('Exception: %s', e) - diff --git a/python_apps/media-monitor/airtime-media-monitor-init-d b/python_apps/media-monitor/airtime-media-monitor-init-d index 8bf311161..38dc92895 100755 --- a/python_apps/media-monitor/airtime-media-monitor-init-d +++ b/python_apps/media-monitor/airtime-media-monitor-init-d @@ -17,13 +17,13 @@ DAEMON=/usr/lib/airtime/media-monitor/airtime-media-monitor PIDFILE=/var/run/airtime-media-monitor.pid start () { - monit monitor airtime-media-monitor >/dev/null 2>&1 + #monit monitor airtime-media-monitor >/dev/null 2>&1 start-stop-daemon --start --background --quiet --chuid $USERID:$GROUPID --make-pidfile --pidfile $PIDFILE --startas $DAEMON } stop () { # Send TERM after 5 seconds, wait at most 30 seconds. - monit unmonitor airtime-media-monitor >/dev/null 2>&1 + #monit unmonitor airtime-media-monitor >/dev/null 2>&1 start-stop-daemon --stop --oknodo --retry TERM/5/0/30 --quiet --pidfile $PIDFILE rm -f $PIDFILE } diff --git a/python_apps/media-monitor/airtimefilemonitor/airtimemediamonitorbootstrap.py b/python_apps/media-monitor/airtimefilemonitor/airtimemediamonitorbootstrap.py new file mode 100644 index 000000000..d6efe9a10 --- /dev/null +++ b/python_apps/media-monitor/airtimefilemonitor/airtimemediamonitorbootstrap.py @@ -0,0 +1,109 @@ +import os +import time + +from subprocess import Popen, PIPE + + +class AirtimeMediaMonitorBootstrap(): + + def __init__(self, logger, multi_queue, pe): + self.logger = logger + self.multi_queue = multi_queue + self.pe = pe + self.airtime_tmp = '/var/tmp/airtime' + + """ + on bootup we want to scan all directories and look for files that + weren't there or files that changed before media-monitor process + went offline. We can do this by doing a hash of the directory metadata. + """ + def scan(self): + directories = ['/srv/airtime/stor'] + + for dir in directories: + self.check_for_diff(dir) + + def check_for_diff(self, dir): + #set to hold new and/or modified files. We use a set to make it ok if files are added + #twice. This is become some of the tests for new files return result sets that are not + #mutually exclusive from each other. + added_files = set() + removed_files = set() + + if os.path.exists(self.airtime_tmp + '/.airtime_media_index'): + + #find files that have been modified since the last time + #media-monitor process was running. + time_diff_sec = time.time() - os.path.getmtime(self.airtime_tmp + '/.airtime_media_index') + command = "find %s -type f -iname '*.ogg' -o -iname '*.mp3' -readable -mmin -%d" % (dir, time_diff_sec/60+1) + self.logger.debug(command) + stdout = self.execCommandAndReturnStdOut(command) + self.logger.info("Files modified since last checkin: \n%s\n", stdout) + + new_files = stdout.split('\n') + + for file_path in new_files: + added_files.add(file_path) + + + + + #a previous index exists, we can do a diff between this + #file and the current state to see whether anything has + #changed. + self.logger.info("Previous index file found.") + + #find deleted files + command = "find %s -type f -iname '*.ogg' -o -iname '*.mp3' -readable > %s/.airtime_media_index.tmp" % (dir, self.airtime_tmp) + self.execCommand(command) + + command = "diff -u %s/.airtime_media_index %s/.airtime_media_index.tmp" % (self.airtime_tmp, self.airtime_tmp) + stdout = self.execCommandAndReturnStdOut(command) + + #remove first 3 lines from the diff output. + stdoutSplit = (stdout.split('\n'))[3:] + + self.logger.info("Changed files since last checkin:\n%s\n", "\n".join(stdoutSplit)) + + for line in stdoutSplit: + if len(line.strip(' ')) > 1: + if line[0] == '+': + added_files.add(line[1:]) + elif line[0] == '-': + removed_files.add(line[1:]) + + self.pe.write_index_file() + else: + #a previous index does not exist. Most likely means that + #media monitor has never seen this directory before. Let's + #notify airtime server about each of these files + self.logger.info("Previous index file does not exist. Creating a new one") + + #create a new index file. + stdout = self.pe.write_index_file() + + new_files = stdout.split('\n') + + for file_path in new_files: + added_files.add(file_path) + + for file_path in removed_files: + self.pe.handle_removed_file(file_path) + + for file_path in added_files: + if os.path.exists(file_path): + self.pe.handle_created_file(False, os.path.basename(file_path), file_path) + + def execCommand(self, command): + p = Popen(command, shell=True) + sts = os.waitpid(p.pid, 0)[1] + if sts != 0: + self.logger.warn("command \n%s\n return with a non-zero return value", command) + + def execCommandAndReturnStdOut(self, command): + p = Popen(command, shell=True, stdout=PIPE) + stdout = p.communicate()[0] + if p.returncode != 0: + self.logger.warn("command \n%s\n return with a non-zero return value", command) + return stdout + \ No newline at end of file diff --git a/python_apps/media-monitor/airtimefilemonitor/airtimenotifier.py b/python_apps/media-monitor/airtimefilemonitor/airtimenotifier.py index ee41c249e..fd29a5e5a 100644 --- a/python_apps/media-monitor/airtimefilemonitor/airtimenotifier.py +++ b/python_apps/media-monitor/airtimefilemonitor/airtimenotifier.py @@ -3,8 +3,6 @@ import time import os import logging -from multiprocessing import Process, Lock, Queue as mpQueue - # For RabbitMQ from kombu.connection import BrokerConnection from kombu.messaging import Exchange, Queue, Consumer, Producer @@ -26,6 +24,7 @@ class AirtimeNotifier(Notifier): self.md_manager = AirtimeMetadata() self.import_processes = {} self.watched_folders = [] + while not self.init_rabbit_mq(): self.logger.error("Error connecting to RabbitMQ Server. Trying again in few seconds") @@ -104,8 +103,14 @@ class AirtimeNotifier(Notifier): mm.watch_directory(new_storage_directory) + #update airtime with information about files discovered in our + #watched directories. Pass in a dict() object with the following + #attributes: + # -filepath + # -mode + # -data + # -is_recorded_show def update_airtime(self, d): - filepath = d['filepath'] mode = d['mode'] diff --git a/python_apps/media-monitor/airtimefilemonitor/airtimeprocessevent.py b/python_apps/media-monitor/airtimefilemonitor/airtimeprocessevent.py index 7ec3c941d..f22ec7993 100644 --- a/python_apps/media-monitor/airtimefilemonitor/airtimeprocessevent.py +++ b/python_apps/media-monitor/airtimefilemonitor/airtimeprocessevent.py @@ -4,7 +4,7 @@ import grp import pwd import logging -from multiprocessing import Process, Lock, Queue as mpQueue +from subprocess import Popen, PIPE import pyinotify from pyinotify import WatchManager, Notifier, ProcessEvent @@ -18,7 +18,7 @@ from airtimefilemonitor.mediaconfig import AirtimeMediaConfig class AirtimeProcessEvent(ProcessEvent): - def my_init(self, airtime_config=None): + def my_init(self, queue, airtime_config=None): """ Method automatically called from ProcessEvent.__init__(). Additional keyworded arguments passed to ProcessEvent.__init__() are then @@ -34,11 +34,12 @@ class AirtimeProcessEvent(ProcessEvent): self.gui_replaced = {} self.renamed_files = {} self.file_events = [] - self.multi_queue = mpQueue() + self.multi_queue = queue self.mask = pyinotify.ALL_EVENTS self.wm = WatchManager() self.md_manager = AirtimeMetadata() + #define which directories the pyinotify WatchManager should watch. def watch_directory(self, directory): return self.wm.add_watch(directory, self.mask, rec=True, auto_add=True) @@ -139,6 +140,7 @@ class AirtimeProcessEvent(ProcessEvent): return filepath + #create path in /srv/airtime/stor/imported/[song-metadata] def create_file_path(self, imported_filepath, orig_md): storage_directory = self.config.storage_directory @@ -177,7 +179,6 @@ class AirtimeProcessEvent(ProcessEvent): #yyyy-mm-dd-hh-MM-ss y = orig_md['MDATA_KEY_YEAR'].split("-") filepath = '%s/%s/%s/%s/%s-%s-%s%s' % (storage_directory, "recorded".encode('utf-8'), y[0], y[1], orig_md['MDATA_KEY_YEAR'], md['MDATA_KEY_TITLE'], md['MDATA_KEY_BITRATE'], file_ext) - is_recorded_show = True elif(md['MDATA_KEY_TRACKNUMBER'] == u'unknown'.encode('utf-8')): filepath = '%s/%s/%s/%s/%s-%s%s' % (storage_directory, "imported".encode('utf-8'), md['MDATA_KEY_CREATOR'], md['MDATA_KEY_SOURCE'], md['MDATA_KEY_TITLE'], md['MDATA_KEY_BITRATE'], file_ext) else: @@ -191,36 +192,55 @@ class AirtimeProcessEvent(ProcessEvent): except Exception, e: self.logger.error('Exception: %s', e) - return filepath, is_recorded_show + return filepath + #event.dir: True if the event was raised against a directory. + #event.name + #event.pathname: pathname (str): Concatenation of 'path' and 'name'. def process_IN_CREATE(self, event): - #self.logger.info("%s: %s", event.maskname, event.pathname) + self.logger.debug("PROCESS_IN_CREATE") + self.handle_created_file(event.dir, event.name, event.pathname) + + + def handle_created_file(self, dir, name, pathname): + + self.logger.debug("dir: %s, name: %s, pathname: %s ", dir, name, pathname) storage_directory = self.config.storage_directory - - if not event.dir: + if not dir: #file created is a tmp file which will be modified and then moved back to the original filename. - if self.is_temp_file(event.name) : - self.temp_files[event.pathname] = None + if self.is_temp_file(name) : + self.temp_files[pathname] = None #This is a newly imported file. - elif self.is_audio_file(event.pathname): - if self.is_parent_directory(event.pathname, storage_directory): - self.set_needed_file_permissions(event.pathname, event.dir) - file_md = self.md_manager.get_md_from_file(event.pathname) - - if file_md is not None: - filepath, is_recorded_show = self.create_file_path(event.pathname, file_md) - self.move_file(event.pathname, filepath) - self.renamed_files[event.pathname] = filepath - self.file_events.append({'mode': self.config.MODE_CREATE, 'filepath': filepath, 'data': file_md, 'is_recorded_show': is_recorded_show}) + elif self.is_audio_file(pathname): + if self.is_parent_directory(pathname, storage_directory): + self.set_needed_file_permissions(pathname, dir) + + self.process_new_file(pathname) else: - self.file_events.append({'mode': self.config.MODE_CREATE, 'filepath': event.pathname, 'is_recorded_show': False}) + self.file_events.append({'mode': self.config.MODE_CREATE, 'filepath': pathname, 'is_recorded_show': False}) else: - if self.is_parent_directory(event.pathname, storage_directory): - self.set_needed_file_permissions(event.pathname, event.dir) - + if self.is_parent_directory(pathname, storage_directory): + self.set_needed_file_permissions(pathname, dir) + + + def process_new_file(self, pathname): + self.logger.info("Processing new file: %s", pathname) + file_md = self.md_manager.get_md_from_file(pathname) + if file_md is not None: + is_recorded_show = 'MDATA_KEY_CREATOR' in file_md and \ + file_md['MDATA_KEY_CREATOR'] == "AIRTIMERECORDERSOURCEFABRIC".encode('utf-8') + if not self.is_parent_directory(pathname, self.config.imported_directory): + filepath = self.create_file_path(pathname, file_md) + self.move_file(pathname, filepath) + self.renamed_files[pathname] = filepath + self.file_events.append({'mode': self.config.MODE_CREATE, 'filepath': filepath, 'data': file_md, 'is_recorded_show': is_recorded_show}) + else: + self.file_events.append({'mode': self.config.MODE_CREATE, 'filepath': pathname, 'data': file_md, 'is_recorded_show': is_recorded_show}) + + def process_IN_MODIFY(self, event): if not event.dir: self.logger.info("%s: %s", event.maskname, event.pathname) @@ -276,18 +296,48 @@ class AirtimeProcessEvent(ProcessEvent): def process_IN_DELETE(self, event): self.logger.info("%s: %s", event.maskname, event.pathname) if not event.dir: - self.file_events.append({'filepath': event.pathname, 'mode': self.config.MODE_DELETE}) + self.handle_removed_file(event.pathname) + + def handle_removed_file(self, pathname): + self.logger.info("Deleting %s", pathname) + self.file_events.append({'filepath': pathname, 'mode': self.config.MODE_DELETE}) + def process_default(self, event): #self.logger.info("%s: %s", event.maskname, event.pathname) pass + + def execCommandAndReturnStdOut(self, command): + p = Popen(command, shell=True, stdout=PIPE) + stdout = p.communicate()[0] + if p.returncode != 0: + self.logger.warn("command \n%s\n return with a non-zero return value", command) + return stdout + + def write_index_file(self): + #create a new index file. + self.logger.debug("writing new index file") + command = "find %s -type f -iname '*.ogg' -o -iname '*.mp3' -readable" % '/srv/airtime/stor' + stdout = self.execCommandAndReturnStdOut(command) + self.write_file('/var/tmp/airtime' + '/.airtime_media_index', stdout) + return stdout + + def write_file(self, file, string): + f = open(file, 'w') + f.write(string) + f.close() def notifier_loop_callback(self, notifier): + + if len(self.file_events) > 0: + for event in self.file_events: + self.multi_queue.put(event) - for event in self.file_events: - self.multi_queue.put(event) - self.file_events = [] + self.file_events = [] + #no file_events and queue is empty. This is a good time + #to write an index file. + self.write_index_file() #check for any events recieved from Airtime. try: diff --git a/python_apps/media-monitor/airtimefilemonitor/mediaconfig.py b/python_apps/media-monitor/airtimefilemonitor/mediaconfig.py index d1a4e04ae..2b1197e24 100644 --- a/python_apps/media-monitor/airtimefilemonitor/mediaconfig.py +++ b/python_apps/media-monitor/airtimefilemonitor/mediaconfig.py @@ -9,14 +9,14 @@ class AirtimeMediaConfig: MODE_MOVED = "moved" MODE_DELETE = "delete" - def __init__(self): + def __init__(self, logger): # loading config file try: config = ConfigObj('/etc/airtime/media-monitor.cfg') self.cfg = config except Exception, e: - print 'Error loading config: ', e + logger.info('Error loading config: ', e) sys.exit() self.storage_directory = None diff --git a/python_apps/monit/airtime-monit.cfg b/python_apps/monit/airtime-monit.cfg index 647fffd10..22f05e2ca 100644 --- a/python_apps/monit/airtime-monit.cfg +++ b/python_apps/monit/airtime-monit.cfg @@ -13,10 +13,10 @@ with pidfile "/var/run/airtime-liquidsoap.pid" start program = "/etc/init.d/airtime-playout start" with timeout 10 seconds stop program = "/etc/init.d/airtime-playout stop" - check process airtime-media-monitor - with pidfile "/var/run/airtime-media-monitor.pid" - start program = "/etc/init.d/airtime-media-monitor start" with timeout 10 seconds - stop program = "/etc/init.d/airtime-media-monitor stop" +# check process airtime-media-monitor +# with pidfile "/var/run/airtime-media-monitor.pid" +# start program = "/etc/init.d/airtime-media-monitor start" with timeout 10 seconds +# stop program = "/etc/init.d/airtime-media-monitor stop" check process airtime-show-recorder with pidfile "/var/run/airtime-show-recorder.pid" start program = "/etc/init.d/airtime-show-recorder start" with timeout 10 seconds