Congegni
/
sintonia
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'devel' of dev.sourcefabric.org:airtime into devel

This commit is contained in:
James 2012-08-14 17:11:10 -04:00
parent 5ece15c1e8 9880a8732f
commit 92568db01f
46 changed files with 2594 additions and 624 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
airtime_mvc/application/controllers/ApiController.php
View File

@ -456,8 +456,8 @@ class ApiController extends Zend_Controller_Action
$this->view->watched_dirs = $watchedDirsPath;
}
public function dispatchMetadataAction($md, $mode, $dry_run=false)
{
public function dispatchMetadata($md, $mode, $dry_run=false)
{
// Replace this compound result in a hash with proper error handling later on
$return_hash = array();
if ( $dry_run ) { // for debugging we return garbage not to screw around with the db
@ -468,6 +468,8 @@ class ApiController extends Zend_Controller_Action
);
}
Application_Model_Preference::SetImportTimestamp();
Logging::log("--->Mode: $mode || file: {$md['MDATA_KEY_FILEPATH']} ");
Logging::log( $md );
if ($mode == "create") {
$filepath = $md['MDATA_KEY_FILEPATH'];
$filepath = Application_Common_OsPath::normpath($filepath);
@ -557,6 +559,12 @@ class ApiController extends Zend_Controller_Action
// Valid requests must start with mdXXX where XXX represents at least 1 digit
if( !preg_match('/^md\d+$/', $k) ) { continue; }
$info_json = json_decode($raw_json, $assoc=true);
$recorded = $info_json["is_record"];
unset( $info_json["is_record"] );
//unset( $info_json["MDATA_KEY_DURATION"] );
//unset( $info_json["MDATA_KEY_SAMPLERATE"] );
//unset( $info_json["MDATA_KEY_BITRATE"] );
if( !array_key_exists('mode', $info_json) ) { // Log invalid requests
Logging::log("Received bad request(key=$k), no 'mode' parameter. Bad request is:");
Logging::log( $info_json );
@ -579,17 +587,14 @@ class ApiController extends Zend_Controller_Action
// Removing 'mode' key from $info_json might not be necessary...
$mode = $info_json['mode'];
unset( $info_json['mode'] );
$response = $this->dispatchMetadataAction($info_json, $mode, $dry_run=$dry);
// We attack the 'key' back to every request in case the would like to associate
$response = $this->dispatchMetadata($info_json, $mode, $dry_run=$dry);
// We tack on the 'key' back to every request in case the would like to associate
// his requests with particular responses
$response['key'] = $k;
array_push($responses, $response);
// On recorded show requests we do some extra work here. Not sure what it actually is and it
// was usually called from the python api client. Now we just call it straight from the controller to
// save the http roundtrip
if( $info_json['is_record'] and !array_key_exists('error', $response) ) {
$this->uploadRecordedActionParam($info_json['showinstanceid'],$info_json['fileid'],$dry_run=$dry);
}
}
die( json_encode($responses) );
}
@ -609,6 +614,8 @@ class ApiController extends Zend_Controller_Action
}
}
Logging::log( $md );
// update import timestamp
Application_Model_Preference::SetImportTimestamp();
if ($mode == "create") {

28
airtime_mvc/application/models/StoredFile.php
View File

@ -96,6 +96,7 @@ class Application_Model_StoredFile
*/
public function setMetadata($p_md=null)
{
Logging::log("entered setMetadata");
if (is_null($p_md)) {
$this->setDbColMetadata();
} else {
@ -450,8 +451,13 @@ class Application_Model_StoredFile
return $baseUrl."/api/get-media/file/".$this->getId().".".$this->getFileExtension();
}
public static function Insert($md=null)
public static function Insert($md)
{
// save some work by checking if filepath is given right away
if( !isset($md['MDATA_KEY_FILEPATH']) ) {
return null;
}
$file = new CcFiles();
$file->setDbUtime(new DateTime("now", new DateTimeZone("UTC")));
$file->setDbMtime(new DateTime("now", new DateTimeZone("UTC")));
@ -459,22 +465,14 @@ class Application_Model_StoredFile
$storedFile = new Application_Model_StoredFile();
$storedFile->_file = $file;
if (isset($md['MDATA_KEY_FILEPATH'])) {
// removed "//" in the path. Always use '/' for path separator
$filepath = str_replace("//", "/", $md['MDATA_KEY_FILEPATH']);
$res = $storedFile->setFilePath($filepath);
if ($res === -1) {
return null;
}
} else {
// removed "//" in the path. Always use '/' for path separator
$filepath = str_replace("//", "/", $md['MDATA_KEY_FILEPATH']);
$res = $storedFile->setFilePath($filepath);
if ($res === -1) {
return null;
}
if (isset($md)) {
$storedFile->setMetadata($md);
}
return $storedFile;
$storedFile->setMetadata($md);
return $storedFile;
}
/**

15
gen-snapshot.sh
View File

@ -41,11 +41,16 @@ rm -r airtime/python_apps/pypo/liquidsoap_bin/
sed -i '84s:print:#print:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '86s:p = Popen:#p = Popen:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '87s:liq_path:#liq_path:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '89s:if p.returncode:#if p.returncode:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '90s:os.symlink:#os.symlink:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '91s:else:#else:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '92s:print:#print:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '93s:sys.exit:#sys.exit:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '88s:symlink_path:#symlink_path:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '90s:if p.returncode:#if p.returncode:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '91s:tr:#tr:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '92s:os.unlink:#os.unlink:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '93s:except:#except:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '95s:pass:#pass:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '98s:os.symlink:#os.symlink:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '99s:else:#else:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '100s:print:#print:g' airtime/python_apps/pypo/install/pypo-initialize.py
sed -i '101s:sys.exit:#sys.exit:g' airtime/python_apps/pypo/install/pypo-initialize.py
#Remove phing library
rm -r airtime/airtime_mvc/library/phing/

6
python_apps/api_clients/api_client.cfg
View File

@ -51,6 +51,10 @@ set_storage_dir = 'set-storage-dir/format/json/api_key/%%api_key%%/path/%%path%%
# URL to tell Airtime about file system mount change
update_fs_mount = 'update-file-system-mount/format/json/api_key/%%api_key%%'
# URL to commit multiple updates from media monitor at the same time
reload_metadata_group = 'reload-metadata-group/format/json/api_key/%%api_key%%'
# URL to tell Airtime about file system mount change
handle_watched_dir_missing = 'handle-watched-dir-missing/format/json/api_key/%%api_key%%/dir/%%dir%%'
@ -66,8 +70,6 @@ upload_file_url = 'upload-file/format/json/api_key/%%api_key%%'
# URL to commit multiple updates from media monitor at the same time
reload_metadata_group = 'reload-metadata-group/format/json/api_key/%%api_key%%'
#number of retries to upload file if connection problem
upload_retries = 3

108
python_apps/api_clients/api_client.py
View File

@ -1,9 +1,9 @@
###############################################################################
# This file holds the implementations for all the API clients.
#
# If you want to develop a new client, here are some suggestions:
# Get the fetch methods working first, then the push, then the liquidsoap notifier.
# You will probably want to create a script on your server side to automatically
# If you want to develop a new client, here are some suggestions: Get the fetch
# methods working first, then the push, then the liquidsoap notifier. You will
# probably want to create a script on your server side to automatically
# schedule a playlist one minute from the current time.
###############################################################################
import sys
@ -41,6 +41,23 @@ def convert_dict_value_to_utf8(md):
class AirtimeApiClient():
# This is a little hacky fix so that I don't have to pass the config object
# everywhere where AirtimeApiClient needs to be initialized
default_config = None
# the purpose of this custom constructor is to remember which config file
# it was called with. So that after the initial call:
# AirtimeApiClient.create_right_config('/path/to/config')
# All subsequence calls to create_right_config will be with that config
# file
@staticmethod
def create_right_config(log=None,config_path=None):
if config_path: AirtimeApiClient.default_config = config_path
elif (not AirtimeApiClient.default_config):
raise ValueError("Cannot slip config_path attribute when it has \
never been passed yet")
return AirtimeApiClient( logger=None,
config_path=AirtimeApiClient.default_config )
def __init__(self, logger=None,config_path='/etc/airtime/api_client.cfg'):
if logger is None:
self.logger = logging
@ -77,14 +94,15 @@ class AirtimeApiClient():
def get_response_into_file(self, url, block=True):
"""
This function will query the server and download its response directly
into a temporary file. This is useful in the situation where the response
from the server can be huge and we don't want to store it into memory (potentially
causing Python to use hundreds of MB's of memory). By writing into a file we can
then open this file later, and read data a little bit at a time and be very mem
efficient.
into a temporary file. This is useful in the situation where the
response from the server can be huge and we don't want to store it into
memory (potentially causing Python to use hundreds of MB's of memory).
By writing into a file we can then open this file later, and read data
a little bit at a time and be very mem efficient.
The return value of this function is the path of the temporary file. Unless specified using
block = False, this function will block until a successful HTTP 200 response is received.
The return value of this function is the path of the temporary file.
Unless specified using block = False, this function will block until a
successful HTTP 200 response is received.
"""
logger = self.logger
@ -114,7 +132,9 @@ class AirtimeApiClient():
def __get_airtime_version(self):
logger = self.logger
url = "http://%s:%s/%s/%s" % (self.config["base_url"], str(self.config["base_port"]), self.config["api_base"], self.config["version_url"])
url = "http://%s:%s/%s/%s" % (self.config["base_url"],
str(self.config["base_port"]), self.config["api_base"],
self.config["version_url"])
logger.debug("Trying to contact %s", url)
url = url.replace("%%api_key%%", self.config["api_key"])
@ -157,7 +177,8 @@ class AirtimeApiClient():
elif (version[0:3] != AIRTIME_VERSION[0:3]):
if (verbose):
logger.info('Airtime version found: ' + str(version))
logger.info('pypo is at version ' + AIRTIME_VERSION + ' and is not compatible with this version of Airtime.\n')
logger.info('pypo is at version ' + AIRTIME_VERSION +
' and is not compatible with this version of Airtime.\n')
return False
else:
if (verbose):
@ -338,6 +359,8 @@ class AirtimeApiClient():
url = self.construct_url("update_media_url")
url = url.replace("%%mode%%", mode)
self.logger.info("Requesting url %s" % url)
md = convert_dict_value_to_utf8(md)
data = urllib.urlencode(md)
@ -345,6 +368,8 @@ class AirtimeApiClient():
response = self.get_response_from_server(req)
logger.info("update media %s, filepath: %s, mode: %s", response, md['MDATA_KEY_FILEPATH'], mode)
self.logger.info("Received response:")
self.logger.info(response)
try: response = json.loads(response)
except ValueError:
logger.info("Could not parse json from response: '%s'" % response)
@ -367,11 +392,12 @@ class AirtimeApiClient():
def send_media_monitor_requests(self, action_list, dry=False):
"""
Send a gang of media monitor events at a time. actions_list is a list of dictionaries
where every dictionary is representing an action. Every action dict must contain a 'mode'
key that says what kind of action it is and an optional 'is_record' key that says whether
the show was recorded or not. The value of this key does not matter, only if it's present
or not.
Send a gang of media monitor events at a time. actions_list is a list
of dictionaries where every dictionary is representing an action. Every
action dict must contain a 'mode' key that says what kind of action it
is and an optional 'is_record' key that says whether the show was
recorded or not. The value of this key does not matter, only if it's
present or not.
"""
logger = self.logger
try:
@ -386,15 +412,13 @@ class AirtimeApiClient():
# debugging
for action in action_list:
if not 'mode' in action:
self.logger.debug("Warning: Sending a request element without a 'mode'")
self.logger.debug("Warning: Trying to send a request element without a 'mode'")
self.logger.debug("Here is the the request: '%s'" % str(action) )
else:
# We alias the value of is_record to true or false no
# matter what it is based on if it's absent in the action
if 'is_record' in action:
self.logger.debug("Sending a 'recorded' action")
action['is_record'] = 1
else: action['is_record'] = 0
if 'is_record' not in action:
action['is_record'] = 0
valid_actions.append(action)
# Note that we must prefix every key with: mdX where x is a number
# Is there a way to format the next line a little better? The
@ -410,6 +434,7 @@ class AirtimeApiClient():
response = self.get_response_from_server(req)
response = json.loads(response)
return response
except ValueError: raise
except Exception, e:
logger.error('Exception: %s', e)
logger.error("traceback: %s", traceback.format_exc())
@ -422,11 +447,8 @@ class AirtimeApiClient():
def list_all_db_files(self, dir_id):
logger = self.logger
try:
url = "http://%s:%s/%s/%s" % (self.config["base_url"], str(self.config["base_port"]), self.config["api_base"], self.config["list_all_db_files"])
url = url.replace("%%api_key%%", self.config["api_key"])
url = self.construct_url("list_all_db_files")
url = url.replace("%%dir_id%%", dir_id)
response = self.get_response_from_server(url)
response = json.loads(response)
except Exception, e:
@ -440,6 +462,7 @@ class AirtimeApiClient():
return []
def list_all_watched_dirs(self):
# Does this include the stor directory as well?
logger = self.logger
try:
url = "http://%s:%s/%s/%s" % (self.config["base_url"], str(self.config["base_port"]), self.config["api_base"], self.config["list_all_watched_dirs"])
@ -451,6 +474,7 @@ class AirtimeApiClient():
except Exception, e:
response = None
logger.error("Exception: %s", e)
self.logger.debug(traceback.format_exc())
return response
@ -517,10 +541,10 @@ class AirtimeApiClient():
return response
"""
Purpose of this method is to contact the server with a "Hey its me!" message.
This will allow the server to register the component's (component = media-monitor, pypo etc.)
ip address, and later use it to query monit via monit's http service, or download log files
via a http server.
Purpose of this method is to contact the server with a "Hey its me!"
message. This will allow the server to register the component's (component
= media-monitor, pypo etc.) ip address, and later use it to query monit via
monit's http service, or download log files via a http server.
"""
def register_component(self, component):
logger = self.logger
@ -588,8 +612,8 @@ class AirtimeApiClient():
logger.error("traceback: %s", traceback.format_exc())
"""
When watched dir is missing(unplugged or something) on boot up, this function will get called
and will call appropriate function on Airtime.
When watched dir is missing(unplugged or something) on boot up, this
function will get called and will call appropriate function on Airtime.
"""
def handle_watched_dir_missing(self, dir):
logger = self.logger
@ -605,16 +629,13 @@ class AirtimeApiClient():
logger.error('Exception: %s', e)
logger.error("traceback: %s", traceback.format_exc())
"""
Retrive infomations needed on bootstrap time
"""
def get_bootstrap_info(self):
"""
Retrive infomations needed on bootstrap time
"""
logger = self.logger
try:
url = "http://%s:%s/%s/%s" % (self.config["base_url"], str(self.config["base_port"]), self.config["api_base"], self.config["get_bootstrap_info"])
url = url.replace("%%api_key%%", self.config["api_key"])
url = self.construct_url("get_bootstrap_info")
response = self.get_response_from_server(url)
response = json.loads(response)
logger.info("Bootstrap info retrieved %s", response)
@ -626,8 +647,9 @@ class AirtimeApiClient():
def get_files_without_replay_gain_value(self, dir_id):
"""
Download a list of files that need to have their ReplayGain value calculated. This list
of files is downloaded into a file and the path to this file is the return value.
Download a list of files that need to have their ReplayGain value
calculated. This list of files is downloaded into a file and the path
to this file is the return value.
"""
#http://localhost/api/get-files-without-replay-gain/dir_id/1
@ -651,8 +673,8 @@ class AirtimeApiClient():
def update_replay_gain_values(self, pairs):
"""
'pairs' is a list of pairs in (x, y), where x is the file's database row id
and y is the file's replay_gain value in dB
'pairs' is a list of pairs in (x, y), where x is the file's database
row id and y is the file's replay_gain value in dB
"""
#http://localhost/api/update-replay-gain-value/

2
python_apps/media-monitor/airtime-media-monitor
View File

@ -8,7 +8,7 @@ virtualenv_bin="/usr/lib/airtime/airtime_virtualenv/bin/"
media_monitor_path="/usr/lib/airtime/media-monitor/"
media_monitor_script="media_monitor.py"
api_client_path="/usr/lib/airtime/"
api_client_path="/usr/lib/airtime/:/usr/lib/airtime/media-monitor/mm2/"
cd ${media_monitor_path}

13
python_apps/media-monitor/airtimefilemonitor/replaygain.py
View File

@ -13,7 +13,7 @@ def get_process_output(command):
"""
Run subprocess and return stdout
"""
logger.debug(command)
#logger.debug(command)
p = Popen(command, shell=True, stdout=PIPE)
return p.communicate()[0].strip()
@ -40,7 +40,7 @@ def duplicate_file(file_path):
fsrc = open(file_path, 'r')
fdst = tempfile.NamedTemporaryFile(delete=False)
logger.info("Copying %s to %s" % (file_path, fdst.name))
#logger.info("Copying %s to %s" % (file_path, fdst.name))
shutil.copyfileobj(fsrc, fdst)
@ -71,16 +71,17 @@ def get_file_type(file_path):
def calculate_replay_gain(file_path):
"""
This function accepts files of type mp3/ogg/flac and returns a calculated ReplayGain value in dB.
If the value cannot be calculated for some reason, then we default to 0 (Unity Gain).
This function accepts files of type mp3/ogg/flac and returns a calculated
ReplayGain value in dB. If the value cannot be calculated for some reason,
then we default to 0 (Unity Gain).
http://wiki.hydrogenaudio.org/index.php?title=ReplayGain_1.0_specification
"""
try:
"""
Making a duplicate is required because the ReplayGain extraction utilities we use
make unwanted modifications to the file.
Making a duplicate is required because the ReplayGain extraction
utilities we use make unwanted modifications to the file.
"""
search = None

25
python_apps/media-monitor/install/media-monitor-copy-files.py
View File

@ -9,34 +9,35 @@ from configobj import ConfigObj
if os.geteuid() != 0:
print "Please run this as root."
sys.exit(1)
def get_current_script_dir():
current_script_dir = os.path.realpath(__file__)
index = current_script_dir.rindex('/')
return current_script_dir[0:index]
def copy_dir(src_dir, dest_dir):
if (os.path.exists(dest_dir)) and (dest_dir != "/"):
shutil.rmtree(dest_dir)
if not (os.path.exists(dest_dir)):
#print "Copying directory "+os.path.realpath(src_dir)+" to "+os.path.realpath(dest_dir)
shutil.copytree(src_dir, dest_dir)
def create_dir(path):
try:
os.makedirs(path)
# TODO : fix this, at least print the error
except Exception, e:
pass
def get_rand_string(length=10):
return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(length))
PATH_INI_FILE = '/etc/airtime/media-monitor.cfg'
try:
# Absolute path this script is in
current_script_dir = get_current_script_dir()
if not os.path.exists(PATH_INI_FILE):
shutil.copy('%s/../media-monitor.cfg'%current_script_dir, PATH_INI_FILE)
@ -46,22 +47,24 @@ try:
except Exception, e:
print 'Error loading config file: ', e
sys.exit(1)
#copy monit files
shutil.copy('%s/../../monit/monit-airtime-generic.cfg'%current_script_dir, '/etc/monit/conf.d/')
subprocess.call('sed -i "s/\$admin_pass/%s/g" /etc/monit/conf.d/monit-airtime-generic.cfg' % get_rand_string(), shell=True)
shutil.copy('%s/../monit-airtime-media-monitor.cfg'%current_script_dir, '/etc/monit/conf.d/')
#create log dir
create_dir(config['log_dir'])
#copy python files
copy_dir("%s/.."%current_script_dir, config["bin_dir"])
# mm2
mm2_source = os.path.realpath(os.path.join(current_script_dir,
"../../media-monitor2"))
copy_dir(mm2_source, os.path.join( config["bin_dir"], "mm2" ))
#copy init.d script
shutil.copy(config["bin_dir"]+"/airtime-media-monitor-init-d", "/etc/init.d/airtime-media-monitor")
except Exception, e:
print e

11
python_apps/media-monitor/media-monitor.cfg
View File

@ -16,7 +16,16 @@ rabbitmq_password = 'guest'
rabbitmq_vhost = '/'
############################################
# Media-Monitor preferences #
# Media-Monitor preferences #
############################################
check_filesystem_events = 5 #how long to queue up events performed on the files themselves.
check_airtime_events = 30 #how long to queue metadata input from airtime.
# MM2 only:
touch_interval = 5
chunking_number = 450
request_max_wait = 3.0
rmq_event_wait = 0.5
logpath = '/var/log/airtime/media-monitor/media-monitor.log'
index_path = '/var/tmp/airtime/media-monitor/last_index'

143
python_apps/media-monitor/media_monitor.py
View File

@ -1,142 +1,11 @@
# -*- coding: utf-8 -*-
import logging
import time
import logging.config
import sys
import os
import traceback
import locale
from configobj import ConfigObj
from api_clients import api_client as apc
import mm2.mm2 as mm2
from std_err_override import LogWriter
from multiprocessing import Queue as mpQueue
global_cfg = '/etc/airtime/media-monitor.cfg'
api_client_cfg = '/etc/airtime/api_client.cfg'
logging_cfg = '/usr/lib/airtime/media-monitor/logging.cfg'
from threading import Thread
from pyinotify import WatchManager
from airtimefilemonitor.airtimenotifier import AirtimeNotifier
from airtimefilemonitor.mediamonitorcommon import MediaMonitorCommon
from airtimefilemonitor.airtimeprocessevent import AirtimeProcessEvent
from airtimefilemonitor.mediaconfig import AirtimeMediaConfig
from airtimefilemonitor.workerprocess import MediaMonitorWorkerProcess
from airtimefilemonitor.airtimemediamonitorbootstrap import AirtimeMediaMonitorBootstrap
def configure_locale():
logger.debug("Before %s", locale.nl_langinfo(locale.CODESET))
current_locale = locale.getlocale()
if current_locale[1] is None:
logger.debug("No locale currently set. Attempting to get default locale.")
default_locale = locale.getdefaultlocale()
if default_locale[1] is None:
logger.debug("No default locale exists. Let's try loading from /etc/default/locale")
if os.path.exists("/etc/default/locale"):
config = ConfigObj('/etc/default/locale')
lang = config.get('LANG')
new_locale = lang
else:
logger.error("/etc/default/locale could not be found! Please run 'sudo update-locale' from command-line.")
sys.exit(1)
else:
new_locale = default_locale
logger.info("New locale set to: %s", locale.setlocale(locale.LC_ALL, new_locale))
reload(sys)
sys.setdefaultencoding("UTF-8")
current_locale_encoding = locale.getlocale()[1].lower()
logger.debug("sys default encoding %s", sys.getdefaultencoding())
logger.debug("After %s", locale.nl_langinfo(locale.CODESET))
if current_locale_encoding not in ['utf-8', 'utf8']:
logger.error("Need a UTF-8 locale. Currently '%s'. Exiting..." % current_locale_encoding)
sys.exit(1)
# configure logging
try:
logging.config.fileConfig("logging.cfg")
#need to wait for Python 2.7 for this..
#logging.captureWarnings(True)
logger = logging.getLogger()
LogWriter.override_std_err(logger)
except Exception, e:
print 'Error configuring logging: ', e
sys.exit(1)
logger.info("\n\n*** Media Monitor bootup ***\n\n")
try:
configure_locale()
config = AirtimeMediaConfig(logger)
api_client = apc.AirtimeApiClient()
api_client.register_component("media-monitor")
logger.info("Setting up monitor")
response = None
while response is None:
response = api_client.setup_media_monitor()
time.sleep(5)
storage_directory = response["stor"]
watched_dirs = response["watched_dirs"]
logger.info("Storage Directory is: %s", storage_directory)
config.storage_directory = os.path.normpath(storage_directory)
config.imported_directory = os.path.normpath(os.path.join(storage_directory, 'imported'))
config.organize_directory = os.path.normpath(os.path.join(storage_directory, 'organize'))
config.recorded_directory = os.path.normpath(os.path.join(storage_directory, 'recorded'))
config.problem_directory = os.path.normpath(os.path.join(storage_directory, 'problem_files'))
dirs = [config.imported_directory, config.organize_directory, config.recorded_directory, config.problem_directory]
for d in dirs:
if not os.path.exists(d):
os.makedirs(d, 02775)
multi_queue = mpQueue()
logger.info("Initializing event processor")
wm = WatchManager()
mmc = MediaMonitorCommon(config, wm=wm)
pe = AirtimeProcessEvent(queue=multi_queue, airtime_config=config, wm=wm, mmc=mmc, api_client=api_client)
bootstrap = AirtimeMediaMonitorBootstrap(logger, pe, api_client, mmc, wm, config)
bootstrap.scan()
notifier = AirtimeNotifier(wm, pe, read_freq=0, timeout=0, airtime_config=config, api_client=api_client, bootstrap=bootstrap, mmc=mmc)
notifier.coalesce_events()
#create 5 worker threads
wp = MediaMonitorWorkerProcess(config, mmc)
for i in range(5):
threadName = "Thread #%d" % i
t = Thread(target=wp.process_file_events, name=threadName, args=(multi_queue, notifier))
t.start()
wdd = notifier.watch_directory(storage_directory)
logger.info("Added watch to %s", storage_directory)
logger.info("wdd result %s", wdd[storage_directory])
for dir in watched_dirs:
wdd = notifier.watch_directory(dir)
logger.info("Added watch to %s", dir)
logger.info("wdd result %s", wdd[dir])
notifier.loop(callback=pe.notifier_loop_callback)
except KeyboardInterrupt:
notifier.stop()
logger.info("Keyboard Interrupt")
except Exception, e:
logger.error('Exception: %s', e)
logger.error("traceback: %s", traceback.format_exc())
mm2.main( global_cfg, api_client_cfg, logging_cfg )

142
python_apps/media-monitor/mm1.py Normal file
View File

@ -0,0 +1,142 @@
# -*- coding: utf-8 -*-
import time
import logging.config
import sys
import os
import traceback
import locale
from configobj import ConfigObj
from api_clients import api_client as apc
from std_err_override import LogWriter
from multiprocessing import Queue as mpQueue
from threading import Thread
from pyinotify import WatchManager
from airtimefilemonitor.airtimenotifier import AirtimeNotifier
from airtimefilemonitor.mediamonitorcommon import MediaMonitorCommon
from airtimefilemonitor.airtimeprocessevent import AirtimeProcessEvent
from airtimefilemonitor.mediaconfig import AirtimeMediaConfig
from airtimefilemonitor.workerprocess import MediaMonitorWorkerProcess
from airtimefilemonitor.airtimemediamonitorbootstrap import AirtimeMediaMonitorBootstrap
def configure_locale():
logger.debug("Before %s", locale.nl_langinfo(locale.CODESET))
current_locale = locale.getlocale()
if current_locale[1] is None:
logger.debug("No locale currently set. Attempting to get default locale.")
default_locale = locale.getdefaultlocale()
if default_locale[1] is None:
logger.debug("No default locale exists. Let's try loading from /etc/default/locale")
if os.path.exists("/etc/default/locale"):
config = ConfigObj('/etc/default/locale')
lang = config.get('LANG')
new_locale = lang
else:
logger.error("/etc/default/locale could not be found! Please run 'sudo update-locale' from command-line.")
sys.exit(1)
else:
new_locale = default_locale
logger.info("New locale set to: %s", locale.setlocale(locale.LC_ALL, new_locale))
reload(sys)
sys.setdefaultencoding("UTF-8")
current_locale_encoding = locale.getlocale()[1].lower()
logger.debug("sys default encoding %s", sys.getdefaultencoding())
logger.debug("After %s", locale.nl_langinfo(locale.CODESET))
if current_locale_encoding not in ['utf-8', 'utf8']:
logger.error("Need a UTF-8 locale. Currently '%s'. Exiting..." % current_locale_encoding)
sys.exit(1)
# configure logging
try:
logging.config.fileConfig("logging.cfg")
#need to wait for Python 2.7 for this..
#logging.captureWarnings(True)
logger = logging.getLogger()
LogWriter.override_std_err(logger)
except Exception, e:
print 'Error configuring logging: ', e
sys.exit(1)
logger.info("\n\n*** Media Monitor bootup ***\n\n")
try:
configure_locale()
config = AirtimeMediaConfig(logger)
api_client = apc.AirtimeApiClient()
api_client.register_component("media-monitor")
logger.info("Setting up monitor")
response = None
while response is None:
response = api_client.setup_media_monitor()
time.sleep(5)
storage_directory = response["stor"]
watched_dirs = response["watched_dirs"]
logger.info("Storage Directory is: %s", storage_directory)
config.storage_directory = os.path.normpath(storage_directory)
config.imported_directory = os.path.normpath(os.path.join(storage_directory, 'imported'))
config.organize_directory = os.path.normpath(os.path.join(storage_directory, 'organize'))
config.recorded_directory = os.path.normpath(os.path.join(storage_directory, 'recorded'))
config.problem_directory = os.path.normpath(os.path.join(storage_directory, 'problem_files'))
dirs = [config.imported_directory, config.organize_directory, config.recorded_directory, config.problem_directory]
for d in dirs:
if not os.path.exists(d):
os.makedirs(d, 02775)
multi_queue = mpQueue()
logger.info("Initializing event processor")
wm = WatchManager()
mmc = MediaMonitorCommon(config, wm=wm)
pe = AirtimeProcessEvent(queue=multi_queue, airtime_config=config, wm=wm, mmc=mmc, api_client=api_client)
bootstrap = AirtimeMediaMonitorBootstrap(logger, pe, api_client, mmc, wm, config)
bootstrap.scan()
notifier = AirtimeNotifier(wm, pe, read_freq=0, timeout=0, airtime_config=config, api_client=api_client, bootstrap=bootstrap, mmc=mmc)
notifier.coalesce_events()
#create 5 worker threads
wp = MediaMonitorWorkerProcess(config, mmc)
for i in range(5):
threadName = "Thread #%d" % i
t = Thread(target=wp.process_file_events, name=threadName, args=(multi_queue, notifier))
t.start()
wdd = notifier.watch_directory(storage_directory)
logger.info("Added watch to %s", storage_directory)
logger.info("wdd result %s", wdd[storage_directory])
for dir in watched_dirs:
wdd = notifier.watch_directory(dir)
logger.info("Added watch to %s", dir)
logger.info("wdd result %s", wdd[dir])
notifier.loop(callback=pe.notifier_loop_callback)
except KeyboardInterrupt:
notifier.stop()
logger.info("Keyboard Interrupt")
except Exception, e:
logger.error('Exception: %s', e)
logger.error("traceback: %s", traceback.format_exc())

1
python_apps/media-monitor2/__init__.py Normal file
View File

@ -0,0 +1 @@

171
python_apps/media-monitor2/media/monitor/airtime.py
View File

@ -1,41 +1,47 @@
# -*- coding: utf-8 -*-
from kombu.messaging import Exchange, Queue, Consumer
from kombu.messaging import Exchange, Queue, Consumer
from kombu.connection import BrokerConnection
import json
import os
import copy
from media.monitor.log import Loggable
from media.monitor.exceptions import BadSongFile
from media.monitor.metadata import Metadata
from media.monitor.log import Loggable
from media.monitor.syncdb import AirtimeDB
from media.monitor.exceptions import DirectoryIsNotListed
from media.monitor.bootstrap import Bootstrapper
from media.monitor.listeners import FileMediator
# Do not confuse with media monitor 1's AirtimeNotifier class that more related
# to pyinotify's Notifier class. AirtimeNotifier just notifies when events come
# from Airtime itself. I.E. changes made in the web UI that must be updated
# through media monitor
from api_clients import api_client as apc
class AirtimeNotifier(Loggable):
"""
AirtimeNotifier is responsible for interecepting RabbitMQ messages and feeding them to the
event_handler object it was initialized with. The only thing it does to the messages is parse
them from json
AirtimeNotifier is responsible for interecepting RabbitMQ messages and
feeding them to the event_handler object it was initialized with. The only
thing it does to the messages is parse them from json
"""
def __init__(self, cfg, message_receiver):
self.cfg = cfg
try:
self.handler = message_receiver
self.logger.info("Initializing RabbitMQ message consumer...")
schedule_exchange = Exchange("airtime-media-monitor", "direct", durable=True, auto_delete=True)
schedule_queue = Queue("media-monitor", exchange=schedule_exchange, key="filesystem")
#self.connection = BrokerConnection(cfg["rabbitmq_host"], cfg["rabbitmq_user"],
#cfg["rabbitmq_password"], cfg["rabbitmq_vhost"])
connection = BrokerConnection(cfg["rabbitmq_host"], cfg["rabbitmq_user"],
cfg["rabbitmq_password"], cfg["rabbitmq_vhost"])
channel = connection.channel()
schedule_exchange = Exchange("airtime-media-monitor", "direct",
durable=True, auto_delete=True)
schedule_queue = Queue("media-monitor", exchange=schedule_exchange,
key="filesystem")
self.connection = BrokerConnection(cfg["rabbitmq_host"],
cfg["rabbitmq_user"], cfg["rabbitmq_password"],
cfg["rabbitmq_vhost"])
channel = self.connection.channel()
consumer = Consumer(channel, schedule_queue)
consumer.register_callback(self.handle_message)
consumer.consume()
self.logger.info("Initialized RabbitMQ consumer.")
except Exception as e:
self.logger.info("Failed to initialize RabbitMQ consumer")
self.logger.error(e)
raise
def handle_message(self, body, message):
"""
@ -49,53 +55,136 @@ class AirtimeNotifier(Loggable):
m = json.loads(message.body)
self.handler.message(m)
class AirtimeMessageReceiver(Loggable):
def __init__(self, cfg):
def __init__(self, cfg, manager):
self.dispatch_table = {
'md_update' : self.md_update,
'new_watch' : self.new_watch,
'md_update' : self.md_update,
'new_watch' : self.new_watch,
'remove_watch' : self.remove_watch,
'rescan_watch' : self.rescan_watch,
'change_stor' : self.change_storage,
'file_delete' : self.file_delete,
'change_stor' : self.change_storage,
'file_delete' : self.file_delete,
}
self.cfg = cfg
self.cfg = cfg
self.manager = manager
def message(self, msg):
"""
This method is called by an AirtimeNotifier instance that consumes the Rabbit MQ events
that trigger this. The method return true when the event was executed and false when it
wasn't
This method is called by an AirtimeNotifier instance that
consumes the Rabbit MQ events that trigger this. The method
return true when the event was executed and false when it wasn't.
"""
msg = copy.deepcopy(msg)
if msg['event_type'] in self.dispatch_table:
evt = msg['event_type']
del msg['event_type']
self.logger.info("Handling RabbitMQ message: '%s'" % evt)
self.execute_message(evt,msg)
self._execute_message(evt,msg)
return True
else:
self.logger.info("Received invalid message with 'event_type': '%s'" % msg['event_type'])
self.logger.info("Received invalid message with 'event_type': '%s'"
% msg['event_type'])
self.logger.info("Message details: %s" % str(msg))
return False
def execute_message(self,evt,message):
def _execute_message(self,evt,message):
self.dispatch_table[evt](message)
def supported_messages(self):
return self.dispatch_table.keys()
def __request_now_bootstrap(self, directory_id=None, directory=None):
if (not directory_id) and (not directory):
raise ValueError("You must provide either directory_id or \
directory")
sdb = AirtimeDB(apc.AirtimeApiClient.create_right_config())
if directory : directory = os.path.normpath(directory)
if directory_id == None : directory_id = sdb.to_id(directory)
if directory == None : directory = sdb.to_directory(directory_id)
try:
bs = Bootstrapper( sdb, self.manager.watch_signal() )
bs.flush_watch( directory=directory, last_ran=self.cfg.last_ran() )
except Exception as e:
self.fatal_exception("Exception bootstrapping: (dir,id)=(%s,%s)" %
(directory, directory_id), e)
raise DirectoryIsNotListed(directory, cause=e)
# Handler methods - Should either fire the events directly with
# pydispatcher or do the necessary changes on the filesystem that will fire
# the events
def md_update(self, msg):
pass
self.logger.info("Updating metadata for: '%s'" %
msg['MDATA_KEY_FILEPATH'])
md_path = msg['MDATA_KEY_FILEPATH']
try: Metadata.write_unsafe(path=md_path, md=msg)
except BadSongFile as e:
self.logger.info("Cannot find metadata file: '%s'" % e.path)
except Exception as e:
# TODO : add md_path to problem path or something?
self.fatal_exception("Unknown error when writing metadata to: '%s'"
% md_path, e)
def new_watch(self, msg):
pass
self.logger.info("Creating watch for directory: '%s'" %
msg['directory'])
if not os.path.exists(msg['directory']):
try: os.makedirs(msg['directory'])
except Exception as e:
self.fatal_exception("Failed to create watched dir '%s'" %
msg['directory'],e)
else: self.new_watch(msg)
else:
self.__request_now_bootstrap( directory=msg['directory'] )
self.manager.add_watch_directory(msg['directory'])
def remove_watch(self, msg):
pass
self.logger.info("Removing watch from directory: '%s'" %
msg['directory'])
self.manager.remove_watch_directory(msg['directory'])
def rescan_watch(self, msg):
pass
self.logger.info("Trying to rescan watched directory: '%s'" %
msg['directory'])
try:
# id is always an integer but in the dictionary the key is always a
# string
self.__request_now_bootstrap( unicode(msg['id']) )
except DirectoryIsNotListed as e:
self.fatal_exception("Bad rescan request", e)
except Exception as e:
self.fatal_exception("Bad rescan request. Unknown error.", e)
else:
self.logger.info("Successfully re-scanned: '%s'" % msg['directory'])
def change_storage(self, msg):
pass
new_storage_directory = msg['directory']
self.manager.change_storage_root(new_storage_directory)
for to_bootstrap in [ self.manager.get_recorded_path(),
self.manager.get_imported_path() ]:
self.__request_now_bootstrap( directory=to_bootstrap )
def file_delete(self, msg):
pass
# Deletes should be requested only from imported folder but we
# don't verify that. Security risk perhaps?
# we only delete if we are passed the special delete flag that is
# necessary with every "delete_file" request
if not msg['delete']:
self.logger.info("No clippy confirmation, ignoring event. \
Out of curiousity we will print some details.")
self.logger.info(msg)
return
if os.path.exists(msg['filepath']):
try:
self.logger.info("Attempting to delete '%s'" %
msg['filepath'])
# We use FileMediator to ignore any paths with
# msg['filepath'] so that we do not send a duplicate delete
# request that we'd normally get form pyinotify. But right
# now event contractor would take care of this sort of
# thing anyway so this might not be necessary after all
FileMediator.ignore(msg['filepath'])
os.unlink(msg['filepath'])
# Verify deletion:
if not os.path.exists(msg['filepath']):
self.logger.info("Successfully deleted: '%s'" %
msg['filepath'])
except Exception as e:
self.fatal_exception("Failed to delete '%s'" % msg['filepath'],
e)
else: # validation for filepath existence failed
self.logger.info("Attempting to delete file '%s' that does not \
exist. Full request:" % msg['filepath'])
self.logger.info(msg)

96
python_apps/media-monitor2/media/monitor/bootstrap.py
View File

@ -1,59 +1,61 @@
import os
from pydispatch import dispatcher
from media.monitor.events import OrganizeFile, NewFile, DeleteFile
from media.monitor.log import Loggable
from pydispatch import dispatcher
from media.monitor.events import NewFile, DeleteFile, ModifyFile
from media.monitor.log import Loggable
import media.monitor.pure as mmp
class Bootstrapper(Loggable):
"""
Bootstrapper reads all the info in the filesystem flushes organize
events and watch events
Bootstrapper reads all the info in the filesystem flushes organize events
and watch events
"""
def __init__(self,db,last_ran,org_channels,watch_channels):
self.db = db
self.org_channels = org_channels
self.watch_channels = watch_channels
self.last_ran = last_ran
def flush_organize(self):
def __init__(self,db,watch_signal):
"""
walks the organize directories and sends an organize event for every file manually
db - AirtimeDB object; small layer over api client
last_ran - last time the program was ran.
watch_signal - the signals should send events for every file on.
"""
flushed = 0
for pc in self.org_channels:
for f in mmp.walk_supported(pc.path, clean_empties=True):
self.logger.info("Bootstrapping: File in 'organize' directory: '%s'" % f)
dispatcher.send(signal=pc.signal, sender=self, event=OrganizeFile(f))
flushed += 1
self.logger.info("Flushed organized directory with %d files" % flushed)
self.db = db
self.watch_signal = watch_signal
def flush_watch(self):
def flush_all(self, last_ran):
"""
Syncs the file system into the database. Walks over deleted/new/modified files since
the last run in mediamonitor and sends requests to make the database consistent with
file system
bootstrap every single watched directory. only useful at startup note
that because of the way list_directories works we also flush the import
directory as well I think
"""
songs = set()
modded = deleted = 0
for pc in self.watch_channels:
for f in mmp.walk_supported(pc.path, clean_empties=False):
songs.add(f)
if os.path.getmtime(f) > self.last_ran:
modded += 1
dispatcher.send(signal=pc.signal, sender=self, event=DeleteFile(f))
dispatcher.send(signal=pc.signal, sender=self, event=NewFile(f))
# Want all files in the database that are not in the filesystem
for to_delete in self.db.exclude(songs):
for pc in self.watch_channels:
if os.path.commonprefix([pc.path, to_delete]) == pc.path:
dispatcher.send(signal=pc.signal, sender=self, event=DeleteFile(f))
os.remove(to_delete)
deleted += 1
break
else:
self.logger.info("Error, could not find watch directory of would be deleted \
file '%s'" % to_delete)
self.logger.info("Flushed watch directories. (modified, deleted) = (%d, %d)"
% (modded, deleted) )
for d in self.db.list_storable_paths(): self.flush_watch(d, last_ran)
def flush_watch(self, directory, last_ran):
"""
flush a single watch/imported directory. useful when wanting to to
rescan, or add a watched/imported directory
"""
songs = set([])
added = modded = deleted = 0
for f in mmp.walk_supported(directory, clean_empties=False):
songs.add(f)
# We decide whether to update a file's metadata by checking its
# system modification date. If it's above the value self.last_ran
# which is passed to us that means media monitor wasn't aware when
# this changes occured in the filesystem hence it will send the
# correct events to sync the database with the filesystem
if os.path.getmtime(f) > last_ran:
modded += 1
dispatcher.send(signal=self.watch_signal, sender=self,
event=ModifyFile(f))
db_songs = set(( song for song in self.db.directory_get_files(directory)
if mmp.sub_path(directory,song) ))
# Get all the files that are in the database but in the file
# system. These are the files marked for deletions
for to_delete in db_songs.difference(songs):
dispatcher.send(signal=self.watch_signal, sender=self,
event=DeleteFile(to_delete))
deleted += 1
for to_add in songs.difference(db_songs):
dispatcher.send(signal=self.watch_signal, sender=self,
event=NewFile(to_add))
added += 1
self.logger.info( "Flushed watch directory (%s). \
(added, modified, deleted) = (%d, %d, %d)"
% (directory, added, modded, deleted) )

16
python_apps/media-monitor2/media/monitor/config.py
View File

@ -1,22 +1,21 @@
# -*- coding: utf-8 -*-
import os
from configobj import ConfigObj
import copy
from configobj import ConfigObj
from media.monitor.log import Loggable
from media.monitor.exceptions import NoConfigFile, ConfigAccessViolation
import media.monitor.pure as mmp
class MMConfig(Loggable):
class MMConfig(object):
def __init__(self, path):
if not os.path.exists(path):
self.logger.error("Configuration file does not exist. Path: '%s'" % path)
raise NoConfigFile(path)
self.cfg = ConfigObj(path)
def __getitem__(self, key):
"""
We always return a copy of the config item to prevent callers from doing any modifications
through the returned objects methods
We always return a copy of the config item to prevent callers from
doing any modifications through the returned objects methods
"""
return copy.deepcopy(self.cfg[key])
@ -29,8 +28,9 @@ class MMConfig(Loggable):
def save(self): self.cfg.write()
def last_ran(self):
return mmp.last_modified(self.cfg['index_path'])
# Remove this after debugging...
def haxxor_set(self, key, value): self.cfg[key] = value
def haxxor_get(self, key): return self.cfg[key]

50
python_apps/media-monitor2/media/monitor/eventcontractor.py Normal file
View File

@ -0,0 +1,50 @@
from media.monitor.log import Loggable
from media.monitor.events import DeleteFile
class EventContractor(Loggable):
"""
This class is responsible for "contracting" events together to ease the
load on airtime. It does this by morphing old events into newer ones
"""
def __init__(self):
self.store = {}
def event_registered(self, evt):
"""
returns true if the event is registered which means that there is
another "unpacked" event somewhere out there with the same path
"""
return evt.path in self.store
def get_old_event(self, evt):
"""
get the previously registered event with the same path as 'evt'
"""
return self.store[ evt.path ]
def register(self, evt):
"""
Returns true if event was actually registered. This means that
no old events were touched. On the other hand returns false if
some other event in the storage was morphed into this newer one.
Which should mean that the old event should be discarded.
"""
if self.event_registered(evt):
old_e = self.get_old_event(evt)
# TODO : Perhaps there are other events that we can "contract"
# together
# If two events are of the same type we can safely discard the old
# one
if evt.__class__ == old_e.__class__:
old_e.morph_into(evt)
return False
# delete overrides any other event
elif isinstance(evt, DeleteFile):
old_e.morph_into(evt)
return False
evt.add_safe_pack_hook( lambda : self.__unregister(evt) )
self.store[ evt.path ] = evt
return True # We actually added something, hence we return true.
# events are unregistered automatically no need to screw around with them
def __unregister(self, evt): del self.store[evt.path]

19
python_apps/media-monitor2/media/monitor/eventdrainer.py Normal file
View File

@ -0,0 +1,19 @@
import socket
from media.monitor.log import Loggable
from media.monitor.toucher import RepeatTimer
class EventDrainer(Loggable):
"""
Flushes events from RabbitMQ that are sent from airtime every
certain amount of time
"""
def __init__(self, connection, interval=1):
def cb():
try : connection.drain_events(timeout=0.3)
except socket.timeout : pass
except Exception as e :
self.fatal_exception("Error flushing events", e)
t = RepeatTimer(interval, cb)
t.daemon = True
t.start()

195
python_apps/media-monitor2/media/monitor/events.py
View File

@ -1,57 +1,188 @@
# -*- coding: utf-8 -*-
import os
import mutagen
import abc
import media.monitor.pure as mmp
from media.monitor.pure import LazyProperty
from media.monitor.metadata import Metadata
from media.monitor.log import Loggable
from media.monitor.exceptions import BadSongFile
from media.monitor.pure import LazyProperty
class PathChannel(object):
"""a dumb struct; python has no record types"""
def __init__(self, signal, path):
self.signal = signal
self.path = path
# It would be good if we could parameterize this class by the attribute
# that would contain the path to obtain the meta data. But it would be too much
# work for little reward
class HasMetaData(object):
# TODO : add documentation for HasMetaData
__metaclass__ = abc.ABCMeta
# doing weird bullshit here because python constructors only
# call the constructor of the leftmost superclass.
@LazyProperty
def metadata(self):
# Normally this would go in init but we don't like
# relying on consumers of this behaviour to have to call
# the constructor
try: f = mutagen.File(self.path, easy=True)
except Exception: raise BadSongFile(self.path)
metadata = {}
for k,v in f:
# Special handling of attributes here
if isinstance(v, list):
if len(v) == 1: metadata[k] = v[0]
else: raise Exception("Weird mutagen %s:%s" % (k,str(v)))
else: metadata[k] = v
return metadata
class EventRegistry(object):
"""
This class's main use is to keep track all events with a cookie attribute.
This is done mainly because some events must be 'morphed' into other events
because we later detect that they are move events instead of delete events.
"""
registry = {}
@staticmethod
def register(evt): EventRegistry.registry[evt.cookie] = evt
@staticmethod
def unregister(evt): del EventRegistry.registry[evt.cookie]
@staticmethod
def registered(evt): return evt.cookie in EventRegistry.registry
@staticmethod
def matching(evt):
event = EventRegistry.registry[evt.cookie]
# Want to disallow accessing the same event twice
EventRegistry.unregister(event)
return event
def __init__(self,*args,**kwargs):
raise Exception("You can instantiate this class. Must only use class \
methods")
class BaseEvent(object):
class HasMetaData(object):
"""
Any class that inherits from this class gains the metadata attribute that
loads metadata from the class's 'path' attribute. This is done lazily so
there is no performance penalty to inheriting from this and subsequent
calls to metadata are cached
"""
__metaclass__ = abc.ABCMeta
@LazyProperty
def metadata(self): return Metadata(self.path)
class BaseEvent(Loggable):
__metaclass__ = abc.ABCMeta
def __init__(self, raw_event):
# TODO : clean up this idiotic hack
# we should use keyword constructors instead of this behaviour checking
# bs to initialize BaseEvent
if hasattr(raw_event,"pathname"):
self.__raw_event = raw_event
self._raw_event = raw_event
self.path = os.path.normpath(raw_event.pathname)
else: self.path = raw_event
self._pack_hook = lambda: None # no op
# into another event
def exists(self): return os.path.exists(self.path)
@LazyProperty
def cookie(self): return getattr( self._raw_event, 'cookie', None )
def __str__(self):
return "Event. Path: %s" % self.__raw_event.pathname
return "Event(%s). Path(%s)" % ( self.path, self.__class__.__name__)
def add_safe_pack_hook(self,k):
"""
adds a callable object (function) that will be called after the event
has been "safe_packed"
"""
self._pack_hook = k
# As opposed to unsafe_pack...
def safe_pack(self):
"""
returns exceptions instead of throwing them to be consistent with
events that must catch their own BadSongFile exceptions since generate
a set of exceptions instead of a single one
"""
# pack will only throw an exception if it processes one file but this
# is a little bit hacky
try:
ret = self.pack()
self._pack_hook()
return ret
except BadSongFile as e: return [e]
# nothing to see here, please move along
def morph_into(self, evt):
self.logger.info("Morphing %s into %s" % ( str(self), str(evt) ) )
self._raw_event = evt
self.path = evt.path
self.__class__ = evt.__class__
self.add_safe_pack_hook(evt._pack_hook)
return self
class FakePyinotify(object):
"""
sometimes we must create our own pyinotify like objects to
instantiate objects from the classes below whenever we want to turn
a single event into multiple events
"""
def __init__(self, path):
self.pathname = path
class OrganizeFile(BaseEvent, HasMetaData):
def __init__(self, *args, **kwargs): super(OrganizeFile, self).__init__(*args, **kwargs)
def __init__(self, *args, **kwargs):
super(OrganizeFile, self).__init__(*args, **kwargs)
def pack(self):
raise AttributeError("You can't send organize events to airtime!!!")
class NewFile(BaseEvent, HasMetaData):
def __init__(self, *args, **kwargs): super(NewFile, self).__init__(*args, **kwargs)
def __init__(self, *args, **kwargs):
super(NewFile, self).__init__(*args, **kwargs)
def pack(self):
"""
packs turns an event into a media monitor request
"""
req_dict = self.metadata.extract()
req_dict['mode'] = u'create'
req_dict['MDATA_KEY_FILEPATH'] = unicode( self.path )
return [req_dict]
class DeleteFile(BaseEvent):
def __init__(self, *args, **kwargs): super(DeleteFile, self).__init__(*args, **kwargs)
def __init__(self, *args, **kwargs):
super(DeleteFile, self).__init__(*args, **kwargs)
def pack(self):
req_dict = {}
req_dict['mode'] = u'delete'
req_dict['MDATA_KEY_FILEPATH'] = unicode( self.path )
return [req_dict]
class MoveFile(BaseEvent, HasMetaData):
"""
Path argument should be the new path of the file that was moved
"""
def __init__(self, *args, **kwargs):
super(MoveFile, self).__init__(*args, **kwargs)
def pack(self):
req_dict = {}
req_dict['mode'] = u'moved'
req_dict['MDATA_KEY_MD5'] = self.metadata.extract()['MDATA_KEY_MD5']
req_dict['MDATA_KEY_FILEPATH'] = unicode( self.path )
return [req_dict]
class ModifyFile(BaseEvent, HasMetaData):
def __init__(self, *args, **kwargs):
super(ModifyFile, self).__init__(*args, **kwargs)
def pack(self):
req_dict = self.metadata.extract()
req_dict['mode'] = u'modify'
# path to directory that is to be removed
req_dict['MDATA_KEY_FILEPATH'] = unicode( self.path )
return [req_dict]
def map_events(directory, constructor):
# -unknown-path should not appear in the path here but more testing
# might be necessary
for f in mmp.walk_supported(directory, clean_empties=False):
try:
for e in constructor( FakePyinotify(f) ).pack(): yield e
except BadSongFile as e: yield e
class DeleteDir(BaseEvent):
def __init__(self, *args, **kwargs):
super(DeleteDir, self).__init__(*args, **kwargs)
def pack(self):
return map_events( self.path, DeleteFile )
class MoveDir(BaseEvent):
def __init__(self, *args, **kwargs):
super(MoveDir, self).__init__(*args, **kwargs)
def pack(self):
return map_events( self.path, MoveFile )
class DeleteDirWatch(BaseEvent):
def __init__(self, *args, **kwargs):
super(DeleteDirWatch, self).__init__(*args, **kwargs)
def pack(self):
req_dict = {}
req_dict['mode'] = u'delete_dir'
req_dict['MDATA_KEY_FILEPATH'] = unicode( self.path + "/" )
return [req_dict]

38
python_apps/media-monitor2/media/monitor/exceptions.py
View File

@ -1,5 +1,4 @@
# -*- coding: utf-8 -*-
class BadSongFile(Exception):
def __init__(self, path): self.path = path
def __str__(self): return "Can't read %s" % self.path
@ -12,3 +11,40 @@ class NoConfigFile(Exception):
class ConfigAccessViolation(Exception):
def __init__(self,key): self.key = key
def __str__(self): return "You must not access key '%s' directly" % self.key
class FailedToSetLocale(Exception):
def __str__(self): return "Failed to set locale"
class FailedToObtainLocale(Exception):
def __init__(self, path, cause):
self.path = path
self.cause = cause
def __str__(self): return "Failed to obtain locale from '%s'" % self.path
class CouldNotCreateIndexFile(Exception):
"""exception whenever index file cannot be created"""
def __init__(self, path, cause):
self.path = path
self.cause = cause
def __str__(self): return "Failed to create touch file '%s'" % self.path
class DirectoryIsNotListed(Exception):
def __init__(self,dir_id,cause=None):
self.dir_id = dir_id
self.cause = cause
def __str__(self):
return "%d was not listed as a directory in the database" % self.dir_id
class FailedToCreateDir(Exception):
def __init__(self,path, parent):
self.path = path
self.parent = parent
def __str__(self): return "Failed to create path '%s'" % self.path
class NoDirectoryInAirtime(Exception):
def __init__(self,path, does_exist):
self.path = path
self.does_exist = does_exist
def __str__(self):
return "Directory '%s' does not exist in Airtime.\n \
However: %s do exist." % (self.path, self.does_exist)

40
python_apps/media-monitor2/media/monitor/handler.py
View File

@ -3,6 +3,7 @@ from pydispatch import dispatcher
import abc
from media.monitor.log import Loggable
import media.monitor.pure as mmp
# Defines the handle interface
class Handles(object):
@ -10,32 +11,49 @@ class Handles(object):
@abc.abstractmethod
def handle(self, sender, event, *args, **kwargs): pass
# TODO : remove the code duplication between ReportHandler and
# ProblemFileHandler. Namely the part where both initialize pydispatch
# TODO : Investigate whether weak reffing in dispatcher.connect could possibly
# cause a memory leak
class ReportHandler(Handles):
"""
A handler that can also report problem files when things go wrong
through the report_problem_file routine
"""
__metaclass__ = abc.ABCMeta
def __init__(self, signal):
def __init__(self, signal, weak=False):
self.signal = signal
self.report_signal = "badfile"
def dummy(sender, event): self.handle(sender,event)
dispatcher.connect(dummy, signal=signal, sender=dispatcher.Any, weak=False)
dispatcher.connect(dummy, signal=signal, sender=dispatcher.Any,
weak=weak)
def report_problem_file(self, event, exception=None):
dispatcher.send(signal=self.report_signal, sender=self, event=event, exception=exception)
dispatcher.send(signal=self.report_signal, sender=self, event=event,
exception=exception)
class ProblemFileHandler(Handles, Loggable):
"""
Responsible for answering to events passed through the 'badfile'
signal. Moves the problem file passed to the designated directory.
"""
def __init__(self, channel, **kwargs):
self.channel = channel
self.signal = self.channel.signal
self.problem_dir = self.channel.path
def dummy(sender, event, exception): self.handle(sender, event, exception)
dispatcher.connect(dummy, signal=self.signal, sender=dispatcher.Any, weak=False)
def dummy(sender, event, exception):
self.handle(sender, event, exception)
dispatcher.connect(dummy, signal=self.signal, sender=dispatcher.Any,
weak=False)
mmp.create_dir( self.problem_dir )
self.logger.info("Initialized problem file handler. Problem dir: '%s'" %
self.problem_dir)
def handle(self, sender, event, exception=None):
self.logger.info("Received problem file: '%s'. Supposed to move it somewhere", event.path)
# TODO : not actually moving it anywhere yet
# TODO : use the exception parameter for something
self.logger.info("Received problem file: '%s'. Supposed to move it to \
problem dir", event.path)
try: mmp.move_to_dir(dir_path=self.problem_dir, file_path=event.path)
except Exception as e:
self.logger.info("Could not move file: '%s' to problem dir: '%s'" %
(event.path, self.problem_dir))
self.logger.info("Exception: %s" % str(e))

154
python_apps/media-monitor2/media/monitor/listeners.py
View File

@ -4,57 +4,157 @@ from pydispatch import dispatcher
import media.monitor.pure as mmp
from media.monitor.pure import IncludeOnly
from media.monitor.events import OrganizeFile, NewFile, DeleteFile
from media.monitor.events import OrganizeFile, NewFile, MoveFile, DeleteFile, \
DeleteDir, EventRegistry, MoveDir,\
DeleteDirWatch
from media.monitor.log import Loggable, get_logger
# We attempt to document a list of all special cases and hacks that the
# following classes should be able to handle.
# TODO : implement all of the following special cases
# following classes should be able to handle. TODO : implement all of
# the following special cases
#
# - Recursive directories being added to organized dirs are not handled
# properly as they only send a request for the dir and not for every file. Also
# more hacks are needed to check that the directory finished moving/copying?
# properly as they only send a request for the dir and not for every
# file. Also more hacks are needed to check that the directory finished
# moving/copying?
#
# - In the case when a 'watched' directory's subdirectory is delete we should
# send a special request telling ApiController to delete a whole dir. This is
# done becasue pyinotify will not send an individual file delete event for
# every file in that directory
# - In the case when a 'watched' directory's subdirectory is delete we
# should send a special request telling ApiController to delete a whole
# dir. This is done becasue pyinotify will not send an individual file
# delete event for every file in that directory
#
# - Special move events are required whenever a file is moved from a 'watched'
# directory into another 'watched' directory (or subdirectory). In this case we
# must identify the file by its md5 signature instead of it's filepath like we
# usually do. Maybe it's best to always identify a file based on its md5
# signature?. Of course that's not possible for some modification events
# because the md5 signature will change...
# - Special move events are required whenever a file is moved
# from a 'watched' directory into another 'watched' directory (or
# subdirectory). In this case we must identify the file by its md5
# signature instead of it's filepath like we usually do. Maybe it's
# best to always identify a file based on its md5 signature?. Of course
# that's not possible for some modification events because the md5
# signature will change...
# Note: Because of the way classes that inherit from pyinotify.ProcessEvent
# interact with constructors. you should only instantiate objects from them
# using keyword arguments. For example:
# OrganizeListener('watch_signal') <= wrong
# OrganizeListener(signal='watch_signal') <= right
class FileMediator(object):
"""
FileMediator is used an intermediate mechanism that filters out certain
events.
"""
ignored_set = set([]) # for paths only
logger = get_logger()
@staticmethod
def is_ignored(path): return path in FileMediator.ignored_set
@staticmethod
def ignore(path): FileMediator.ignored_set.add(path)
@staticmethod
def unignore(path): FileMediator.ignored_set.remove(path)
def mediate_ignored(fn):
def wrapped(self, event, *args,**kwargs):
event.pathname = unicode(event.pathname, "utf-8")
if FileMediator.is_ignored(event.pathname):
FileMediator.logger.info("Ignoring: '%s' (once)" % event.pathname)
FileMediator.unignore(event.pathname)
else: return fn(self, event, *args, **kwargs)
return wrapped
class BaseListener(object):
def my_init(self, signal):
self.signal = signal
class OrganizeListener(BaseListener, pyinotify.ProcessEvent):
class OrganizeListener(BaseListener, pyinotify.ProcessEvent, Loggable):
# this class still don't handle the case where a dir was copied recursively
def process_IN_CLOSE_WRITE(self, event): self.process_to_organize(event)
def process_IN_CLOSE_WRITE(self, event):
self.process_to_organize(event)
# got cookie
def process_IN_MOVED_TO(self, event): self.process_to_organize(event)
def process_IN_MOVED_TO(self, event):
self.process_to_organize(event)
def flush_events(self, path):
"""
organize the whole directory at path. (pretty much by doing what
handle does to every file
"""
flushed = 0
for f in mmp.walk_supported(path, clean_empties=True):
self.logger.info("Bootstrapping: File in 'organize' directory: \
'%s'" % f)
dispatcher.send(signal=self.signal, sender=self,
event=OrganizeFile(f))
flushed += 1
self.logger.info("Flushed organized directory with %d files" % flushed)
@mediate_ignored
@IncludeOnly(mmp.supported_extensions)
def process_to_organize(self, event):
dispatcher.send(signal=self.signal, sender=self, event=OrganizeFile(event))
dispatcher.send(signal=self.signal, sender=self,
event=OrganizeFile(event))
class StoreWatchListener(BaseListener, pyinotify.ProcessEvent):
def process_IN_CLOSE_WRITE(self, event): self.process_create(event)
def process_IN_MOVED_TO(self, event): self.process_create(event)
def process_IN_MOVED_FROM(self, event): self.process_delete(event)
class StoreWatchListener(BaseListener, Loggable, pyinotify.ProcessEvent):
def process_IN_CLOSE_WRITE(self, event):
self.process_create(event)
def process_IN_MOVED_TO(self, event):
if EventRegistry.registered(event):
# We need this trick because we don't how to "expand" dir events
# into file events until we know for sure if we deleted or moved
morph = MoveDir(event) if event.dir else MoveFile(event)
EventRegistry.matching(event).morph_into(morph)
else: self.process_create(event)
def process_IN_MOVED_FROM(self, event):
# Is either delete dir or delete file
evt = self.process_delete(event)
# evt can be none whenever event points that a file that would be
# ignored by @IncludeOnly
if hasattr(event,'cookie') and (evt != None):
EventRegistry.register(evt)
def process_IN_DELETE(self,event): self.process_delete(event)
def process_IN_MOVE_SELF(self, event):
if '-unknown-path' in event.pathname:
event.pathname = event.pathname.replace('-unknown-path','')
self.delete_watch_dir(event)
def delete_watch_dir(self, event):
e = DeleteDirWatch(event)
dispatcher.send(signal='watch_move', sender=self, event=e)
dispatcher.send(signal=self.signal, sender=self, event=e)
@mediate_ignored
@IncludeOnly(mmp.supported_extensions)
def process_create(self, event):
dispatcher.send(signal=self.signal, sender=self, event=NewFile(event))
evt = NewFile(event)
dispatcher.send(signal=self.signal, sender=self, event=evt)
return evt
@mediate_ignored
@IncludeOnly(mmp.supported_extensions)
def process_delete(self, event):
dispatcher.send(signal=self.signal, sender=self, event=DeleteFile(event))
evt = None
if event.dir: evt = DeleteDir(event)
else: evt = DeleteFile(event)
dispatcher.send(signal=self.signal, sender=self, event=evt)
return evt
@mediate_ignored
def process_delete_dir(self, event):
evt = DeleteDir(event)
dispatcher.send(signal=self.signal, sender=self, event=evt)
return evt
def flush_events(self, path):
"""
walk over path and send a NewFile event for every file in this
directory. Not to be confused with bootstrapping which is a more
careful process that involved figuring out what's in the database
first.
"""
# Songs is a dictionary where every key is the watched the directory
# and the value is a set with all the files in that directory.
added = 0
for f in mmp.walk_supported(path, clean_empties=False):
added += 1
dispatcher.send( signal=self.signal, sender=self, event=NewFile(f) )
self.logger.info( "Flushed watch directory. added = %d" % added )

42
python_apps/media-monitor2/media/monitor/log.py
View File

@ -1,15 +1,43 @@
import logging
import abc
import traceback
from media.monitor.pure import LazyProperty
logger = logging.getLogger('mediamonitor2')
logging.basicConfig(filename='/home/rudi/throwaway/mm2.log', level=logging.DEBUG)
appname = 'root'
def setup_logging(log_path):
"""
Setup logging by writing log to 'log_path'
"""
#logger = logging.getLogger(appname)
logging.basicConfig(filename=log_path, level=logging.DEBUG)
def get_logger():
"""
in case we want to use the common logger from a procedural interface
"""
return logging.getLogger()
class Loggable(object):
"""
Any class that wants to log can inherit from this class and automatically
get a logger attribute that can be used like: self.logger.info(...) etc.
"""
__metaclass__ = abc.ABCMeta
# TODO : replace this boilerplate with LazyProperty
@LazyProperty
def logger(self):
# TODO : Clean this up
if not hasattr(self,"_logger"): self._logger = logging.getLogger('mediamonitor2')
return self._logger
def logger(self): return get_logger()
def unexpected_exception(self,e):
"""
Default message for 'unexpected' exceptions
"""
self.fatal_exception("'Unexpected' exception has occured:", e)
def fatal_exception(self, message, e):
"""
Prints an exception 'e' with 'message'. Also outputs the traceback.
"""
self.logger.error( message )
self.logger.error( str(e) )
self.logger.error( traceback.format_exc() )

210
python_apps/media-monitor2/media/monitor/manager.py Normal file
View File

@ -0,0 +1,210 @@
import pyinotify
from pydispatch import dispatcher
from os.path import normpath
from media.monitor.events import PathChannel
from media.monitor.log import Loggable
from media.monitor.listeners import StoreWatchListener, OrganizeListener
from media.monitor.handler import ProblemFileHandler
from media.monitor.organizer import Organizer
import media.monitor.pure as mmp
class Manager(Loggable):
"""
An abstraction over media monitors core pyinotify functions. These
include adding watched,store, organize directories, etc. Basically
composes over WatchManager from pyinotify
"""
global_inst = None
all_signals = set(['add_watch', 'remove_watch'])
def __init__(self):
self.wm = pyinotify.WatchManager()
# These two instance variables are assumed to be constant
self.watch_channel = 'watch'
self.organize_channel = 'organize'
self.watch_listener = StoreWatchListener(signal = self.watch_channel)
self.organize = {
'organize_path' : None,
'imported_path' : None,
'recorded_path' : None,
'problem_files_path' : None,
'organizer' : None,
'problem_handler' : None,
'organize_listener' : OrganizeListener(signal=
self.organize_channel),
}
def dummy(sender, event): self.watch_move( event.path, sender=sender )
dispatcher.connect(dummy, signal='watch_move', sender=dispatcher.Any,
weak=False)
# A private mapping path => watch_descriptor
# we use the same dictionary for organize, watch, store wd events.
# this is a little hacky because we are unable to have multiple wd's
# on the same path.
self.__wd_path = {}
# The following set isn't really necessary anymore. Should be
# removed...
self.watched_directories = set([])
Manager.global_inst = self
# This is the only event that we are unable to process "normally". I.e.
# through dedicated handler objects. Because we must have access to a
# manager instance. Hence we must slightly break encapsulation.
def watch_move(self, watch_dir, sender=None):
"""
handle 'watch move' events directly sent from listener
"""
self.logger.info("Watch dir '%s' has been renamed (hence removed)" %
watch_dir)
self.remove_watch_directory(normpath(watch_dir))
def watch_signal(self): return self.watch_listener.signal
def __remove_watch(self,path):
# only delete if dir is actually being watched
if path in self.__wd_path:
wd = self.__wd_path[path]
self.wm.rm_watch(wd, rec=True)
del(self.__wd_path[path])
def __add_watch(self,path,listener):
wd = self.wm.add_watch(path, pyinotify.ALL_EVENTS, rec=True,
auto_add=True, proc_fun=listener)
self.__wd_path[path] = wd.values()[0]
def __create_organizer(self, target_path, recorded_path):
"""
creates an organizer at new destination path or modifies the old one
"""
# TODO : find a proper fix for the following hack
# We avoid creating new instances of organize because of the way
# it interacts with pydispatch. We must be careful to never have
# more than one instance of OrganizeListener but this is not so
# easy. (The singleton hack in Organizer) doesn't work. This is
# the only thing that seems to work.
if self.organize['organizer']:
o = self.organize['organizer']
o.channel = self.organize_channel
o.target_path = target_path
o.recorded_path = recorded_path
else:
self.organize['organizer'] = Organizer(channel=
self.organize_channel, target_path=target_path,
recorded_path=recorded_path)
def get_problem_files_path(self):
"""
returns the path where problem files should go
"""
return self.organize['problem_files_path']
def set_problem_files_path(self, new_path):
"""
Set the path where problem files should go
"""
self.organize['problem_files_path'] = new_path
self.organize['problem_handler'] = \
ProblemFileHandler( PathChannel(signal='badfile',path=new_path) )
def get_recorded_path(self):
"""
returns the path of the recorded directory
"""
return self.organize['recorded_path']
def set_recorded_path(self, new_path):
self.__remove_watch(self.organize['recorded_path'])
self.organize['recorded_path'] = new_path
self.__create_organizer( self.organize['imported_path'], new_path)
self.__add_watch(new_path, self.watch_listener)
def get_organize_path(self):
"""
returns the current path that is being watched for organization
"""
return self.organize['organize_path']
def set_organize_path(self, new_path):
"""
sets the organize path to be new_path. Under the current scheme there
is only one organize path but there is no reason why more cannot be
supported
"""
# if we are already organizing a particular directory we remove the
# watch from it first before organizing another directory
self.__remove_watch(self.organize['organize_path'])
self.organize['organize_path'] = new_path
# the OrganizeListener instance will walk path and dispatch an organize
# event for every file in that directory
self.organize['organize_listener'].flush_events(new_path)
self.__add_watch(new_path, self.organize['organize_listener'])
def get_imported_path(self):
return self.organize['imported_path']
def set_imported_path(self,new_path):
"""
set the directory where organized files go to.
"""
self.__remove_watch(self.organize['imported_path'])
self.organize['imported_path'] = new_path
self.__create_organizer( new_path, self.organize['recorded_path'])
self.__add_watch(new_path, self.watch_listener)
def change_storage_root(self, store):
"""
hooks up all the directories for you. Problem, recorded, imported,
organize.
"""
store_paths = mmp.expand_storage(store)
self.set_problem_files_path(store_paths['problem_files'])
self.set_imported_path(store_paths['imported'])
self.set_recorded_path(store_paths['recorded'])
self.set_organize_path(store_paths['organize'])
mmp.create_dir(store)
for p in store_paths.values():
mmp.create_dir(p)
def has_watch(self, path):
"""
returns true if the path is being watched or not. Any kind of watch:
organize, store, watched.
"""
return path in self.__wd_path
def add_watch_directory(self, new_dir):
"""
adds a directory to be "watched". "watched" directories are
those that are being monitored by media monitor for airtime in
this context and not directories pyinotify calls watched
"""
if self.has_watch(new_dir):
self.logger.info("Cannot add '%s' to watched directories. It's \
already being watched" % new_dir)
else:
self.logger.info("Adding watched directory: '%s'" % new_dir)
self.__add_watch(new_dir, self.watch_listener)
def remove_watch_directory(self, watch_dir):
"""
removes a directory from being "watched". Undoes add_watch_directory
"""
if self.has_watch(watch_dir):
self.logger.info("Removing watched directory: '%s'", watch_dir)
self.__remove_watch(watch_dir)
else:
self.logger.info("'%s' is not being watched, hence cannot be \
removed" % watch_dir)
def pyinotify(self):
return pyinotify.Notifier(self.wm)
def loop(self):
"""
block until we receive pyinotify events
"""
pyinotify.Notifier(self.wm).loop()
# Experiments with running notifier in different modes
# There are 3 options: normal, async, threaded.
#import asyncore
#pyinotify.AsyncNotifier(self.wm).loop()
#asyncore.loop()

187
python_apps/media-monitor2/media/monitor/metadata.py Normal file
View File

@ -0,0 +1,187 @@
# -*- coding: utf-8 -*-
import mutagen
import math
import os
import copy
from media.monitor.exceptions import BadSongFile
from media.monitor.log import Loggable
import media.monitor.pure as mmp
"""
list of supported easy tags in mutagen version 1.20
['albumartistsort', 'musicbrainz_albumstatus', 'lyricist', 'releasecountry',
'date', 'performer', 'musicbrainz_albumartistid', 'composer', 'encodedby',
'tracknumber', 'musicbrainz_albumid', 'album', 'asin', 'musicbrainz_artistid',
'mood', 'copyright', 'author', 'media', 'length', 'version', 'artistsort',
'titlesort', 'discsubtitle', 'website', 'musicip_fingerprint', 'conductor',
'compilation', 'barcode', 'performer:*', 'composersort', 'musicbrainz_discid',
'musicbrainz_albumtype', 'genre', 'isrc', 'discnumber', 'musicbrainz_trmid',
'replaygain_*_gain', 'musicip_puid', 'artist', 'title', 'bpm',
'musicbrainz_trackid', 'arranger', 'albumsort', 'replaygain_*_peak',
'organization']
"""
airtime2mutagen = {
"MDATA_KEY_TITLE" : "title",
"MDATA_KEY_CREATOR" : "artist",
"MDATA_KEY_SOURCE" : "album",
"MDATA_KEY_GENRE" : "genre",
"MDATA_KEY_MOOD" : "mood",
"MDATA_KEY_TRACKNUMBER" : "tracknumber",
"MDATA_KEY_BPM" : "bpm",
"MDATA_KEY_LABEL" : "organization",
"MDATA_KEY_COMPOSER" : "composer",
"MDATA_KEY_ENCODER" : "encodedby",
"MDATA_KEY_CONDUCTOR" : "conductor",
"MDATA_KEY_YEAR" : "date",
"MDATA_KEY_URL" : "website",
"MDATA_KEY_ISRC" : "isrc",
"MDATA_KEY_COPYRIGHT" : "copyright",
}
# Some airtime attributes are special because they must use the mutagen object
# itself to calculate the value that they need. The lambda associated with each
# key should attempt to extract the corresponding value from the mutagen object
# itself pass as 'm'. In the case when nothing can be extracted the lambda
# should return some default value to be assigned anyway or None so that the
# airtime metadata object will skip the attribute outright.
airtime_special = {
"MDATA_KEY_DURATION" :
lambda m: format_length(getattr(m.info, u'length', 0.0)),
"MDATA_KEY_BITRATE" :
lambda m: getattr(m.info, "bitrate", 0),
"MDATA_KEY_SAMPLERATE" :
lambda m: getattr(m.info, u'sample_rate', 0),
"MDATA_KEY_MIME" :
lambda m: m.mime[0] if len(m.mime) > 0 else u'',
}
mutagen2airtime = dict( (v,k) for k,v in airtime2mutagen.iteritems()
if isinstance(v, str) )
truncate_table = {
'MDATA_KEY_GENRE' : 64,
'MDATA_KEY_TITLE' : 512,
'MDATA_KEY_CREATOR' : 512,
'MDATA_KEY_SOURCE' : 512,
'MDATA_KEY_MOOD' : 64,
'MDATA_KEY_LABEL' : 512,
'MDATA_KEY_COMPOSER' : 512,
'MDATA_KEY_ENCODER' : 255,
'MDATA_KEY_CONDUCTOR' : 512,
'MDATA_KEY_YEAR' : 16,
'MDATA_KEY_URL' : 512,
'MDATA_KEY_ISRC' : 512,
'MDATA_KEY_COPYRIGHT' : 512,
}
def format_length(mutagen_length):
"""
Convert mutagen length to airtime length
"""
t = float(mutagen_length)
h = int(math.floor(t / 3600))
t = t % 3600
m = int(math.floor(t / 60))
s = t % 60
# will be ss.uuu
s = str(s)
seconds = s.split(".")
s = seconds[0]
# have a maximum of 6 subseconds.
if len(seconds[1]) >= 6: ss = seconds[1][0:6]
else: ss = seconds[1][0:]
return "%s:%s:%s.%s" % (h, m, s, ss)
def truncate_to_length(item, length):
if isinstance(item, int): item = str(item)
if isinstance(item, basestring):
if len(item) > length: return item[0:length]
else: return item
class Metadata(Loggable):
# TODO : refactor the way metadata is being handled. Right now things are a
# little bit messy. Some of the handling is in m.m.pure while the rest is
# here. Also interface is not very consistent
@staticmethod
def write_unsafe(path,md):
"""
Writes 'md' metadata into 'path' through mutagen
"""
if not os.path.exists(path):
raise BadSongFile(path)
song_file = mutagen.File(path, easy=True)
for airtime_k, airtime_v in md.iteritems():
if airtime_k in airtime2mutagen:
# The unicode cast here is mostly for integers that need to be
# strings
song_file[ airtime2mutagen[airtime_k] ] = unicode(airtime_v)
song_file.save()
def __init__(self, fpath):
# Forcing the unicode through
try: fpath = fpath.decode("utf-8")
except: pass
try: full_mutagen = mutagen.File(fpath, easy=True)
except Exception: raise BadSongFile(fpath)
self.path = fpath
# TODO : Simplify the way all of these rules are handled right not it's
# extremely unclear and needs to be refactored.
metadata = {}
# Load only the metadata avilable in mutagen into metdata
for k,v in full_mutagen.iteritems():
# Special handling of attributes here
if isinstance(v, list):
# TODO : some files have multiple fields for the same metadata.
# genre is one example. In that case mutagen will return a list
# of values
metadata[k] = v[0]
#if len(v) == 1: metadata[k] = v[0]
#else: raise Exception("Unknown mutagen %s:%s" % (k,str(v)))
else: metadata[k] = v
self.__metadata = {}
# Start populating a dictionary of airtime metadata in __metadata
for muta_k, muta_v in metadata.iteritems():
# We must check if we can actually translate the mutagen key into
# an airtime key before doing the conversion
if muta_k in mutagen2airtime:
airtime_key = mutagen2airtime[muta_k]
# Apply truncation in the case where airtime_key is in our
# truncation table
muta_v = \
truncate_to_length(muta_v, truncate_table[airtime_key])\
if airtime_key in truncate_table else muta_v
self.__metadata[ airtime_key ] = muta_v
# Now we extra the special values that are calculated from the mutagen
# object itself:
for special_key,f in airtime_special.iteritems():
new_val = f(full_mutagen)
if new_val is not None:
self.__metadata[special_key] = f(full_mutagen)
# Finally, we "normalize" all the metadata here:
self.__metadata = mmp.normalized_metadata(self.__metadata, fpath)
# Now we must load the md5:
self.__metadata['MDATA_KEY_MD5'] = mmp.file_md5(fpath,max_length=100)
def is_recorded(self):
"""
returns true if the file has been created by airtime through recording
"""
return mmp.is_airtime_recorded( self.__metadata )
def extract(self):
"""
returns a copy of the metadata that was loaded when object was
constructed
"""
return copy.deepcopy(self.__metadata)
def utf8(self):
"""
Returns a unicode aware representation of the data that is compatible
with what is spent to airtime
"""
return mmp.convert_dict_value_to_utf8(self.extract())

50
python_apps/media-monitor2/media/monitor/organizer.py
View File

@ -1,26 +1,58 @@
# -*- coding: utf-8 -*-
from media.monitor.handler import ReportHandler
import media.monitor.pure as mmp
from media.monitor.log import Loggable
from media.monitor.handler import ReportHandler
from media.monitor.log import Loggable
from media.monitor.exceptions import BadSongFile
class Organizer(ReportHandler,Loggable):
def __init__(self, channel, target_path):
"""
Organizer is responsible to to listening to OrganizeListener events
and committing the appropriate changes to the filesystem. It does
not in any interact with WatchSyncer's even when the the WatchSyncer
is a "storage directory". The "storage" directory picks up all of
its events through pyinotify. (These events are fed to it through
StoreWatchListener)
"""
_instance = None
def __new__(cls, channel, target_path, recorded_path):
if cls._instance:
cls._instance.channel = channel
cls._instance.target_path = target_path
cls._instance.recorded_path = recorded_path
else:
cls._instance = super(Organizer, cls).__new__( cls, channel,
target_path, recorded_path)
return cls._instance
def __init__(self, channel, target_path, recorded_path):
self.channel = channel
self.target_path = target_path
super(Organizer, self).__init__(signal=self.channel.signal)
self.recorded_path = recorded_path
super(Organizer, self).__init__(signal=self.channel, weak=False)
def handle(self, sender, event):
"""Intercept events where a new file has been added to the organize
directory and place it in the correct path (starting with self.target_path)"""
"""
Intercept events where a new file has been added to the organize
directory and place it in the correct path (starting with
self.target_path)
"""
try:
normal_md = mmp.normalized_metadata(event.metadata, event.path)
new_path = mmp.organized_path(event.path, self.target_path, normal_md)
# We must select the target_path based on whether file was recorded
# by airtime or not.
# Do we need to "massage" the path using mmp.organized_path?
target_path = self.recorded_path if event.metadata.is_recorded() \
else self.target_path
new_path = mmp.organized_path(event.path, target_path,
event.metadata.extract())
mmp.magic_move(event.path, new_path)
self.logger.info('Organized: "%s" into "%s"' % (event.path, new_path))
self.logger.info('Organized: "%s" into "%s"' %
(event.path, new_path))
except BadSongFile as e:
self.report_problem_file(event=event, exception=e)
# probably general error in mmp.magic.move...
except Exception as e:
self.unexpected_exception( e )
self.report_problem_file(event=event, exception=e)

294
python_apps/media-monitor2/media/monitor/pure.py
View File

@ -2,7 +2,18 @@
import copy
import os
import shutil
supported_extensions = ["mp3", "ogg"]
import sys
import hashlib
import locale
import operator as op
from os.path import normpath
from itertools import takewhile
from configobj import ConfigObj
from media.monitor.exceptions import FailedToSetLocale, FailedToCreateDir
supported_extensions = [u"mp3", u"ogg", u"oga"]
unicode_unknown = u'unknown'
class LazyProperty(object):
@ -11,7 +22,7 @@ class LazyProperty(object):
property should represent non-mutable data, as it replaces itself.
"""
def __init__(self,fget):
self.fget = fget
self.fget = fget
self.func_name = fget.__name__
def __get__(self,obj,cls):
@ -20,11 +31,12 @@ class LazyProperty(object):
setattr(obj,self.func_name,value)
return value
class IncludeOnly(object):
"""
A little decorator to help listeners only be called on extensions they support
NOTE: this decorator only works on methods and not functions. Maybe fix this?
A little decorator to help listeners only be called on extensions
they support
NOTE: this decorator only works on methods and not functions. Maybe
fix this?
"""
def __init__(self, *deco_args):
self.exts = set([])
@ -35,19 +47,44 @@ class IncludeOnly(object):
def __call__(self, func):
def _wrap(moi, event, *args, **kwargs):
ext = extension(event.pathname)
if ext in self.exts: func(moi, event, *args, **kwargs)
# Checking for emptiness b/c we don't want to skip direcotries
if (ext.lower() in self.exts) or event.dir:
return func(moi, event, *args, **kwargs)
return _wrap
def is_file_supported(path):
return extension(path) in supported_extensions
# In the future we would like a better way to find out
# whether a show has been recorded
def partition(f, alist):
"""
Partition is very similar to filter except that it also returns the
elements for which f return false but in a tuple.
>>> partition(lambda x : x > 3, [1,2,3,4,5,6])
[4,5,6],[1,2,3]
"""
return (filter(f, alist), filter(lambda x: not f(x), alist))
def is_file_supported(path):
"""
Checks if a file's path(filename) extension matches the kind that we
support note that this is case insensitive.
>>> is_file_supported("test.mp3")
True
>>> is_file_supported("/bs/path/test.mP3")
True
>>> is_file_supported("test.txt")
False
"""
return extension(path).lower() in supported_extensions
# TODO : In the future we would like a better way to find out whether a show
# has been recorded
def is_airtime_recorded(md):
return md['MDATA_KEY_CREATOR'] == u'Airtime Show Recorder'
def clean_empty_dirs(path):
""" walks path and deletes every empty directory it finds """
"""
walks path and deletes every empty directory it finds
"""
# TODO : test this function
if path.endswith('/'): clean_empty_dirs(path[0:-1])
else:
for root, dirs, _ in os.walk(path, topdown=False):
@ -58,9 +95,9 @@ def clean_empty_dirs(path):
def extension(path):
"""
return extension of path, empty string otherwise. Prefer
to return empty string instead of None because of bad handling of "maybe"
types in python. I.e. interpreter won't enforce None checks on the programmer
return extension of path, empty string otherwise. Prefer to return empty
string instead of None because of bad handling of "maybe" types in python.
I.e. interpreter won't enforce None checks on the programmer
>>> extension("testing.php")
'php'
>>> extension('/no/extension')
@ -82,69 +119,126 @@ def no_extension_basename(path):
>>> no_extension_basename('blah.ml')
'blah'
"""
base = os.path.basename(path)
base = unicode(os.path.basename(path))
if extension(base) == "": return base
else: return base.split(".")[-2]
def walk_supported(directory, clean_empties=False):
"""
A small generator wrapper around os.walk to only give us files that support the extensions
we are considering. When clean_empties is True we recursively delete empty directories
left over in directory after the walk.
A small generator wrapper around os.walk to only give us files that support
the extensions we are considering. When clean_empties is True we
recursively delete empty directories left over in directory after the walk.
"""
for root, dirs, files in os.walk(directory):
full_paths = ( os.path.join(root, name) for name in files if is_file_supported(name) )
full_paths = ( os.path.join(root, name) for name in files
if is_file_supported(name) )
for fp in full_paths: yield fp
if clean_empties: clean_empty_dirs(directory)
def magic_move(old, new):
# TODO : document this
"""
Moves path old to new and constructs the necessary to directories for new
along the way
"""
new_dir = os.path.dirname(new)
if not os.path.exists(new_dir): os.makedirs(new_dir)
shutil.move(old,new)
def move_to_dir(dir_path,file_path):
"""
moves a file at file_path into dir_path/basename(filename)
"""
bs = os.path.basename(file_path)
magic_move(file_path, os.path.join(dir_path, bs))
def apply_rules_dict(d, rules):
# TODO : document this
"""
Consumes a dictionary of rules that maps some keys to lambdas which it
applies to every matching element in d and returns a new dictionary with
the rules applied
"""
new_d = copy.deepcopy(d)
for k, rule in rules.iteritems():
if k in d: new_d[k] = rule(d[k])
return new_d
def default_to(dictionary, keys, default):
# TODO : document default_to
"""
Checks if the list of keys 'keys' exists in 'dictionary'. If not then it
returns a new dictionary with all those missing keys defaults to 'default'
"""
new_d = copy.deepcopy(dictionary)
for k in keys:
if not (k in new_d): new_d[k] = default
return new_d
def remove_whitespace(dictionary):
"""
Remove values that empty whitespace in the dictionary
"""
nd = copy.deepcopy(dictionary)
bad_keys = []
for k,v in nd.iteritems():
if hasattr(v,'strip'):
stripped = v.strip()
# ghetto and maybe unnecessary
if stripped == '' or stripped == u'':
bad_keys.append(k)
for bad_key in bad_keys: del nd[bad_key]
return nd
def parse_int(s):
"""
Tries very hard to get some sort of integer result from s. Defaults to 0
when it failes
>>> parse_int("123")
123
>>> parse_int("123saf")
123
>>> parse_int("asdf")
0
"""
if s.isdigit(): return s
else:
try:
return reduce(op.add, takewhile(lambda x: x.isdigit(), s))
except: return 0
def normalized_metadata(md, original_path):
""" consumes a dictionary of metadata and returns a new dictionary with the
"""
consumes a dictionary of metadata and returns a new dictionary with the
formatted meta data. We also consume original_path because we must set
MDATA_KEY_CREATOR based on in it sometimes """
MDATA_KEY_CREATOR based on in it sometimes
"""
new_md = copy.deepcopy(md)
# replace all slashes with dashes
for k,v in new_md.iteritems():
new_md[k] = str(v).replace('/','-')
new_md[k] = unicode(v).replace('/','-')
# Specific rules that are applied in a per attribute basis
format_rules = {
# It's very likely that the following isn't strictly necessary. But the old
# code would cast MDATA_KEY_TRACKNUMBER to an integer as a byproduct of
# formatting the track number to 2 digits.
'MDATA_KEY_TRACKNUMBER' : lambda x: int(x),
'MDATA_KEY_BITRATE' : lambda x: str(int(x) / 1000) + "kbps",
# note: you don't actually need the lambda here. It's only used for clarity
'MDATA_KEY_FILEPATH' : lambda x: os.path.normpath(x),
# It's very likely that the following isn't strictly necessary. But the
# old code would cast MDATA_KEY_TRACKNUMBER to an integer as a
# byproduct of formatting the track number to 2 digits.
'MDATA_KEY_TRACKNUMBER' : parse_int,
'MDATA_KEY_BITRATE' : lambda x: str(int(x) / 1000) + "kbps",
'MDATA_KEY_FILEPATH' : lambda x: os.path.normpath(x),
'MDATA_KEY_MIME' : lambda x: x.replace('-','/'),
'MDATA_KEY_BPM' : lambda x: x[0:8],
}
path_md = ['MDATA_KEY_TITLE', 'MDATA_KEY_CREATOR', 'MDATA_KEY_SOURCE',
'MDATA_KEY_TRACKNUMBER', 'MDATA_KEY_BITRATE']
# note that we could have saved a bit of code by rewriting new_md using
# defaultdict(lambda x: "unknown"). But it seems to be too implicit and
# could possibly lead to subtle bugs down the road. Plus the following
# approach gives us the flexibility to use different defaults for
# different attributes
# approach gives us the flexibility to use different defaults for different
# attributes
new_md = apply_rules_dict(new_md, format_rules)
new_md = default_to(dictionary=new_md, keys=['MDATA_KEY_TITLE'], default=no_extension_basename(original_path))
new_md = default_to(dictionary=new_md, keys=path_md, default=unicode_unknown)
new_md = default_to(dictionary=new_md, keys=['MDATA_KEY_TITLE'],
default=no_extension_basename(original_path))
new_md = default_to(dictionary=new_md, keys=path_md,
default=unicode_unknown)
new_md = default_to(dictionary=new_md, keys=['MDATA_KEY_FTYPE'],
default=u'audioclip')
# In the case where the creator is 'Airtime Show Recorder' we would like to
# format the MDATA_KEY_TITLE slightly differently
# Note: I don't know why I'm doing a unicode string comparison here
@ -153,19 +247,20 @@ def normalized_metadata(md, original_path):
hour,minute,second,name = md['MDATA_KEY_TITLE'].split("-",4)
# We assume that MDATA_KEY_YEAR is always given for airtime recorded
# shows
new_md['MDATA_KEY_TITLE'] = '%s-%s-%s:%s:%s' % \
new_md['MDATA_KEY_TITLE'] = u'%s-%s-%s:%s:%s' % \
(name, new_md['MDATA_KEY_YEAR'], hour, minute, second)
# IMPORTANT: in the original code. MDATA_KEY_FILEPATH would also
# be set to the original path of the file for airtime recorded shows
# (before it was "organized"). We will skip this procedure for now
# because it's not clear why it was done
return new_md
return remove_whitespace(new_md)
def organized_path(old_path, root_path, normal_md):
"""
old_path - path where file is store at the moment <= maybe not necessary?
old_path - path where file is store at the moment <= maybe not necessary?
root_path - the parent directory where all organized files go
normal_md - original meta data of the file as given by mutagen AFTER being normalized
normal_md - original meta data of the file as given by mutagen AFTER being
normalized
return value: new file path
"""
filepath = None
@ -173,24 +268,137 @@ def organized_path(old_path, root_path, normal_md):
# The blocks for each if statement look awfully similar. Perhaps there is a
# way to simplify this code
if is_airtime_recorded(normal_md):
fname = u'%s-%s-%s.%s' % ( normal_md['MDATA_KEY_YEAR'], normal_md['MDATA_KEY_TITLE'],
fname = u'%s-%s-%s.%s' % ( normal_md['MDATA_KEY_YEAR'],
normal_md['MDATA_KEY_TITLE'],
normal_md['MDATA_KEY_BITRATE'], ext )
yyyy, mm, _ = normal_md['MDATA_KEY_YEAR'].split('-',3)
path = os.path.join(root_path, yyyy, mm)
filepath = os.path.join(path,fname)
elif normal_md['MDATA_KEY_TRACKNUMBER'] == unicode_unknown:
fname = u'%s-%s.%s' % (normal_md['MDATA_KEY_TITLE'], normal_md['MDATA_KEY_BITRATE'], ext)
fname = u'%s-%s.%s' % (normal_md['MDATA_KEY_TITLE'],
normal_md['MDATA_KEY_BITRATE'], ext)
path = os.path.join(root_path, normal_md['MDATA_KEY_CREATOR'],
normal_md['MDATA_KEY_SOURCE'] )
filepath = os.path.join(path, fname)
else: # The "normal" case
fname = u'%s-%s-%s.%s' % (normal_md['MDATA_KEY_TRACKNUMBER'], normal_md['MDATA_KEY_TITLE'],
fname = u'%s-%s-%s.%s' % (normal_md['MDATA_KEY_TRACKNUMBER'],
normal_md['MDATA_KEY_TITLE'],
normal_md['MDATA_KEY_BITRATE'], ext)
path = os.path.join(root_path, normal_md['MDATA_KEY_CREATOR'],
normal_md['MDATA_KEY_SOURCE'])
filepath = os.path.join(path, fname)
return filepath
def file_md5(path,max_length=100):
"""
Get md5 of file path (if it exists). Use only max_length characters to save
time and memory. Pass max_length=-1 to read the whole file (like in mm1)
"""
if os.path.exists(path):
with open(path, 'rb') as f:
m = hashlib.md5()
# If a file is shorter than "max_length" python will just return
# whatever it was able to read which is acceptable behaviour
m.update(f.read(max_length))
return m.hexdigest()
else: raise ValueError("'%s' must exist to find its md5")
def encode_to(obj, encoding='utf-8'):
# TODO : add documentation + unit tests for this function
if isinstance(obj, unicode): obj = obj.encode(encoding)
return obj
def convert_dict_value_to_utf8(md):
"""
formats a dictionary to send as a request to api client
"""
return dict([(item[0], encode_to(item[1], "utf-8")) for item in md.items()])
def get_system_locale(locale_path='/etc/default/locale'):
"""
Returns the configuration object for the system's default locale. Normally
requires root access.
"""
if os.path.exists(locale_path):
try:
config = ConfigObj(locale_path)
return config
except Exception as e:
raise FailedToSetLocale(locale_path,cause=e)
else: raise ValueError("locale path '%s' does not exist. \
permissions issue?" % locale_path)
def configure_locale(config):
"""
sets the locale according to the system's locale.
"""
current_locale = locale.getlocale()
if current_locale[1] is None:
default_locale = locale.getdefaultlocale()
if default_locale[1] is None:
lang = config.get('LANG')
new_locale = lang
else:
new_locale = default_locale
locale.setlocale(locale.LC_ALL, new_locale)
reload(sys)
sys.setdefaultencoding("UTF-8")
current_locale_encoding = locale.getlocale()[1].lower()
if current_locale_encoding not in ['utf-8', 'utf8']:
raise FailedToSetLocale()
def fondle(path,times=None):
# TODO : write unit tests for this
"""
touch a file to change the last modified date. Beware of calling this
function on the same file from multiple threads.
"""
with file(path, 'a'): os.utime(path, times)
def last_modified(path):
"""
return the time of the last time mm2 was ran. path refers to the index file
whose date modified attribute contains this information. In the case when
the file does not exist we set this time 0 so that any files on the
filesystem were modified after it
"""
if os.path.exists(path): return os.path.getmtime(path)
else: return 0
def expand_storage(store):
"""
A storage directory usually consists of 4 different subdirectories. This
function returns their paths
"""
store = os.path.normpath(store)
return {
'organize' : os.path.join(store, 'organize'),
'recorded' : os.path.join(store, 'recorded'),
'problem_files' : os.path.join(store, 'problem_files'),
'imported' : os.path.join(store, 'imported'),
}
def create_dir(path):
"""
will try and make sure that path exists at all costs. raises an exception
if it fails at this task.
"""
if not os.path.exists(path):
try : os.makedirs(path)
except Exception as e : raise FailedToCreateDir(path, e)
else: # if no error occurs we still need to check that dir exists
if not os.path.exists: raise FailedToCreateDir(path)
def sub_path(directory,f):
"""
returns true if 'f' is in the tree of files under directory.
NOTE: does not look at any symlinks or anything like that, just looks at
the paths.
"""
normalized = normpath(directory)
common = os.path.commonprefix([ directory, normpath(f) ])
return common == normalized
if __name__ == '__main__':
import doctest
doctest.testmod()

120
python_apps/media-monitor2/media/monitor/syncdb.py
View File

@ -1,13 +1,109 @@
# -*- coding: utf-8 -*-
class SyncDB(object):
"""
Represents the database returned by airtime_mvc. We do not use a list or some other
fixed data structure because we might want to change the internal representation for
performance reasons later on.
"""
def __init__(self, source):
self.source = source
def has_file(self, path):
return True
def file_mdata(self, path):
return None
import os
from media.monitor.log import Loggable
from media.monitor.exceptions import NoDirectoryInAirtime
from os.path import normpath
import media.monitor.pure as mmp
class AirtimeDB(Loggable):
def __init__(self, apc, reload_now=True):
self.apc = apc
if reload_now: self.reload_directories()
def reload_directories(self):
"""
this is the 'real' constructor, should be called if you ever want the
class reinitialized. there's not much point to doing it yourself
however, you should just create a new AirtimeDB instance.
"""
# dirs_setup is a dict with keys:
# u'watched_dirs' and u'stor' which point to lists of corresponding
# dirs
dirs_setup = self.apc.setup_media_monitor()
dirs_setup[u'stor'] = normpath( dirs_setup[u'stor'] )
dirs_setup[u'watched_dirs'] = map(normpath, dirs_setup[u'watched_dirs'])
dirs_with_id = dict([ (k,normpath(v)) for k,v in
self.apc.list_all_watched_dirs()['dirs'].iteritems() ])
self.id_to_dir = dirs_with_id
self.dir_to_id = dict([ (v,k) for k,v in dirs_with_id.iteritems() ])
self.base_storage = dirs_setup[u'stor']
self.storage_paths = mmp.expand_storage( self.base_storage )
self.base_id = self.dir_to_id[self.base_storage]
# hack to get around annoying schema of airtime db
self.dir_to_id[ self.recorded_path() ] = self.base_id
self.dir_to_id[ self.import_path() ] = self.base_id
# We don't know from the x_to_y dict which directory is watched or
# store...
self.watched_directories = set([ os.path.normpath(p) for p in
dirs_setup[u'watched_dirs'] ])
def to_id(self, directory):
"""
directory path -> id
"""
return self.dir_to_id[ directory ]
def to_directory(self, dir_id):
"""
id -> directory path
"""
return self.id_to_dir[ dir_id ]
def storage_path(self): return self.base_storage
def organize_path(self): return self.storage_paths['organize']
def problem_path(self): return self.storage_paths['problem_files']
def import_path(self): return self.storage_paths['imported']
def recorded_path(self): return self.storage_paths['recorded']
def list_watched(self):
"""
returns all watched directories as a list
"""
return list(self.watched_directories)
def list_storable_paths(self):
"""
returns a list of all the watched directories in the datatabase.
(Includes the imported directory and the recorded directory)
"""
l = self.list_watched()
l.append(self.import_path())
l.append(self.recorded_path())
return l
def dir_id_get_files(self, dir_id):
"""
Get all files in a directory with id dir_id
"""
base_dir = self.id_to_dir[ dir_id ]
return set(( os.path.join(base_dir,p) for p in
self.apc.list_all_db_files( dir_id ) ))
def directory_get_files(self, directory):
"""
returns all the files(recursively) in a directory. a directory is an
"actual" directory path instead of its id. This is super hacky because
you create one request for the recorded directory and one for the
imported directory even though they're the same dir in the database so
you get files for both dirs in 1 request...
"""
normal_dir = os.path.normpath(unicode(directory))
if normal_dir not in self.dir_to_id:
raise NoDirectoryInAirtime( normal_dir, self.dir_to_id )
all_files = self.dir_id_get_files( self.dir_to_id[normal_dir] )
if normal_dir == self.recorded_path():
all_files = [ p for p in all_files if
mmp.sub_path( self.recorded_path(), p ) ]
elif normal_dir == self.import_path():
all_files = [ p for p in all_files if
mmp.sub_path( self.import_path(), p ) ]
elif normal_dir == self.storage_path():
self.logger.info("Warning, you're getting all files in '%s' which \
includes imported + record" % normal_dir)
return set(all_files)

83
python_apps/media-monitor2/media/monitor/toucher.py Normal file
View File

@ -0,0 +1,83 @@
# -*- coding: utf-8 -*-
import media.monitor.pure as mmp
import os
from media.monitor.log import Loggable
from media.monitor.exceptions import CouldNotCreateIndexFile
class Toucher(Loggable):
"""
Class responsible for touching a file at a certain path when called
"""
def __init__(self,path):
self.path = path
def __call__(self):
try: mmp.fondle(self.path)
except Exception as e:
self.logger.info("Failed to touch file: '%s'. Logging exception." %
self.path)
self.logger.info(str(e))
#http://code.activestate.com/lists/python-ideas/8982/
from datetime import datetime
import threading
class RepeatTimer(threading.Thread):
def __init__(self, interval, callable, args=[], kwargs={}):
threading.Thread.__init__(self)
# interval_current shows number of milliseconds in currently triggered
# <tick>
self.interval_current = interval
# interval_new shows number of milliseconds for next <tick>
self.interval_new = interval
self.callable = callable
self.args = args
self.kwargs = kwargs
self.event = threading.Event()
self.event.set()
self.activation_dt = None
self.__timer = None
def run(self):
while self.event.is_set():
self.activation_dt = datetime.utcnow()
self.__timer = threading.Timer(self.interval_new,
self.callable,
self.args,
self.kwargs)
self.interval_current = self.interval_new
self.__timer.start()
self.__timer.join()
def cancel(self):
self.event.clear()
if self.__timer is not None:
self.__timer.cancel()
def trigger(self):
self.callable(*self.args, **self.kwargs)
if self.__timer is not None:
self.__timer.cancel()
def change_interval(self, value):
self.interval_new = value
class ToucherThread(Loggable):
"""
Creates a thread that touches a file 'path' every 'interval' seconds
"""
def __init__(self, path, interval=5):
if not os.path.exists(path):
try:
# TODO : rewrite using with?
f = open(path,'w')
f.write('')
f.close()
except Exception as e:
raise CouldNotCreateIndexFile(path,e)
cb = Toucher(path)
t = RepeatTimer(interval, cb)
t.daemon = True # thread terminates once process is done
t.start()

189
python_apps/media-monitor2/media/monitor/watchersyncer.py
View File

@ -2,36 +2,80 @@
import threading
import time
import copy
import traceback
from media.monitor.handler import ReportHandler
from media.monitor.events import NewFile, DeleteFile
from media.monitor.log import Loggable
from media.monitor.exceptions import BadSongFile
from media.monitor.pure import LazyProperty
from media.monitor.handler import ReportHandler
from media.monitor.log import Loggable
from media.monitor.exceptions import BadSongFile
from media.monitor.pure import LazyProperty
from media.monitor.eventcontractor import EventContractor
import api_clients.api_client as ac
class RequestSync(threading.Thread,Loggable):
"""
This class is responsible for making the api call to send a request
to airtime. In the process it packs the requests and retries for
some number of times
"""
def __init__(self, watcher, requests):
threading.Thread.__init__(self)
self.watcher = watcher
self.requests = requests
self.retries = 1
self.request_wait = 0.3
@LazyProperty
def apiclient(self):
return ac.AirTimeApiClient()
return ac.AirtimeApiClient.create_right_config()
def run(self):
# TODO : implement proper request sending
self.logger.info("launching request with %d items." % len(self.requests))
# Note that we must attach the appropriate mode to every response. Also
# Not forget to attach the 'is_record' to any requests that are related
# to recorded shows
# A simplistic request would like:
# self.apiclient.send_media_monitor_requests(requests)
self.logger.info("Attempting request with %d items." %
len(self.requests))
# Note that we must attach the appropriate mode to every
# response. Also Not forget to attach the 'is_record' to any
# requests that are related to recorded shows
# TODO : recorded shows aren't flagged right
# Is this retry shit even necessary? Consider getting rid of this.
packed_requests = []
for request_event in self.requests:
try:
for request in request_event.safe_pack():
if isinstance(request, BadSongFile):
self.logger.info("Bad song file: '%s'" % request.path)
else: packed_requests.append(request)
except BadSongFile as e:
self.logger.info("This should never occur anymore!!!")
self.logger.info("Bad song file: '%s'" % e.path)
except Exception as e:
self.logger.info("An evil exception occured")
self.logger.error( traceback.format_exc() )
def make_req():
self.apiclient.send_media_monitor_requests( packed_requests )
for try_index in range(0,self.retries):
try: make_req()
# most likely we did not get json response as we expected
except ValueError:
self.logger.info("ApiController.php probably crashed, we \
diagnose this from the fact that it did not return \
valid json")
self.logger.info("Trying again after %f seconds" %
self.request_wait)
time.sleep( self.request_wait )
except Exception as e: self.unexpected_exception(e)
else:
self.logger.info("Request worked on the '%d' try" %
(try_index + 1))
break
else: self.logger.info("Failed to send request after '%d' tries..." %
self.retries)
self.watcher.flag_done()
class TimeoutWatcher(threading.Thread,Loggable):
"""
The job of this thread is to keep an eye on WatchSyncer and force a
request whenever the requests go over time out
"""
def __init__(self, watcher, timeout=5):
self.logger.info("Created timeout thread...")
threading.Thread.__init__(self)
@ -43,93 +87,126 @@ class TimeoutWatcher(threading.Thread,Loggable):
# so that the people do not have to wait for the queue to fill up
while True:
time.sleep(self.timeout)
# If there is any requests left we launch em.
# Note that this isn't strictly necessary since RequestSync threads
# already chain themselves
# If there is any requests left we launch em. Note that this
# isn't strictly necessary since RequestSync threads already
# chain themselves
if self.watcher.requests_in_queue():
self.logger.info("We got %d requests waiting to be launched" % self.watcher.requests_left_count())
self.logger.info("We got %d requests waiting to be launched" %
self.watcher.requests_left_count())
self.watcher.request_do()
# Same for events, this behaviour is mandatory however.
if self.watcher.events_in_queue():
self.logger.info("We got %d events that are unflushed" % self.watcher.events_left_count())
self.logger.info("We got %d events that are unflushed" %
self.watcher.events_left_count())
self.watcher.flush_events()
class WatchSyncer(ReportHandler,Loggable):
def __init__(self, channel, chunking_number = 50, timeout=15):
self.channel = channel
self.timeout = timeout
self.chunking_number = chunking_number
self.__queue = []
def __init__(self, signal, chunking_number = 100, timeout=15):
self.timeout = float(timeout)
self.chunking_number = int(chunking_number)
self.__reset_queue()
# Even though we are not blocking on the http requests, we are still
# trying to send the http requests in order
self.__requests = []
self.request_running = False
# we don't actually use this "private" instance variable anywhere
self.__current_thread = None
tc = TimeoutWatcher(self, timeout)
self.contractor = EventContractor()
tc = TimeoutWatcher(self, self.timeout)
tc.daemon = True
tc.start()
super(WatchSyncer, self).__init__(signal=channel.signal)
@property
def target_path(self): return self.channel.path
def signal(self): return self.channel.signal
super(WatchSyncer, self).__init__(signal=signal)
def handle(self, sender, event):
"""We implement this abstract method from ReportHandler"""
# Using isinstance like this is usually considered to be bad style
# because you are supposed to use polymorphism instead however we would
# separate event handling itself from the events so there seems to be
# no better way to do this
if isinstance(event, NewFile):
"""
We implement this abstract method from ReportHandler
"""
if hasattr(event, 'pack'):
# We push this event into queue
self.logger.info("Received event '%s'. Path: '%s'" % \
( event.__class__.__name__,
getattr(event,'path','No path exists') ))
try:
self.logger.info("'%s' : New file added: '%s'" % (self.target_path, event.path))
self.push_queue(event)
# If there is a strange bug anywhere in the code the next line
# should be a suspect
if self.contractor.register( event ):
self.push_queue( event )
except BadSongFile as e:
self.report_problem_file(event=event, exception=e)
elif isinstance(event, DeleteFile):
self.logger.info("'%s' : Deleted file: '%s'" % (self.target_path, event.path))
self.push_queue(event)
else: raise Exception("Unknown event: %s" % str(event))
self.fatal_exception("Received bas song file '%s'" % e.path, e)
except Exception as e:
self.unexpected_exception(e)
else:
self.logger.info("Received event that does not implement packing.\
Printing its representation:")
self.logger.info( repr(event) )
def requests_left_count(self): return len(self.__requests)
def events_left_count(self): return len(self.__queue)
def requests_left_count(self):
"""
returns the number of requests left in the queue. requests are
functions that create RequestSync threads
"""
return len(self.__requests)
def events_left_count(self):
"""
Returns the number of events left in the queue to create a request
"""
return len(self.__queue)
def push_queue(self, elem):
"""
Added 'elem' to the event queue and launch a request if we are
over the the chunking number
"""
self.logger.info("Added event into queue")
if self.events_left_count() == self.chunking_number:
if self.events_left_count() >= self.chunking_number:
self.push_request()
self.request_do() # Launch the request if nothing is running
self.__queue.append(elem)
def flush_events(self):
"""
Force flush the current events held in the queue
"""
self.logger.info("Force flushing events...")
self.push_request()
self.request_do()
def events_in_queue(self):
"""returns true if there are events in the queue that haven't been processed yet"""
"""
returns true if there are events in the queue that haven't been
processed yet
"""
return len(self.__queue) > 0
def requests_in_queue(self):
"""
Returns true if there are any requests in the queue. False otherwise.
"""
return len(self.__requests) > 0
def flag_done(self):
""" called by request thread when it finishes operating """
"""
called by request thread when it finishes operating
"""
self.request_running = False
self.__current_thread = None
# This call might not be necessary but we would like
# to get the ball running with the requests as soon as possible
# This call might not be necessary but we would like to get the
# ball running with the requests as soon as possible
if self.requests_in_queue() > 0: self.request_do()
def request_do(self):
""" launches a request thread only if one is not running right now """
"""
launches a request thread only if one is not running right now
"""
if not self.request_running:
self.request_running = True
self.__requests.pop()()
def push_request(self):
self.logger.info("'%s' : Unleashing request" % self.target_path)
"""
Create a request from the current events in the queue and schedule it
"""
self.logger.info("WatchSyncer : Unleashing request")
# want to do request asyncly and empty the queue
requests = copy.copy(self.__queue)
def launch_request():
@ -138,10 +215,14 @@ class WatchSyncer(ReportHandler,Loggable):
t.start()
self.__current_thread = t
self.__requests.append(launch_request)
self.__queue = []
self.__reset_queue()
def __reset_queue(self): self.__queue = []
def __del__(self):
# Ideally we would like to do a little more to ensure safe shutdown
if self.events_in_queue(): self.logger.warn("Terminating with events in the queue still pending...")
if self.requests_in_queue(): self.logger.warn("Terminating with http requests still pending...")
if self.events_in_queue():
self.logger.warn("Terminating with events still in the queue...")
if self.requests_in_queue():
self.logger.warn("Terminating with http requests still pending...")

82
python_apps/media-monitor2/media/update/replaygain.py
View File

@ -4,11 +4,16 @@ import os
import sys
import shutil
import tempfile
import logging
logger = logging.getLogger()
def get_process_output(command):
"""
Run subprocess and return stdout
"""
logger.debug(command)
p = Popen(command, shell=True, stdout=PIPE)
return p.communicate()[0].strip()
@ -22,7 +27,7 @@ def run_process(command):
def get_mime_type(file_path):
"""
Attempts to get the mime type but will return prematurely if the process
takes longer than 5 seconds. Note that this function should only be called
takes longer than 5 seconds. Note that this function should only be called
for files which do not have a mp3/ogg/flac extension.
"""
@ -35,7 +40,7 @@ def duplicate_file(file_path):
fsrc = open(file_path, 'r')
fdst = tempfile.NamedTemporaryFile(delete=False)
print "Copying %s to %s" % (file_path, fdst.name)
logger.info("Copying %s to %s" % (file_path, fdst.name))
shutil.copyfileobj(fsrc, fdst)
@ -44,53 +49,72 @@ def duplicate_file(file_path):
return fdst.name
def get_file_type(file_path):
file_type = None
if re.search(r'mp3$', file_path, re.IGNORECASE):
file_type = 'mp3'
elif re.search(r'og(g|a)$', file_path, re.IGNORECASE):
file_type = 'vorbis'
elif re.search(r'flac$', file_path, re.IGNORECASE):
file_type = 'flac'
else:
mime_type = get_mime_type(file_path) == "audio/mpeg"
if 'mpeg' in mime_type:
file_type = 'mp3'
elif 'ogg' in mime_type:
file_type = 'vorbis'
elif 'flac' in mime_type:
file_type = 'flac'
return file_type
def calculate_replay_gain(file_path):
"""
This function accepts files of type mp3/ogg/flac and returns a calculated ReplayGain value in dB.
If the value cannot be calculated for some reason, then we default to 0 (Unity Gain).
http://wiki.hydrogenaudio.org/index.php?title=ReplayGain_1.0_specification
"""
try:
"""
Making a duplicate is required because the ReplayGain extraction utilities we use
make unwanted modifications to the file.
make unwanted modifications to the file.
"""
search = None
temp_file_path = duplicate_file(file_path)
if re.search(r'mp3$', temp_file_path, re.IGNORECASE) or get_mime_type(temp_file_path) == "audio/mpeg":
if run_process("which mp3gain > /dev/null") == 0:
out = get_process_output('mp3gain -q "%s" 2> /dev/null' % temp_file_path)
search = re.search(r'Recommended "Track" dB change: (.*)', out)
file_type = get_file_type(file_path)
if file_type:
if file_type == 'mp3':
if run_process("which mp3gain > /dev/null") == 0:
out = get_process_output('mp3gain -q "%s" 2> /dev/null' % temp_file_path)
search = re.search(r'Recommended "Track" dB change: (.*)', out)
else:
logger.warn("mp3gain not found")
elif file_type == 'vorbis':
if run_process("which vorbisgain > /dev/null && which ogginfo > /dev/null") == 0:
run_process('vorbisgain -q -f "%s" 2>/dev/null >/dev/null' % temp_file_path)
out = get_process_output('ogginfo "%s"' % temp_file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
logger.warn("vorbisgain/ogginfo not found")
elif file_type == 'flac':
if run_process("which metaflac > /dev/null") == 0:
out = get_process_output('metaflac --show-tag=REPLAYGAIN_TRACK_GAIN "%s"' % temp_file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
logger.warn("metaflac not found")
else:
print "mp3gain not found"
#Log warning
elif re.search(r'ogg$', temp_file_path, re.IGNORECASE) or get_mime_type(temp_file_path) == "application/ogg":
if run_process("which vorbisgain > /dev/null && which ogginfo > /dev/null") == 0:
run_process('vorbisgain -q -f "%s" 2>/dev/null >/dev/null' % temp_file_path)
out = get_process_output('ogginfo "%s"' % temp_file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
print "vorbisgain/ogginfo not found"
#Log warning
elif re.search(r'flac$', temp_file_path, re.IGNORECASE) or get_mime_type(temp_file_path) == "audio/x-flac":
if run_process("which metaflac > /dev/null") == 0:
out = get_process_output('metaflac --show-tag=REPLAYGAIN_TRACK_GAIN "%s"' % temp_file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
print "metaflac not found"
#Log warning
else:
pass
#Log unknown file type.
pass
#no longer need the temp, file simply remove it.
os.remove(temp_file_path)
except Exception, e:
print e
logger.error(str(e))
replay_gain = 0
if search:

77
python_apps/media-monitor2/media/update/replaygainupdater.py
View File

@ -2,67 +2,78 @@ from threading import Thread
import traceback
import os
import logging
import json
import time
from api_clients import api_client
from media.update import replaygain
from media.monitor.log import Loggable
class ReplayGainUpdater(Thread):
class ReplayGainUpdater(Thread, Loggable):
"""
The purpose of the class is to query the server for a list of files which do not have a ReplayGain
value calculated. This class will iterate over the list calculate the values, update the server and
repeat the process until the server reports there are no files left.
The purpose of the class is to query the server for a list of files which
do not have a ReplayGain value calculated. This class will iterate over the
list calculate the values, update the server and repeat the process until
the server reports there are no files left.
This class will see heavy activity right after a 2.1->2.2 upgrade since 2.2 introduces ReplayGain
normalization. A fresh install of Airtime 2.2 will see this class not used at all since a file
imported in 2.2 will automatically have its ReplayGain value calculated.
This class will see heavy activity right after a 2.1->2.2 upgrade since 2.2
introduces ReplayGain normalization. A fresh install of Airtime 2.2 will
see this class not used at all since a file imported in 2.2 will
automatically have its ReplayGain value calculated.
"""
def __init__(self, logger):
@staticmethod
def start_reply_gain(apc):
me = ReplayGainUpdater(apc)
me.daemon = True
me.start()
def __init__(self,apc):
Thread.__init__(self)
self.logger = logger
self.api_client = api_client.AirtimeApiClient()
self.api_client = apc
def main(self):
raw_response = self.api_client.list_all_watched_dirs()
if 'dirs' not in raw_response:
self.logger.error("Could not get a list of watched directories \
with a dirs attribute. Printing full request:")
self.logger.error( raw_response )
return
#TODO make sure object has 'dirs' attr
directories = self.api_client.list_all_watched_dirs()['dirs']
directories = raw_response['dirs']
for dir_id, dir_path in directories.iteritems():
try:
processed_data = []
#keep getting few rows at a time for current music_dir (stor or watched folder).
#When we get a response with 0 rows, then we will set 'finished' to True.
finished = False
while not finished:
# return a list of pairs where the first value is the file's database row id
# and the second value is the filepath
# keep getting few rows at a time for current music_dir (stor
# or watched folder).
total = 0
while True:
# return a list of pairs where the first value is the
# file's database row id and the second value is the
# filepath
files = self.api_client.get_files_without_replay_gain_value(dir_id)
processed_data = []
for f in files:
full_path = os.path.join(dir_path, f['fp'])
processed_data.append((f['id'], replaygain.calculate_replay_gain(full_path)))
self.api_client.update_replay_gain_values(processed_data)
finished = (len(files) == 0)
if len(files) == 0: break
self.logger.info("Processed: %d songs" % total)
except Exception, e:
self.logger.error(e)
self.logger.debug(traceback.format_exc())
def run(self):
try: self.main()
try:
while True:
self.logger.info("Runnning replaygain updater")
self.main()
# Sleep for 5 minutes in case new files have been added
time.sleep(60 * 5)
except Exception, e:
self.logger.error('ReplayGainUpdater Exception: %s', traceback.format_exc())
self.logger.error(e)
if __name__ == "__main__":
try:
rgu = ReplayGainUpdater(logging)
rgu.main()
except Exception, e:
print e
print traceback.format_exc()
rgu = ReplayGainUpdater()
rgu.main()

4
python_apps/media-monitor2/mm1.99.sh Executable file
View File

@ -0,0 +1,4 @@
#export PYTHONPATH="/home/rudi/Airtime/python_apps/:/home/rudi/Airtime/python_apps/media-monitor2/"
PYTHONPATH='/home/rudi/Airtime/python_apps/:/home/rudi/Airtime/python_apps/media-monitor2/'
export PYTHONPATH
python ./mm2.py --config="/home/rudi/Airtime/python_apps/media-monitor2/tests/live_client.cfg" --apiclient="/home/rudi/Airtime/python_apps/media-monitor2/tests/live_client.cfg" --log="/home/rudi/Airtime/python_apps/media-monitor/logging.cfg"

182
python_apps/media-monitor2/mm2.py
View File

@ -1,43 +1,157 @@
# -*- coding: utf-8 -*-
# testing ground for the script
import pyinotify
from media.monitor.listeners import OrganizeListener, StoreWatchListener
from media.monitor.organizer import Organizer
from media.monitor.events import PathChannel
from media.monitor.watchersyncer import WatchSyncer
from media.monitor.handler import ProblemFileHandler
#from media.monitor.bootstrap import Bootstrapper
import sys
import os
import logging
import logging.config
channels = {
# note that org channel still has a 'watch' path because that is the path
# it supposed to be moving the organized files to. it doesn't matter where
# are all the "to organize" files are coming from
'org' : PathChannel('org', '/home/rudi/throwaway/fucking_around/organize'),
'watch' : PathChannel('watch', '/home/rudi/throwaway/fucking_around/watch'),
'badfile' : PathChannel('badfile', '/home/rudi/throwaway/fucking_around/problem_dir'),
}
from media.monitor.manager import Manager
from media.monitor.bootstrap import Bootstrapper
from media.monitor.log import get_logger, setup_logging
from media.monitor.config import MMConfig
from media.monitor.toucher import ToucherThread
from media.monitor.syncdb import AirtimeDB
from media.monitor.exceptions import FailedToObtainLocale, \
FailedToSetLocale, \
NoConfigFile
from media.monitor.airtime import AirtimeNotifier, \
AirtimeMessageReceiver
from media.monitor.watchersyncer import WatchSyncer
from media.monitor.eventdrainer import EventDrainer
from media.update.replaygainupdater import ReplayGainUpdater
from std_err_override import LogWriter
org = Organizer(channel=channels['org'],target_path=channels['watch'].path)
watch = WatchSyncer(channel=channels['watch'])
problem_files = ProblemFileHandler(channel=channels['badfile'])
import media.monitor.pure as mmp
from api_clients import api_client as apc
# do the bootstrapping before any listening is going one
#conn = Connection('localhost', 'more', 'shit', 'here')
#db = DBDumper(conn).dump_block()
#bs = Bootstrapper(db, [channels['org']], [channels['watch']])
#bs.flush_organize()
#bs.flush_watch()
wm = pyinotify.WatchManager()
def main(global_config, api_client_config, log_config,
index_create_attempt=False):
for cfg in [global_config, api_client_config]:
if not os.path.exists(cfg): raise NoConfigFile(cfg)
# MMConfig is a proxy around ConfigObj instances. it does not allow
# itself users of MMConfig instances to modify any config options
# directly through the dictionary. Users of this object muse use the
# correct methods designated for modification
try: config = MMConfig(global_config)
except NoConfigFile as e:
print("Cannot run mediamonitor2 without configuration file.")
print("Current config path: '%s'" % global_config)
sys.exit(1)
except Exception as e:
print("Unknown error reading configuration file: '%s'" % global_config)
print(str(e))
# Listeners don't care about which directory they're related to. All they care
# about is which signal they should respond to
o1 = OrganizeListener(signal=channels['org'].signal)
o2 = StoreWatchListener(signal=channels['watch'].signal)
notifier = pyinotify.Notifier(wm)
wdd1 = wm.add_watch(channels['org'].path, pyinotify.ALL_EVENTS, rec=True, auto_add=True, proc_fun=o1)
wdd2 = wm.add_watch(channels['watch'].path, pyinotify.ALL_EVENTS, rec=True, auto_add=True, proc_fun=o2)
logging.config.fileConfig(log_config)
notifier.loop()
#need to wait for Python 2.7 for this..
#logging.captureWarnings(True)
logger = logging.getLogger()
LogWriter.override_std_err(logger)
logfile = unicode( config['logpath'] )
setup_logging(logfile)
log = get_logger()
if not index_create_attempt:
if not os.path.exists(config['index_path']):
log.info("Attempting to create index file:...")
try:
f = open(config['index_path'])
f.write(" ")
f.close()
except Exception as e:
log.info("Failed to create index file with exception: %s" % str(e))
else:
log.info("Created index file, reloading configuration:")
main( global_config, api_client_config, log_config,
index_create_attempt=True )
else:
log.info("Already tried to create index. Will not try again ")
if not os.path.exists(config['index_path']):
log.info("Index file does not exist. Terminating")
log.info("Attempting to set the locale...")
try:
mmp.configure_locale(mmp.get_system_locale())
except FailedToSetLocale as e:
log.info("Failed to set the locale...")
sys.exit(1)
except FailedToObtainLocale as e:
log.info("Failed to obtain the locale form the default path: \
'/etc/default/locale'")
sys.exit(1)
except Exception as e:
log.info("Failed to set the locale for unknown reason. \
Logging exception.")
log.info(str(e))
watch_syncer = WatchSyncer(signal='watch',
chunking_number=config['chunking_number'],
timeout=config['request_max_wait'])
apiclient = apc.AirtimeApiClient.create_right_config(log=log,
config_path=api_client_config)
ReplayGainUpdater.start_reply_gain(apiclient)
sdb = AirtimeDB(apiclient)
manager = Manager()
airtime_receiver = AirtimeMessageReceiver(config,manager)
airtime_notifier = AirtimeNotifier(config, airtime_receiver)
store = apiclient.setup_media_monitor()
airtime_receiver.change_storage({ 'directory':store[u'stor'] })
for watch_dir in store[u'watched_dirs']:
if not os.path.exists(watch_dir):
# Create the watch_directory here
try: os.makedirs(watch_dir)
except Exception as e:
log.error("Could not create watch directory: '%s' \
(given from the database)." % watch_dir)
if os.path.exists(watch_dir):
airtime_receiver.new_watch({ 'directory':watch_dir })
bs = Bootstrapper( db=sdb, watch_signal='watch' )
ed = EventDrainer(airtime_notifier.connection,
interval=float(config['rmq_event_wait']))
# Launch the toucher that updates the last time when the script was
# ran every n seconds.
tt = ToucherThread(path=config['index_path'],
interval=int(config['touch_interval']))
pyi = manager.pyinotify()
pyi.loop()
__doc__ = """
Usage:
mm2.py --config=<path> --apiclient=<path> --log=<path>
Options:
-h --help Show this screen
--config=<path> path to mm2 config
--apiclient=<path> path to apiclient config
--log=<path> log config at <path>
"""
#original debugging paths
#base_path = u'/home/rudi/Airtime/python_apps/media-monitor2/tests'
#global_config = os.path.join(base_path, u'live_client.cfg')
#api_client_config = global_config
if __name__ == '__main__':
from docopt import docopt
args = docopt(__doc__,version="mm1.99")
for k in ['--apiclient','--config','--log']:
if not os.path.exists(args[k]):
print("'%s' must exist" % args[k])
sys.exit(0)
print("Running mm1.99")
main(args['--config'],args['--apiclient'],args['--log'])

30
python_apps/media-monitor2/pyitest.py Executable file
View File

@ -0,0 +1,30 @@
#!/usr/bin/python
import sys
import os
import getopt
import pyinotify
import pprint
# a little shit script to test out pyinotify events
class AT(pyinotify.ProcessEvent):
def process_default(self, event):
pprint.pprint(event)
def main():
optlist, arguments = getopt.getopt(sys.argv[1:], '', ["dir="])
ldir = ""
for k,v in optlist:
if k == '--dir':
ldir = v
break
if not os.path.exists(ldir):
print("can't pyinotify dir: '%s'. it don't exist" % ldir)
sys.exit(0)
wm = pyinotify.WatchManager()
notifier = pyinotify.Notifier(wm)
print("Watching: '%s'" % ldir)
wm.add_watch(ldir, pyinotify.ALL_EVENTS, auto_add=True, rec=True, proc_fun=AT())
notifier.loop()
if __name__ == '__main__': main()

138
python_apps/media-monitor2/tests/live_client.cfg Normal file
View File

@ -0,0 +1,138 @@
bin_dir = "/usr/lib/airtime/api_clients"
############################################
# RabbitMQ settings #
############################################
rabbitmq_host = 'localhost'
rabbitmq_user = 'guest'
rabbitmq_password = 'guest'
rabbitmq_vhost = '/'
############################################
# Media-Monitor preferences #
############################################
check_filesystem_events = 5 #how long to queue up events performed on the files themselves.
check_airtime_events = 30 #how long to queue metadata input from airtime.
touch_interval = 5
chunking_number = 450
request_max_wait = 3.0
rmq_event_wait = 0.5
logpath = '/home/rudi/throwaway/mm2.log'
#############################
## Common
#############################
index_path = '/home/rudi/Airtime/python_apps/media-monitor2/sample_post.txt'
# Value needed to access the API
api_key = '5LF5D953RNS3KJSHN6FF'
# Path to the base of the API
api_base = 'api'
# URL to get the version number of the server API
version_url = 'version/api_key/%%api_key%%'
#URL to register a components IP Address with the central web server
register_component = 'register-component/format/json/api_key/%%api_key%%/component/%%component%%'
# Hostname
base_url = 'localhost'
base_port = 80
#############################
## Config for Media Monitor
#############################
# URL to setup the media monitor
media_setup_url = 'media-monitor-setup/format/json/api_key/%%api_key%%'
# Tell Airtime the file id associated with a show instance.
upload_recorded = 'upload-recorded/format/json/api_key/%%api_key%%/fileid/%%fileid%%/showinstanceid/%%showinstanceid%%'
# URL to tell Airtime to update file's meta data
update_media_url = 'reload-metadata/format/json/api_key/%%api_key%%/mode/%%mode%%'
# URL to tell Airtime we want a listing of all files it knows about
list_all_db_files = 'list-all-files/format/json/api_key/%%api_key%%/dir_id/%%dir_id%%'
# URL to tell Airtime we want a listing of all dirs its watching (including the stor dir)
list_all_watched_dirs = 'list-all-watched-dirs/format/json/api_key/%%api_key%%'
# URL to tell Airtime we want to add watched directory
add_watched_dir = 'add-watched-dir/format/json/api_key/%%api_key%%/path/%%path%%'
# URL to tell Airtime we want to add watched directory
remove_watched_dir = 'remove-watched-dir/format/json/api_key/%%api_key%%/path/%%path%%'
# URL to tell Airtime we want to add watched directory
set_storage_dir = 'set-storage-dir/format/json/api_key/%%api_key%%/path/%%path%%'
# URL to tell Airtime about file system mount change
update_fs_mount = 'update-file-system-mount/format/json/api_key/%%api_key%%'
# URL to tell Airtime about file system mount change
handle_watched_dir_missing = 'handle-watched-dir-missing/format/json/api_key/%%api_key%%/dir/%%dir%%'
#############################
## Config for Recorder
#############################
# URL to get the schedule of shows set to record
show_schedule_url = 'recorded-shows/format/json/api_key/%%api_key%%'
# URL to upload the recorded show's file to Airtime
upload_file_url = 'upload-file/format/json/api_key/%%api_key%%'
# URL to commit multiple updates from media monitor at the same time
reload_metadata_group = 'reload-metadata-group/format/json/api_key/%%api_key%%'
#number of retries to upload file if connection problem
upload_retries = 3
#time to wait between attempts to upload file if connection problem (in seconds)
upload_wait = 60
################################################################################
# Uncomment *one of the sets* of values from the API clients below, and comment
# out all the others.
################################################################################
#############################
## Config for Pypo
#############################
# Schedule export path.
# %%from%% - starting date/time in the form YYYY-MM-DD-hh-mm
# %%to%% - starting date/time in the form YYYY-MM-DD-hh-mm
export_url = 'schedule/api_key/%%api_key%%'
get_media_url = 'get-media/file/%%file%%/api_key/%%api_key%%'
# Update whether a schedule group has begun playing.
update_item_url = 'notify-schedule-group-play/api_key/%%api_key%%/schedule_id/%%schedule_id%%'
# Update whether an audio clip is currently playing.
update_start_playing_url = 'notify-media-item-start-play/api_key/%%api_key%%/media_id/%%media_id%%/schedule_id/%%schedule_id%%'
# URL to tell Airtime we want to get stream setting
get_stream_setting = 'get-stream-setting/format/json/api_key/%%api_key%%/'
#URL to update liquidsoap status
update_liquidsoap_status = 'update-liquidsoap-status/format/json/api_key/%%api_key%%/msg/%%msg%%/stream_id/%%stream_id%%/boot_time/%%boot_time%%'
#URL to check live stream auth
check_live_stream_auth = 'check-live-stream-auth/format/json/api_key/%%api_key%%/username/%%username%%/password/%%password%%/djtype/%%djtype%%'
#URL to update source status
update_source_status = 'update-source-status/format/json/api_key/%%api_key%%/sourcename/%%sourcename%%/status/%%status%%'
get_bootstrap_info = 'get-bootstrap-info/format/json/api_key/%%api_key%%'
get_files_without_replay_gain = 'get-files-without-replay-gain/api_key/%%api_key%%/dir_id/%%dir_id%%'
update_replay_gain_value = 'update-replay-gain-value/api_key/%%api_key%%'

19
python_apps/media-monitor2/tests/prepare_tests.py Normal file
View File

@ -0,0 +1,19 @@
import shutil
import subprocess
# The tests rely on a lot of absolute paths and other garbage so this file
# configures all of that
music_folder = u'/home/rudi/music'
o_path = u'/home/rudi/throwaway/ACDC_-_Back_In_Black-sample-64kbps.ogg'
watch_path = u'/home/rudi/throwaway/fucking_around/watch/',
real_path1 = u'/home/rudi/throwaway/fucking_around/watch/unknown/unknown/ACDC_-_Back_In_Black-sample-64kbps-64kbps.ogg'
opath = u"/home/rudi/Airtime/python_apps/media-monitor2/tests/"
ppath = u"/home/rudi/Airtime/python_apps/media-monitor2/media/"
sample_config = u'/home/rudi/Airtime/python_apps/media-monitor2/tests/api_client.cfg'
real_config = u'/home/rudi/Airtime/python_apps/media-monitor2/tests/live_client.cfg'
api_client_path = '/etc/airtime/api_client.cfg'
if __name__ == "__main__":
shutil.copy(api_client_path, real_config)
# TODO : fix this to use liberal permissions
subprocess.call(["chown","rudi",real_config])

7
python_apps/media-monitor2/tests/run_tests.pl Executable file
View File

@ -0,0 +1,7 @@
#!/usr/bin/perl
use strict;
use warnings;
foreach my $file (glob "*.py") {
system("python $file") unless $file =~ /prepare_tests.py/;
}

5
python_apps/media-monitor2/tests/api_client.py → python_apps/media-monitor2/tests/test_api_client.py
View File

@ -4,9 +4,12 @@ import os
import sys
from api_clients import api_client as apc
import prepare_tests
class TestApiClient(unittest.TestCase):
def setUp(self):
test_path = '/home/rudi/Airtime/python_apps/media-monitor2/tests/api_client.cfg'
test_path = prepare_tests.real_config
if not os.path.exists(test_path):
print("path for config does not exist: '%s' % test_path")
# TODO : is there a cleaner way to exit the unit testing?

0
python_apps/media-monitor2/tests/config.py → python_apps/media-monitor2/tests/test_config.py
View File

41
python_apps/media-monitor2/tests/test_manager.py Normal file
View File

@ -0,0 +1,41 @@
import unittest
from media.monitor.manager import Manager
def add_paths(m,paths):
for path in paths:
m.add_watch_directory(path)
class TestManager(unittest.TestCase):
def setUp(self):
self.opath = "/home/rudi/Airtime/python_apps/media-monitor2/tests/"
self.ppath = "/home/rudi/Airtime/python_apps/media-monitor2/media/"
self.paths = [self.opath, self.ppath]
def test_init(self):
man = Manager()
self.assertTrue( len(man.watched_directories) == 0 )
self.assertTrue( man.watch_channel is not None )
self.assertTrue( man.organize_channel is not None )
def test_organize_path(self):
man = Manager()
man.set_organize_path( self.opath )
self.assertEqual( man.get_organize_path(), self.opath )
man.set_organize_path( self.ppath )
self.assertEqual( man.get_organize_path(), self.ppath )
def test_add_watch_directory(self):
man = Manager()
add_paths(man, self.paths)
for path in self.paths:
self.assertTrue( man.has_watch(path) )
def test_remove_watch_directory(self):
man = Manager()
add_paths(man, self.paths)
for path in self.paths:
self.assertTrue( man.has_watch(path) )
man.remove_watch_directory( path )
self.assertTrue( not man.has_watch(path) )
if __name__ == '__main__': unittest.main()

51
python_apps/media-monitor2/tests/test_metadata.py Normal file
View File

@ -0,0 +1,51 @@
# -*- coding: utf-8 -*-
import os
import unittest
import sys
import media.monitor.metadata as mmm
class TestMetadata(unittest.TestCase):
def setUp(self):
self.music_folder = u'/home/rudi/music'
def test_got_music_folder(self):
t = os.path.exists(self.music_folder)
self.assertTrue(t)
if not t:
print("'%s' must exist for this test to run." % self.music_folder )
sys.exit(1)
def test_metadata(self):
full_paths = (os.path.join(self.music_folder,filename) for filename in os.listdir(self.music_folder))
i = 0
for full_path in full_paths:
if os.path.isfile(full_path):
md_full = mmm.Metadata(full_path)
md = md_full.extract()
if i < 3:
i += 1
print("Sample metadata: '%s'" % md)
self.assertTrue( len( md.keys() ) > 0 )
self.assertTrue( 'MDATA_KEY_MD5' in md )
utf8 = md_full.utf8()
for k,v in md.iteritems():
if hasattr(utf8[k], 'decode'):
self.assertEqual( utf8[k].decode('utf-8'), md[k] )
else: print("Skipping '%s' because it's a directory" % full_path)
def test_airtime_mutagen_dict(self):
for muta,airtime in mmm.mutagen2airtime.iteritems():
self.assertEqual( mmm.airtime2mutagen[airtime], muta )
def test_format_length(self):
# TODO : add some real tests for this function
x1 = 123456
print("Formatting '%s' to '%s'" % (x1, mmm.format_length(x1)))
def test_truncate_to_length(self):
s1 = "testing with non string literal"
s2 = u"testing with unicode literal"
self.assertEqual( len(mmm.truncate_to_length(s1, 5)), 5)
self.assertEqual( len(mmm.truncate_to_length(s2, 8)), 8)
if __name__ == '__main__': unittest.main()

15
python_apps/media-monitor2/tests/notifier.py → python_apps/media-monitor2/tests/test_notifier.py
View File

@ -6,23 +6,24 @@ from media.monitor.airtime import AirtimeNotifier, AirtimeMessageReceiver
from mock import patch, Mock
from media.monitor.config import MMConfig
from media.monitor.manager import Manager
def filter_ev(d): return { i : j for i,j in d.iteritems() if i != 'event_type' }
class TestReceiver(unittest.TestCase):
def setUp(self):
# TODO : properly mock this later
cfg = {}
self.amr = AirtimeMessageReceiver(cfg)
def test_supported_messages(self):
self.assertTrue( len(self.amr.supported_messages()) > 0 )
self.amr = AirtimeMessageReceiver(cfg, Manager())
def test_supported(self):
# Every supported message should fire something
for event_type in self.amr.supported_messages():
for event_type in self.amr.dispatch_tables.keys():
msg = { 'event_type' : event_type, 'extra_param' : 123 }
filtered = filter_ev(msg)
with patch.object(self.amr, 'execute_message') as mock_method:
# There should be a better way to test the following without
# patching private methods
with patch.object(self.amr, '_execute_message') as mock_method:
mock_method.side_effect = None
ret = self.amr.message(msg)
self.assertTrue(ret)
@ -31,7 +32,7 @@ class TestReceiver(unittest.TestCase):
def test_no_mod_message(self):
ev = { 'event_type' : 'new_watch', 'directory' : 'something here' }
filtered = filter_ev(ev)
with patch.object(self.amr, 'execute_message') as mock_method:
with patch.object(self.amr, '_execute_message') as mock_method:
mock_method.return_value = "tested"
ret = self.amr.message(ev)
self.assertTrue( ret ) # message passing worked

9
python_apps/media-monitor2/tests/pure.py → python_apps/media-monitor2/tests/test_pure.py
View File

@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
import unittest
import os
import media.monitor.pure as mmp
class TestMMP(unittest.TestCase):
@ -51,4 +52,12 @@ class TestMMP(unittest.TestCase):
# for recorded it should be something like this
# ./recorded/2012/07/2012-07-09-17-55-00-Untitled Show-256kbps.ogg
def test_file_md5(self):
p = os.path.realpath(__file__)
m1 = mmp.file_md5(p)
m2 = mmp.file_md5(p,10)
self.assertTrue( m1 != m2 )
self.assertRaises( ValueError, lambda : mmp.file_md5('/bull/shit/path') )
self.assertTrue( m1 == mmp.file_md5(p) )
if __name__ == '__main__': unittest.main()

35
python_apps/media-monitor2/tests/test_syncdb.py Normal file
View File

@ -0,0 +1,35 @@
# -*- coding: utf-8 -*-
import unittest
import os
from media.monitor.syncdb import AirtimeDB
from media.monitor.log import get_logger
from media.monitor.pure import partition
import api_clients.api_client as ac
import prepare_tests
class TestAirtimeDB(unittest.TestCase):
def setUp(self):
self.ac = ac.AirtimeApiClient(logger=get_logger(),
config_path=prepare_tests.real_config)
def test_syncdb_init(self):
sdb = AirtimeDB(self.ac)
self.assertTrue( len(sdb.list_storable_paths()) > 0 )
def test_list(self):
self.sdb = AirtimeDB(self.ac)
for watch_dir in self.sdb.list_storable_paths():
self.assertTrue( os.path.exists(watch_dir) )
def test_directory_get_files(self):
sdb = AirtimeDB(self.ac)
print(sdb.list_storable_paths())
for wdir in sdb.list_storable_paths():
files = sdb.directory_get_files(wdir)
print( "total files: %d" % len(files) )
self.assertTrue( len(files) >= 0 )
self.assertTrue( isinstance(files, set) )
exist, deleted = partition(os.path.exists, files)
print("(exist, deleted) = (%d, %d)" % ( len(exist), len(deleted) ) )
if __name__ == '__main__': unittest.main()

54
python_apps/media-monitor2/tests/test_toucher.py Normal file
View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
import unittest
import time
import media.monitor.pure as mmp
from media.monitor.toucher import Toucher, ToucherThread
class BaseTest(unittest.TestCase):
def setUp(self):
self.p = "api_client.cfg"
class TestToucher(BaseTest):
def test_toucher(self):
t1 = mmp.last_modified(self.p)
t = Toucher(self.p)
t()
t2 = mmp.last_modified(self.p)
print("(t1,t2) = (%d, %d) diff => %d" % (t1, t2, t2 - t1))
self.assertTrue( t2 > t1 )
class TestToucherThread(BaseTest):
def test_thread(self):
t1 = mmp.last_modified(self.p)
ToucherThread(self.p, interval=1)
time.sleep(2)
t2 = mmp.last_modified(self.p)
print("(t1,t2) = (%d, %d) diff => %d" % (t1, t2, t2 - t1))
self.assertTrue( t2 > t1 )
if __name__ == '__main__': unittest.main()

BIN
python_apps/python-virtualenv/airtime_virtual_env.pybundle
View File

Binary file not shown.