airtime_nginx_php7/python_apps/media-monitor/mm2/media/metadata/process.py

# -*- coding: utf-8 -*-
from contextlib             import contextmanager
from ..monitor.pure         import truncate_to_value, truncate_to_length, toposort
from os.path                import normpath
from ..monitor.exceptions   import BadSongFile
from ..monitor.log          import Loggable
from ..monitor              import pure as mmp
from collections            import namedtuple
import mutagen
import subprocess
import json
import logging

class FakeMutagen(dict):
    """
    Need this fake mutagen object so that airtime_special functions
    return a proper default value instead of throwing an exceptions for
    files that mutagen doesn't recognize
    """
    FakeInfo = namedtuple('FakeInfo','length bitrate')
    def __init__(self,path):
        self.path = path
        self.mime = ['audio/wav']
        self.info = FakeMutagen.FakeInfo(0.0, '')
        dict.__init__(self)
    def set_length(self,l):
        old_bitrate = self.info.bitrate
        self.info = FakeMutagen.FakeInfo(l, old_bitrate)


class MetadataAbsent(Exception):
    def __init__(self, name): self.name = name
    def __str__(self): return "Could not obtain element '%s'" % self.name

class MetadataElement(Loggable):

    def __init__(self,name):
        self.name = name
        # "Sane" defaults
        self.__deps          = set()
        self.__normalizer    = lambda x: x
        self.__optional      = True
        self.__default       = None
        self.__is_normalized = lambda _ : True
        self.__max_length    = -1
        self.__max_value     = -1
        self.__translator    = None

    def max_length(self,l):
        self.__max_length = l

    def max_value(self,v):
        self.__max_value = v

    def optional(self, setting):
        self.__optional = setting

    def is_optional(self):
        return self.__optional

    def depends(self, *deps):
        self.__deps = set(deps)

    def dependencies(self):
        return self.__deps

    def translate(self, f):
        self.__translator = f

    def is_normalized(self, f):
        self.__is_normalized = f

    def normalize(self, f):
        self.__normalizer = f

    def default(self,v):
        self.__default = v

    def get_default(self):
        if hasattr(self.__default, '__call__'): return self.__default()
        else: return self.__default

    def has_default(self):
        return self.__default is not None

    def path(self):
        return self.__path

    def __slice_deps(self, d):
        """
        returns a dictionary of all the key value pairs in d that are also
        present in self.__deps
        """
        return dict( (k,v) for k,v in d.iteritems() if k in self.__deps)

    def __str__(self):
        return "%s(%s)" % (self.name, ' '.join(list(self.__deps)))

    def read_value(self, path, original, running={}):

        # If value is present and normalized then we only check if it's
        # normalized or not. We normalize if it's not normalized already

        if self.name in original:
            v = original[self.name]
            if self.__is_normalized(v): return v
            else: return self.__normalizer(v)

        # We slice out only the dependencies that are required for the metadata
        # element.
        dep_slice_orig    = self.__slice_deps(original)
        dep_slice_running = self.__slice_deps(running)
        # TODO : remove this later
        dep_slice_special = self.__slice_deps({'path' : path})
        # We combine all required dependencies into a single dictionary
        # that we will pass to the translator
        full_deps         = dict( dep_slice_orig.items()
                                + dep_slice_running.items()
                                + dep_slice_special.items())

        # check if any dependencies are absent
        # note: there is no point checking the case that len(full_deps) >
        # len(self.__deps) because we make sure to "slice out" any supefluous
        # dependencies above.
        if len(full_deps) != len(self.dependencies()) or \
            len(self.dependencies()) == 0:
            # If we have a default value then use that. Otherwise throw an
            # exception
            if self.has_default(): return self.get_default()
            else: raise MetadataAbsent(self.name)

        # We have all dependencies. Now for actual for parsing
        def def_translate(dep):
            def wrap(k):
                e = [ x for x in dep ][0]
                return k[e]
            return wrap

        # Only case where we can select a default translator
        if self.__translator is None:
            self.translate(def_translate(self.dependencies()))
            if len(self.dependencies()) > 2: # dependencies include themselves
                self.logger.info("Ignoring some dependencies in translate %s"
                                 % self.name)
                self.logger.info(self.dependencies())

        r = self.__normalizer( self.__translator(full_deps) )
        if self.__max_length != -1:
            r = truncate_to_length(r, self.__max_length)
        if self.__max_value != -1:
            try: r = truncate_to_value(r, self.__max_value)
            except ValueError, e: r = ''
        return r

def normalize_mutagen(path):
    """
    Consumes a path and reads the metadata using mutagen. normalizes some of
    the metadata that isn't read through the mutagen hash
    """
    if not mmp.file_playable(path): raise BadSongFile(path)
    try              : m = mutagen.File(path, easy=True)
    except Exception : raise BadSongFile(path)
    if m is None: m = FakeMutagen(path)
    try:
        if mmp.extension(path) == 'wav':
            m.set_length(mmp.read_wave_duration(path))
    except Exception: raise BadSongFile(path)
    md = {}
    for k,v in m.iteritems():
        if type(v) is list:
            if len(v) > 0: md[k] = v[0]
        else: md[k] = v
    # populate special metadata values
    md['length']      = getattr(m.info, 'length', 0.0)
    md['bitrate']     = getattr(m.info, 'bitrate', u'')
    md['sample_rate'] = getattr(m.info, 'sample_rate', 0)
    md['mime']        = m.mime[0] if len(m.mime) > 0 else u''
    md['path']        = normpath(path)

    # silence detect(set default cue in and out)
    #try:
        #command = ['silan', '-b', '-f', 'JSON', md['path']]
        #proc = subprocess.Popen(command, stdout=subprocess.PIPE)
        #out = proc.communicate()[0].strip('\r\n')

        #info = json.loads(out)
        #md['cuein'] = info['sound'][0][0]
        #md['cueout'] = info['sound'][0][1]
    #except Exception:
        #self.logger.debug('silan is missing')

    if 'title' not in md: md['title']  = u''
    return md


class OverwriteMetadataElement(Exception):
    def __init__(self, m): self.m = m
    def __str__(self): return "Trying to overwrite: %s" % self.m

class MetadataReader(object):
    def __init__(self):
        self.clear()

    def register_metadata(self,m):
        if m in self.__mdata_name_map:
            raise OverwriteMetadataElement(m)
        self.__mdata_name_map[m.name] = m
        d = dict( (name,m.dependencies()) for name,m in
                self.__mdata_name_map.iteritems() )
        new_list = list( toposort(d) )
        self.__metadata = [ self.__mdata_name_map[name] for name in new_list
                if name in self.__mdata_name_map]

    def clear(self):
        self.__mdata_name_map = {}
        self.__metadata       = []

    def read(self, path, muta_hash):
        normalized_metadata = {}
        for mdata in self.__metadata:
            try:
                normalized_metadata[mdata.name] = mdata.read_value(
                        path, muta_hash, normalized_metadata)
            except MetadataAbsent:
                if not mdata.is_optional(): raise
        return normalized_metadata

    def read_mutagen(self, path):
        return self.read(path, normalize_mutagen(path))

global_reader = MetadataReader()

@contextmanager
def metadata(name):
    t = MetadataElement(name)
    yield t
    global_reader.register_metadata(t)