###----- Licensing notice ---------------------------------------------------
###
-### This program is free software; you can redistribute it and/or modify
+### This file is part of the `autoys' audio tools collection.
+###
+### `autoys' is free software; you can redistribute it and/or modify
### it under the terms of the GNU General Public License as published by
### the Free Software Foundation; either version 2 of the License, or
### (at your option) any later version.
###
-### This program is distributed in the hope that it will be useful,
+### `autoys' is distributed in the hope that it will be useful,
### but WITHOUT ANY WARRANTY; without even the implied warranty of
### MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
### GNU General Public License for more details.
###
### You should have received a copy of the GNU General Public License
-### along with this program; if not, write to the Free Software Foundation,
+### along with `autoys'; if not, write to the Free Software Foundation,
### Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
###--------------------------------------------------------------------------
import optparse as OP
import threading as TH
import shlex as L
-from math import sqrt
+from math import sqrt, ceil
from contextlib import contextmanager
## eyeD3 tag fettling.
-import eyeD3 as E3
+import eyed3 as E3
## Gstreamer. It picks up command-line arguments -- most notably `--help' --
## and processes them itself. Of course, its help is completely wrong. This
else: w += 1
## Done.
- #print ';; %r -> %d' % (s, w)
return w
class StatusLine (object):
## Eyecandy update.
if me.eyecandyp:
- #print
- #print ';; new status %r' % line
## If the old line was longer, we need to clobber its tail, so work out
## what that involves.
## Actually do the output, all in one syscall.
b = charwidth(me._last[i:])
SYS.stdout.write(pre + '\b'*b + line[i:])
- #print ';; => %r' % (pre + '\b'*b + line[i:])
SYS.stdout.flush()
## Update our idea of what's gone on.
## Work out -- well, guess -- the time remaining.
if cur:
t = T.time()
- eta = me._fmt_time((t - me._start)*(max - cur)/cur)
+ eta = me._fmt_time(ceil((t - me._start)*(max - cur)/cur))
else:
eta = '???'
## Handy abbreviations for constructed parser elements.
def K(k): return P.Keyword(k).suppress()
def D(d): return P.Literal(d).suppress()
-##R = P.ZeroOrMore
def R(p): return P.ZeroOrMore(p).setParseAction(lambda s, l, t: [t])
O = P.Optional
named format and its superclasses.
"""
+ name = 'format-spec'
+
## We cache the parser elements we generate to avoid enormous consing.
CACHE = {}
class OggVorbisFormat (AudioFormat):
"AudioFormat object for Ogg Vorbis."
- ## From http://en.wikipedia.org/wiki/Vorbis
+ ## From https://en.wikipedia.org/wiki/Vorbis
QMAP = [(-1, 45), ( 0, 64), ( 1, 80), ( 2, 96),
( 3, 112), ( 4, 128), ( 5, 160), ( 6, 192),
( 7, 224), ( 8, 256), ( 9, 320), (10, 500)]
EXT = 'ogg'
def encoder_chain(me):
- for q, br in me.QMAP:
- if br >= me.bitrate:
- break
- else:
- raise ValueError, 'no suitable quality setting found'
- return [make_element('vorbisenc',
- quality = q/10.0),
+ encprops = {}
+ if me.bitrate is not None:
+ for q, br in me.QMAP:
+ if br >= me.bitrate:
+ break
+ else:
+ raise ValueError, 'no suitable quality setting found'
+ encprops['quality'] = q/10.0
+ return [make_element('vorbisenc', **encprops),
make_element('oggmux')]
defformat('ogg-vorbis', OggVorbisFormat)
EXT = 'mp3'
def encoder_chain(me):
- return [make_element('lame',
- vbr_mean_bitrate = me.bitrate,
- vbr = 4),
+ encprops = {}
+ if me.bitrate is not None: encprops['vbr_mean_bitrate'] = me.bitrate
+ return [make_element('lame', vbr = 4, **encprops),
make_element('xingmux'),
make_element('id3v2mux')]
GStreamer produces ID3v2 tags, but not ID3v1. This seems unnecessarily
unkind to stupid players.
"""
- tag = E3.Tag()
- tag.link(path)
- tag.setTextEncoding(E3.UTF_8_ENCODING)
- try:
- tag.update(E3.ID3_V1_1)
- except (UnicodeEncodeError, E3.tag.GenreException):
- pass
+ f = E3.load(path)
+ if f is None: return
+ t = f.tag
+ if t is None: return
+ for v in [E3.id3.ID3_V2_3, E3.id3.ID3_V1]:
+ try: f.tag.save(version = v)
+ except (UnicodeEncodeError,
+ E3.id3.GenreException,
+ E3.id3.TagException):
+ pass
defformat('mp3', MP3Format)
optimize
If present, take a second pass to select optimal encoder settings.
- progression
+ progressive
If present, make a progressive file.
quality Integer from 1--100 (worst to best); default is 75.
defformat('bmp', BMPFormat)
###--------------------------------------------------------------------------
+### Remaining parsing machinery.
+
+Type = K('type') - Name - D('{') - R(Policy) - D('}')
+def build_type(s, l, t):
+ try:
+ cat = CATEGORYMAP[t[0]]
+ except KeyError:
+ raise P.ParseException(s, loc, "Unknown category `%s'" % t[0])
+ pols = t[1]
+ if len(pols) == 1: pol = pols[0]
+ else: pol = AndPolicy(pols)
+ pol.setcategory(cat)
+ return pol
+Type.setParseAction(build_type)
+
+TARGETS = []
+class TargetJob (object):
+ def __init__(me, targetdir, policies):
+ me.targetdir = targetdir
+ me.policies = policies
+ def perform(me):
+ TARGETS.append(me)
+
+Target = K('target') - String - D('{') - R(Type) - D('}')
+def build_target(s, l, t):
+ return TargetJob(t[0], t[1])
+Target.setParseAction(build_target)
+
+VARS = { 'master': None }
+class VarsJob (object):
+ def __init__(me, vars):
+ me.vars = vars
+ def perform(me):
+ for k, v in me.vars:
+ VARS[k] = v
+
+Var = prop('master', String)
+Vars = K('vars') - D('{') - R(Var) - D('}')
+def build_vars(s, l, t):
+ return VarsJob(t[0])
+Vars.setParseAction(build_vars)
+
+TopLevel = Vars | Target
+Config = R(TopLevel)
+Config.ignore(P.pythonStyleComment)
+
+###--------------------------------------------------------------------------
### The directory grobbler.
-class Grobbler (object):
+def grobble(master, targets, noact = False):
"""
- The directory grobbler copies a directory tree, converting files.
+ Work through the MASTER directory, writing converted files to TARGETS.
+
+ The TARGETS are a list of `TargetJob' objects, each describing a target
+ directory and a policy to apply to it.
+
+ If NOACT is true, then don't actually do anything permanent to the
+ filesystem.
"""
- def __init__(me, policies, noact = False):
- """
- Create a new Grobbler, working with the given POLICIES.
- """
- me._pmap = {}
- me._noact = noact
- for p in policies:
- me._pmap.setdefault(p.cat, []).append(p)
- me._dirs = []
+ ## Transform the targets into a more convenient data structure.
+ tpolmap = []
+ for t in targets:
+ pmap = {}
+ tpolmap.append(pmap)
+ for p in t.policies: pmap.setdefault(p.cat, []).append(p)
- def _grobble_file(me, master, targetdir, cohorts):
- """
- Convert MASTER, writing the result to TARGETDIR.
+ ## Keep track of the current position in the master tree.
+ dirs = []
- The COHORTS are actually (CAT, ID, COHORT) triples, where a COHORT is a
- list of (FILENAME, ID) pairs.
+ ## And the files which haven't worked.
+ broken = []
- Since this function might convert the MASTER file, the caller doesn't
- know the name of the output files, so we return then as a list.
- """
+ def grobble_file(master, pmap, targetdir, cohorts):
+ ## Convert MASTER, writing the result to TARGETDIR.
+ ##
+ ## The COHORTS are actually (CAT, ID, COHORT) triples, where a COHORT is
+ ## a list of (FILENAME, ID) pairs.
+ ##
+ ## Since this function might convert the MASTER file, the caller doesn't
+ ## know the name of the output files, so we return then as a list.
done = set()
st_m = OS.stat(master)
## Go through the category's policies and see if any match. If we fail
## here, see if there are more categories to try.
- for pol in me._pmap[cat]:
+ for pol in pmap[cat]:
acts = pol.actions(master, targetdir, id, cohort)
if acts: break
else:
## Remove the target. (A hardlink will fail if the target already
## exists.)
- if not me._noact:
+ if not noact:
try:
OS.unlink(a.target)
except OSError, err:
raise
## Do whatever it is we decided to do.
- if me._noact:
+ if noact:
STATUS.commit(filestatus(master, a))
else:
a.perform()
return list(done)
@contextmanager
- def _wrap(me, masterfile):
- """
- Handle exceptions found while trying to convert a particular file or
- directory.
- """
+ def wrap(masterfile):
+ ## Handle exceptions found while trying to convert a particular file or
+ ## directory.
try:
yield masterfile
except (IOError, OSError), exc:
STATUS.clear()
STATUS.commit(filestatus(masterfile, 'failed (%s)' % exc))
- me._broken.append((masterfile, exc))
-
- def _grobble_dir(me, master, target):
- """
- Recursively convert files in MASTER, writing them to TARGET.
- """
-
- ## Make sure the TARGET exists and is a directory. It's a fundamental
- ## assumption of this program that the entire TARGET tree is disposable,
- ## so if something exists but isn't a directory, we should kill it.
- if OS.path.isdir(target):
- pass
- else:
- if OS.path.exists(target):
- STATUS.commit(filestatus(target, 'clear nondirectory'))
- if not me._noact:
- OS.unlink(target)
- STATUS.commit(filestatus(target, 'create directory'))
- if not me._noact:
- OS.mkdir(target)
-
- ## Keep a list of things in the target. As we convert files, we'll check
- ## them off. Anything left over is rubbish and needs to be deleted.
- checklist = {}
- try:
- for i in OS.listdir(target):
- checklist[i] = False
- except OSError, err:
- if err.errno not in (E.ENOENT, E.ENOTDIR):
- raise
-
- ## Keep track of the files in each category.
- catmap = {}
- todo = []
- done = []
-
- ## Work through the master files.
- for f in sorted(OS.listdir(master)):
-
- ## If the killswitch has been pulled then stop. The whole idea is that
- ## we want to cause a clean shutdown if possible, so we don't want to
- ## do it in the middle of encoding because the encoding effort will
- ## have been wasted. This is the only place we need to check. If
- ## we've exited the loop, then clearing old files will probably be
- ## fast, and we'll either end up here when the recursive call returns
- ## or we'll be in the same boat as before, clearing old files, only up
- ## a level. If worst comes to worst, we'll be killed forcibly
- ## somewhere inside `SH.rmtree', and that can continue where it left
- ## off.
- if KILLSWITCH.is_set():
- return
-
- ## Do something with the file.
- with me._wrap(OS.path.join(master, f)) as masterfile:
-
- ## If it's a directory then grobble it recursively. Keep the user
- ## amused by telling him where we are in the tree.
- if OS.path.isdir(masterfile):
- me._dirs.append(f)
- STATUS.set('/'.join(me._dirs))
- try:
- done += me._grobble_dir(masterfile, OS.path.join(target, f))
- finally:
- me._dirs.pop()
- STATUS.set('/'.join(me._dirs))
-
- ## Otherwise it's a file. Work out what kind, and stash it under
- ## the appropriate categories. Later, we'll apply policy to the
- ## files, by category, and work out what to do with them all.
- else:
- gf = GIO.File(masterfile)
- mime = gf.query_info('standard::content-type').get_content_type()
- cats = []
- for cat in me._pmap.iterkeys():
- id = cat.identify(masterfile, mime)
- if id is None: continue
- catmap.setdefault(cat, []).append((masterfile, id))
- cats.append((cat, id))
- if not cats:
- catmap.setdefault(None, []).append((masterfile, id))
- todo.append((masterfile, cats))
-
- ## Work through the categorized files to see what actions to do for
- ## them.
- for masterfile, cats in todo:
- with me._wrap(masterfile):
- done += me._grobble_file(masterfile, target,
- [(cat, id, catmap[cat])
- for cat, id in cats])
-
- ## Check the results off the list so that we don't clear it later.
- for f in done:
- checklist[OS.path.basename(f)] = True
-
- ## Maybe there's stuff in the target which isn't accounted for. Delete
- ## it: either the master has changed, or the policy for this target has
- ## changed. Either way, the old files aren't wanted.
- for f in checklist:
- if not checklist[f]:
- STATUS.commit(filestatus(f, 'clear bogus file'))
- if not me._noact:
- bogus = OS.path.join(target, f)
- try:
- if OS.path.isdir(bogus):
- SH.rmtree(bogus)
- else:
- OS.unlink(bogus)
- except OSError, err:
- if err.errno != E.ENOENT:
- raise
-
- ## Return the target name, so that it can be checked off.
- return [target]
-
- def grobble(me, master, target):
- """
- Convert MASTER, writing a directory tree TARGET.
-
- Returns a list of files which couldn't be converted.
- """
- try:
- me._broken = []
- me._grobble_dir(master, target)
- return me._broken
- finally:
- del me._broken
-
-###--------------------------------------------------------------------------
-### Remaining parsing machinery.
-
-Type = K('type') - Name - D('{') - R(Policy) - D('}')
-def build_type(s, l, t):
- try:
- cat = CATEGORYMAP[t[0]]
- except KeyError:
- raise P.ParseException(s, loc, "Unknown category `%s'" % t[0])
- pols = t[1]
- if len(pols) == 1: pol = pols[0]
- else: pol = AndPolicy(pols)
- pol.setcategory(cat)
- return pol
-Type.setParseAction(build_type)
+ broken.append((masterfile, exc))
-TARGETS = []
-class TargetJob (object):
- def __init__(me, targetdir, policies):
- me.targetdir = targetdir
- me.policies = policies
- def perform(me):
- TARGETS.append(me)
+ def grobble_dir(master, targets):
+ ## Recursively convert files in MASTER, writing them to the TARGETS.
-Target = K('target') - String - D('{') - R(Type) - D('}')
-def build_target(s, l, t):
- return TargetJob(t[0], t[1])
-Target.setParseAction(build_target)
+ ## Keep track of the subdirectories we encounter, because we'll need to
+ ## do all of those in one go at the end.
+ subdirs = set()
-VARS = { 'master': None }
-class VarsJob (object):
- def __init__(me, vars):
- me.vars = vars
- def perform(me):
- for k, v in me.vars:
- VARS[k] = v
+ ## Work through each target directory in turn.
+ for target, pmap in zip(targets, tpolmap):
-Var = prop('master', String)
-Vars = K('vars') - D('{') - R(Var) - D('}')
-def build_vars(s, l, t):
- return VarsJob(t[0])
-Vars.setParseAction(build_vars)
+ ## Make sure the TARGET exists and is a directory. It's a fundamental
+ ## assumption of this program that the entire TARGET tree is
+ ## disposable, so if something exists but isn't a directory, we should
+ ## kill it.
+ if OS.path.isdir(target):
+ pass
+ else:
+ if OS.path.exists(target):
+ STATUS.commit(filestatus(target, 'clear nondirectory'))
+ if not noact:
+ OS.unlink(target)
+ STATUS.commit(filestatus(target, 'create directory'))
+ if not noact:
+ OS.mkdir(target)
+
+ ## Keep a list of things in the target. As we convert files, we'll
+ ## check them off. Anything left over is rubbish and needs to be
+ ## deleted.
+ checklist = {}
+ try:
+ for i in OS.listdir(target):
+ checklist[i] = False
+ except OSError, err:
+ if err.errno not in (E.ENOENT, E.ENOTDIR):
+ raise
+
+ ## Keep track of the files in each category.
+ catmap = {}
+ todo = []
+ done = []
+
+ ## Work through the master files.
+ for f in sorted(OS.listdir(master)):
+
+ ## If the killswitch has been pulled then stop. The whole idea is
+ ## that we want to cause a clean shutdown if possible, so we don't
+ ## want to do it in the middle of encoding because the encoding
+ ## effort will have been wasted. This is the only place we need to
+ ## check. If we've exited the loop, then clearing old files will
+ ## probably be fast, and we'll either end up here when the recursive
+ ## call returns or we'll be in the same boat as before, clearing old
+ ## files, only up a level. If worst comes to worst, we'll be killed
+ ## forcibly somewhere inside `SH.rmtree', and that can continue where
+ ## it left off.
+ if KILLSWITCH.is_set():
+ return
+
+ ## Do something with the file.
+ with wrap(OS.path.join(master, f)) as masterfile:
+
+ ## If it's a directory then prepare to grobble it recursively, but
+ ## don't do that yet.
+ if OS.path.isdir(masterfile):
+ subdirs.add(f)
+ done.append(OS.path.join(target, f))
+
+ ## Otherwise it's a file. Work out what kind, and stash it under
+ ## the appropriate categories. Later, we'll apply policy to the
+ ## files, by category, and work out what to do with them all.
+ else:
+ gf = GIO.File(masterfile)
+ mime = gf.query_info('standard::content-type').get_content_type()
+ cats = []
+ for cat in pmap.iterkeys():
+ id = cat.identify(masterfile, mime)
+ if id is None: continue
+ catmap.setdefault(cat, []).append((masterfile, id))
+ cats.append((cat, id))
+ if not cats:
+ catmap.setdefault(None, []).append((masterfile, id))
+ todo.append((masterfile, cats))
+
+ ## Work through the categorized files to see what actions to do for
+ ## them.
+ for masterfile, cats in todo:
+ with wrap(masterfile):
+ done += grobble_file(masterfile, pmap, target,
+ [(cat, id, catmap[cat]) for cat, id in cats])
+
+ ## Check the results off the list so that we don't clear it later.
+ for f in done:
+ checklist[OS.path.basename(f)] = True
+
+ ## Maybe there's stuff in the target which isn't accounted for. Delete
+ ## it: either the master has changed, or the policy for this target has
+ ## changed. Either way, the old files aren't wanted.
+ for f in checklist:
+ if not checklist[f]:
+ STATUS.commit(filestatus(f, 'clear bogus file'))
+ if not noact:
+ bogus = OS.path.join(target, f)
+ try:
+ if OS.path.isdir(bogus):
+ SH.rmtree(bogus)
+ else:
+ OS.unlink(bogus)
+ except OSError, err:
+ if err.errno != E.ENOENT:
+ raise
+
+ ## If there are subdirectories which want processing then do those.
+ ## Keep the user amused by telling him where we are in the tree.
+ for d in sorted(subdirs):
+ dirs.append(d)
+ STATUS.set('/'.join(dirs))
+ with wrap(OS.path.join(master, d)) as masterdir:
+ try:
+ grobble_dir(masterdir,
+ [OS.path.join(target, d) for target in targets])
+ finally:
+ dirs.pop()
+ STATUS.set('/'.join(dirs))
-TopLevel = Vars | Target
-Config = R(TopLevel)
-Config.ignore(P.pythonStyleComment)
+ ## Right. We're ready to go.
+ grobble_dir(master, [t.targetdir for t in targets])
+ return broken
###--------------------------------------------------------------------------
### Command-line interface.
## Build the option parser object.
op = OP.OptionParser(prog = QUIS, version = VERSION,
- usage = '%prog [-t TIMEOUT] CONFIG',
+ usage = '%prog [-in] [-t TIMEOUT] [-T TIMEOUT] '
+ 'CONFIG',
description = """\
Convert a directory tree of files according to the configuration file
CONFIG.
opts = parse_opts(SYS.argv[1:])
if 'master' not in VARS:
die("no master directory set")
- broken = []
- for t in TARGETS:
- g = Grobbler(t.policies, opts.noact)
- b = g.grobble(VARS['master'], t.targetdir)
- broken += b
+ broken = grobble(VARS['master'], TARGETS, opts.noact)
if broken:
moan('failed to convert some files:')
for file, exc in broken:
~mdw / autoys / blobdiff