process.py

import argparse
import sys
from pypushflow.Workflow import ProcessableWorkflow
from pypushflow.representation.scheme.ows_parser import OwsParser

try:
    from est.io.utils.pymca import read_spectrum

    has_read_spectrum = True
except ImportError:
    has_read_spectrum = False
from est.core.types import Spectrum, XASObject
from est.core.io import read_xas
import logging
import signal
from pypushflow.representation.scheme.scheme import Scheme

try:
    import h5py

    has_h5py = True
except:
    has_h5py = False
from typing import Union

logging.basicConfig(level=logging.DEBUG)
_logger = logging.getLogger(__name__)


def exec_(
    scheme: Scheme,
    input_: Union[str, None, dict] = None,
    output_: Union[str, None, dict] = None,
    timeout: Union[int, None] = None,
):
    def manage_input():
        if input_:
            # monkey patch the input file for start nodes if an input is given
            for node in scheme.nodes:
                if node.properties and "_input_file_setting" in node.properties:
                    node.properties["_input_file_setting"] = input_

    manage_input()

    def manage_output():
        if output_:
            found_output = False
            # monkey patch the input file for start nodes if an input is given
            for node in scheme.nodes:
                if node.properties and "_output_file_setting" in node.properties:
                    node.properties["_output_file_setting"] = output_
                    found_output = True
            if not found_output:
                _logger.warning(
                    "No node for processing output found. output "
                    "information provided will be ignored"
                )

    # TODO: if the input file contains configuration, then overwrite the node
    # properties...
    manage_output()

    workflow = ProcessableWorkflow(scheme=scheme)

    # add SIGINT capture
    def signal_handler(sig, frame):
        _logger.warning("stop workflow execution on user request")
        workflow._end_actor.join(0)
        sys.exit(0)

    signal.signal(signal.SIGINT, signal_handler)

    # TODO: handle several entries from url on top of .dat file... using read_xas
    if has_read_spectrum:
        energy, mu = read_spectrum(input_)
        spectrum = Spectrum(energy=energy, mu=mu)
        xas_obj = XASObject(energy=energy, spectra=(spectrum,), dim1=1, dim2=1)
    else:
        raise ValueError("Unable to read spectrum")

    workflow._start_actor.trigger(("data", xas_obj.to_dict()))
    workflow._end_actor.join(timeout)
    res = workflow._end_actor.out_data
    title = scheme.title or "unknow"
    _logger.info("workflow '{}' completed with {}".format(title, str(input_)))
    return res


def getinputinfo():
    return "xas process workflow_file_desc.ows [input]"


def getInputFrmHdf5(file_path):
    raise NotImplementedError()


def main(argv):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "workflow_file",
        help="Workflow file providing the workflow description (.ows, .xml)",
    )
    parser.add_argument(
        "-i", "--input",
        dest="input_",
        default=None,
        help="Input of the workflow. require at most one " "instance of XASInputOW",
    )
    parser.add_argument(
        "-o", "--output",
        dest="output_",
        default=None,
        help="Output file of the workflow. Require at most one "
        "instance of XASOutputOW",
    )
    options = parser.parse_args(argv[1:])

    scheme = OwsParser.scheme_load(options.workflow_file, load_handlers=True)
    exec_(scheme, options.input_, options.output_)


if __name__ == "__main__":
    main(sys.argv)