power-fv/power_fv/session.py

import argparse
import os
import multiprocessing
import sys

from power_fv.build import sby
from power_fv.core import PowerFVCore
from power_fv.check import PowerFVCheck
from power_fv.check.all import *

from pprint import pprint


__all__ = ["PowerFVSession"]


class PowerFVCommandExit(Exception):
    pass

class PowerFVCommandError(Exception):
    pass


class _ArgumentParser(argparse.ArgumentParser):
    def exit(self, status=0, message=None):
        raise PowerFVCommandExit()

    def error(self, message):
        raise PowerFVCommandError()


class PowerFVSession:
    def __init_subclass__(cls, *, core_cls, **kwargs):
        super().__init_subclass__(**kwargs)
        if not issubclass(core_cls, PowerFVCore):
            raise TypeError("Core class must be a subclass of PowerFVCore, not {!r}"
                            .format(core_cls))
        cls.core_cls = core_cls

    def __init__(self, prog=None):
        self.parser     = _ArgumentParser(prog=prog, add_help=False)
        self.subparsers = self.parser.add_subparsers(help="commands")
        self._checks    = dict()

        self.add_help_subparser()
        self.add_check_subparser()
        self.add_dump_subparser()
        self.add_build_subparser()
        self.add_exit_subparser()

    def main(self):
        parser = argparse.ArgumentParser(prog=self.parser.prog)
        group  = parser.add_mutually_exclusive_group()
        group.add_argument(
            "-i", dest="interact", action="store_true",
            help="run commands interactively")
        group.add_argument(
            "-c", dest="cmdfile", type=argparse.FileType("r"), default=None,
            help="run commands from CMDFILE")

        args = parser.parse_args()
        try:
            self._loop(args)
        except EOFError:
            pass

    def _loop(self, args):
        if args.cmdfile is None:
            _readline = lambda: input("powerfv> ")
        else:
            _readline = args.cmdfile.readline

        while True:
            line = _readline().rstrip("\n")
            if not line or line.startswith("#"):
                continue
            self._eval(line.split())

    def _eval(self, args=None):
        try:
            args = self.parser.parse_args(args)
            assert hasattr(args, "_cmd")
            cmd = args._cmd
            del args._cmd
            cmd(**vars(args))
        except PowerFVCommandExit:
            pass
        except PowerFVCommandError:
            self.help()

    # Subparsers

    def add_help_subparser(self):
        parser = self.subparsers.add_parser("help", help="show this help message")
        parser.set_defaults(_cmd=self.help)

    def add_check_subparser(self):
        parser = self.subparsers.add_parser("check", help="add checks to this session")
        parser.set_defaults(_cmd=self.check)

        PowerFVCheck .add_check_arguments(parser)
        self.core_cls.add_check_arguments(parser)

        parser.add_argument(
            "--exclude", type=str, default="",
            help="exclude a comma-separated list of checks from the selection")

    def add_dump_subparser(self):
        parser = self.subparsers.add_parser("dump", help="inspect check parameters")
        parser.set_defaults(_cmd=self.dump)

    def add_build_subparser(self):
        parser = self.subparsers.add_parser("build", help="execute the build plan")
        parser.set_defaults(_cmd=self.build)

        parser.add_argument(
            "-j", "--jobs", type=int, default=os.cpu_count(),
            help="number of worker processes (default: %(default)s)")

        PowerFVCheck .add_build_arguments(parser)
        self.core_cls.add_build_arguments(parser)

    def add_exit_subparser(self):
        parser = self.subparsers.add_parser("exit", help="exit")
        parser.set_defaults(_cmd=self.exit)

    # Commands

    def help(self, **kwargs):
        self.parser.print_help()

    def check(self, *, name, exclude, **kwargs):
        exclude    = [f"{name}:{subname}" for subname in exclude.split(",")]
        matches    = list(PowerFVCheck.find(name))
        new_checks = dict()

        for check_name, check_cls in matches:
            if check_name in exclude:
                continue
            new_checks[check_name] = dict(**kwargs)

        self._checks.update(new_checks)

    def dump(self, **kwargs):
        pprint(self._checks, sort_dicts=False)

    @staticmethod
    def _build_check(core_cls, check_name, check_args, build_args):
        check_cls = PowerFVCheck.all_checks[tuple(check_name.split(":"))]
        core  = core_cls()
        check = check_cls(core=core, **check_args)
        check.build(**build_args)

    def build(self, *, jobs, **kwargs):
        map_func = PowerFVSession._build_check
        map_args = []

        for check_name, check_args in self._checks.items():
            map_args.append((self.core_cls, check_name, check_args, kwargs))

        with multiprocessing.Pool(jobs) as pool:
            pool.starmap(map_func, map_args)

    def exit(self, **kwargs):
        print("exiting")
        sys.exit()