Overcomplicated untested work on automagical ninja pipeline for X-ray processing

chandra_repro.py

# -*- mode: python; coding: utf-8 -*-
# Copyright 2016 Peter Williams <[email protected]> and collaborators.
# Licensed under the MIT License.

"""Framework to call the CIAO chandra_repro program in the context of a Ninja
build system.

"""
from __future__ import absolute_import, division, print_function, unicode_literals

__all__ = str('''
ChandraReproCommand
ChandraReproNinja
''').split ()

import six
from ...io import Path
from ...cli import die, multitool, warn, wrapout
from ...ninja import Rule, Target


# XXX ugh hardcoding outputs

def rename_one (outpath, outname, glob):
    matches = list (outpath.glob (glob))
    if len (matches) != 1:
        warn ('multiple matches for %s: <%s> <%s> <%s>', glob, matches, outpath, outname)
    matches[0].rename (outpath / outname)

def build_asol_list (outpath, outname, dummyarg):
    for asol_lis in outpath.glob ('*asol1.lis'):
        asol_lis.unlink ()

    with (outpath / 'asol.lis').open ('wb') as f:
        for asol in outpath.glob ('*asol1.fits'):
            print (asol.absolute (), file=f)

repro_outputs = [
    ('rbpix.fits', rename_one, '*repro_bpix1.fits'),
    ('rfov.fits', rename_one, '*repro_fov1.fits'),
    ('msk.fits', rename_one, '*msk1.fits'),
    ('mtl.fits', rename_one, '*mtl1.fits'),
    ('stat.fits', rename_one, '*stat1.fits'),
    ('revt.fits', rename_one, '*repro_evt2.fits'),
    ('rflt.fits', rename_one, '*repro_flt2.fits'),
    ('pbk.fits', rename_one, '*pbk0.fits'),
    ('asol.lis', build_asol_list, 'dummy'),
]


class ChandraReproCommand (multitool.Command):
    name = 'chandra-repro'
    argspec = '<in> <out> [keywords...]'
    summary = 'Ninja-friendly version of chandra_repro tool.'

    def invoke (self, args, envclass=None, **kwargs):
        if len (args) < 2:
            raise multitool.UsageError ('chandra-repro takes at least two arguments')

        inpath = Path (args[0])
        outpath = Path (args[1])
        passthrough = args[2:]

        env = envclass ()

        outpath.rmtree (errors='ignore')
        outpath.ensure_dir (parents=True)

        argv = ['chandra_repro', str(inpath), str(outpath)] + passthrough

        with (outpath / 'repro.log').open ('wb') as log:
            w = env.get_wrapout_wrapper (destination=log)
            rc = w.launch ('chandra_repro', argv)

        if rc: # error?
            import sys
            with (outpath / 'repro.log').open ('r') as log:
                for line in log:
                    print (line.strip (), file=sys.stderr)
            die ('command failed')

        for outname, func, arg in repro_outputs:
            func (outpath, outname, arg)


class ChandraReproNinja (object):
    def __init__ (self, spec):
        self.spec = spec
        spec.add_rule (Rule ('chandra_repro',
                             'pkenvtool ciao chandra-repro $in $outdir $keywords',
                             description='REPRO $outdir'))


    def reprocess (self, inpath, outpath, keywords=''):
        outputs = {}

        for outname, func, arg in repro_outputs:
            outbase = outname.split ('.')[0]
            outputs[outbase] = outpath / outname

        self.spec.add_target (Target (
            six.viewvalues (outputs), 'chandra_repro',
            inputs=inpath,
            implicit=__file__,
            variables={
                'outdir': str(outpath),
                'keywords': keywords,
            },
        ))

        return outputs

dmcopy.py

# -*- mode: python; coding: utf-8 -*-
# Copyright 2016 Peter Williams <[email protected]> and collaborators.
# Licensed under the MIT License.

"""Framework to call the CIAO dmcopy program in the context of a Ninja build
system.

"""
from __future__ import absolute_import, division, print_function, unicode_literals

__all__ = str('''
DmcopyCommand
DmcopyNinja
''').split ()

import six
from ...io import Path
from ...cli import die, multitool, warn, wrapout
from ...ninja import Rule, Target


class DmcopyCommand (multitool.Command):
    name = 'dmcopy'
    argspec = '<in> <out> [keywords...]'
    summary = 'Ninja-friendly version of dmcopy tool.'

    def invoke (self, args, envclass=None, **kwargs):
        if len (args) < 2:
            raise multitool.UsageError ('dmcopy takes at least two arguments')

        inspec = args[0] # may have extra filters, etc.
        outpath = Path (args[1])
        passthrough = args[2:]

        env = envclass ()

        outpath.ensure_parent (parents=True)
        outpath.unlink ()

        # dmcopy is silent by default so we can log to stdout
        argv = ['dmcopy', inspec, str(outpath)] + passthrough
        if env.launch (argv).wait ():
            die ('command failed')


class DmcopyNinja (object):
    def __init__ (self, spec):
        self.spec = spec
        spec.add_rule (Rule ('dmcopy',
                             'pkenvtool ciao dmcopy "$in$infilters" $out $keywords',
                             description='DMCOPY $out'))


    def copy (self, inpath, outpath, infilters='', keywords=''):
        self.spec.add_target (Target (
            outpath, 'dmcopy',
            inputs=inpath,
            implicit=__file__,
            variables={
                'infilters': infilters,
                'keywords': keywords,
            },
        ))

ninja.py

# -*- mode: python; coding: utf-8 -*-
# Copyright 2016 Peter Williams <[email protected]> and collaborators.
# Licensed under the MIT License.

"""Simple framework for generating Ninja build files and running the tool.

The `ninja_syntax` module applies no structure whatsoever. This one aims to
provide a very simple framework for building Ninja files by composing several
pieces.

"""
from __future__ import absolute_import, division, print_function, unicode_literals

__all__ = str('''
Rule
Target
BuildSpecification
''').split ()

from collections import namedtuple
import six
from .io import Path
from . import ninja_syntax


def _str_or_none (x):
    if x is None:
        return None
    return str (x)

_RuleBase = namedtuple ('_RuleBase', '''name command description depfile generator pool restat rspfile
                        rspfile_content deps'''.split ())

class Rule (_RuleBase):
    """Definition of a build rule. This is implemented as a named tuple to keep
    this unmodifiable so that we can test for rule equality.

    """
    def __new__ (cls, name, command, description=None, depfile=None,
                  generator=False, pool=None, restat=False, rspfile=None,
                  rspfile_content=None, deps=None):
        return super (Rule, cls).__new__ (cls,
                                          str (name),
                                          str (command),
                                          _str_or_none (description),
                                          _str_or_none (depfile),
                                          bool (generator),
                                          _str_or_none (pool),
                                          bool (restat),
                                          _str_or_none (rspfile),
                                          _str_or_none (rspfile_content),
                                          _str_or_none (deps))

    def emit (self, spec, writer):
        writer.rule (self.name, self.command, self.description, self.depfile,
                     self.generator, self.pool, self.restat, self.rspfile,
                     self.rspfile_content, self.deps)

    def __repr__ (self):
        extras = ['cmd=' + repr(self.command)]
        if self.description is not None:
            extras.append ('desc=' + repr(self.description))
        if self.depfile is not None:
            extras.append ('depfile=' + self.depfile)
        if self.generator:
            extras.append ('generator')
        if self.pool is not None:
            extras.append ('pool=' + self.pool)
        if self.restat:
            extras.append ('restat')
        if self.rspfile is not None:
            extras.append ('rspfile=' + self.rspfile)
        if self.rspfile_content is not None:
            extras.append ('rspfile_content=' + self.rspfile_content)
        if self.deps is not None:
            extras.append ('deps=' + repr(self.deps))
        return '<Rule %s %s>' % (self.name, ' '.join (extras))


def _to_str_tuple (x):
    if x is None:
        return ()

    if isinstance (x, six.string_types):
        return (x,)

    try:
        iter (x)
        return tuple (str(i) for i in x)
    except TypeError:
        return (str (x),)


def _to_frozen_str_dict (d):
    return tuple (sorted (((str(k), str(v))
                           for k, v in six.viewitems (d)),
                          key=lambda t: t[0]))


def _stringify_str_tuple (t):
    if len (t) == 1:
        if ' ' in t[0]:
            return repr (t[0])
        return t[0]

    return '(%s)' % (', '.join (t))


_TargetBase = namedtuple ('_TargetBase', '''outputs rule inputs implicit order_only variables'''.split ())

class Target (_TargetBase):
    """Definition of a build target. This is implemented as a named tuple to keep
    it unmodifiable so that we can test for target equality.

    """
    def __new__ (cls, outputs, rule, inputs=None, implicit=None, order_only=None, variables={}):
        return super (Target, cls).__new__ (cls,
                                            _to_str_tuple (outputs),
                                            str (rule),
                                            _to_str_tuple (inputs),
                                            _to_str_tuple (implicit),
                                            _to_str_tuple (order_only),
                                            _to_frozen_str_dict (variables))

    def emit (self, spec, writer):
        writer.build (list(self.outputs), self.rule, list(self.inputs),
                      list(self.implicit), list(self.order_only),
                      self.variables)

    def __repr__ (self):
        deps = _stringify_str_tuple (self.inputs)
        if len (self.implicit):
            deps += ' | ' + _stringify_str_tuple (self.implicit)
        if len (self.order_only):
            deps += ' || ' + _stringify_str_tuple (self.order_only)
        if len (self.variables):
            deps += '; ' + ' '.join ('$%s=%r' % kv for kv in self.variables)

        return '<Target %s = %s: %s>' % (
            _stringify_str_tuple (self.outputs),
            self.rule,
            deps
        )


class BuildSpecification (object):
    topdir = None
    "A `pwkit.io.Path` giving the top of the build directory."

    preambles = None
    "A list of functions to call before doing anything else."

    _rules = None
    "A `dict` mapping rule names to their specifications."

    _targets = None
    """A `dict` mapping file names to target specifications. Note that one target
    may produce multiple output files.

    """
    def __init__ (self, topdir):
        self.topdir = Path (topdir)
        self.preambles = []
        self._rules = {}
        self._targets = {}


    def add_rule (self, rule):
        prev = self._rules.get (rule.name)
        if prev is None:
            self._rules[rule.name] = rule
        elif prev != rule:
            raise ValueError ('a rule named "%s" already exists' % rule.name)
        return rule


    def add_target (self, target):
        for o in target.outputs:
            prev = self._targets.get (o)
            if prev is None:
                self._targets[o] = target
            elif prev != target:
                raise ValueError ('a target producing "%s" already exists' % o)
        return target


    def write (self):
        with (self.topdir / 'build.ninja').make_tempfile (want='handle', resolution='overwrite') as dest:
            w = ninja_syntax.Writer (dest)
            w.comment ('Automatically generated.')

            for prefunc in self.preambles:
                prefunc (self, w)

            for name in sorted (six.viewkeys (self._rules)):
                self._rules[name].emit (self, w)

            done_targets = set ()

            for name in sorted (six.viewkeys (self._targets)):
                t = self._targets[name]
                if t in done_targets:
                    continue

                if t.rule not in self._rules:
                    raise Exception ('target %s references an undefined rule' % (t,))

                t.emit (self, w)
                done_targets.add (t)

The ninja module wasn't properly designed since the whole idea is that objects should be able to reference each other and say "I depend on target X" and without having to worry about the paths.