flowcraft/generator/engine.py
def _set_status_channels(self):
        """Compiles all status channels for the status compiler process
        """

        status_inst = pc.StatusCompiler(template="status_compiler")
        report_inst = pc.ReportCompiler(template="report_compiler")

        # Compile status channels from pipeline process
        status_channels = []
        for p in [p for p in self.processes]:
            if not any([isinstance(p, x) for x in self.skip_class]):

                status_channels.extend(p.status_strs)

        if not status_channels:
            logger.debug("No status channels found. Skipping status compiler"
                         "process")
            return

        logger.debug("Setting status channels: {}".format(status_channels))

        # Check for duplicate channels. Raise exception if found.
        if len(status_channels) != len(set(status_channels)):
            raise eh.ProcessError(
                "Duplicate status channels detected. Please ensure that "
                "the 'status_channels' attributes of each process are "
                "unique. Here are the status channels:\n\n{}".format(
                    ", ".join(status_channels)
                ))

        status_inst.set_compiler_channels(status_channels)

        report_channels = ["REPORT_{}".format(x.lstrip("STATUS_")) for x in
                           status_channels]

        report_inst.set_compiler_channels(report_channels)

        self.processes.extend([status_inst, report_inst])
def _set_status_channels(self):
        """Compiles all status channels for the status compiler process
        """

        status_inst = pc.StatusCompiler(template="status_compiler")
        report_inst = pc.ReportCompiler(template="report_compiler")

        # Compile status channels from pipeline process
        status_channels = []
        for p in [p for p in self.processes]:
            if not any([isinstance(p, x) for x in self.skip_class]):

                status_channels.extend(p.status_strs)

        if not status_channels:
            logger.debug("No status channels found. Skipping status compiler"
                         "process")
            return

        logger.debug("Setting status channels: {}".format(status_channels))

        # Check for duplicate channels. Raise exception if found.
        if len(status_channels) != len(set(status_channels)):
            raise eh.ProcessError(
                "Duplicate status channels detected. Please ensure that "
                "the 'status_channels' attributes of each process are "
                "unique. Here are the status channels:\n\n{}".format(
                    ", ".join(status_channels)
                ))

        status_inst.set_compiler_channels(status_channels)

        report_channels = ["REPORT_{}".format(x.lstrip("STATUS_")) for x in
                           status_channels]

        report_inst.set_compiler_channels(report_channels)

        self.processes.extend([status_inst, report_inst])
def _set_status_channels ( self ) : status_inst = pc . StatusCompiler ( template = "status_compiler" ) report_inst = pc . ReportCompiler ( template = "report_compiler" ) # Compile status channels from pipeline process status_channels = [ ] for p in [ p for p in self . processes ] : if not any ( [ isinstance ( p , x ) for x in self . skip_class ] ) : status_channels . extend ( p . status_strs ) if not status_channels : logger . debug ( "No status channels found. Skipping status compiler" "process" ) return logger . debug ( "Setting status channels: {}" . format ( status_channels ) ) # Check for duplicate channels. Raise exception if found. if len ( status_channels ) != len ( set ( status_channels ) ) : raise eh . ProcessError ( "Duplicate status channels detected. Please ensure that " "the 'status_channels' attributes of each process are " "unique. Here are the status channels:\n\n{}" . format ( ", " . join ( status_channels ) ) ) status_inst . set_compiler_channels ( status_channels ) report_channels = [ "REPORT_{}" . format ( x . lstrip ( "STATUS_" ) ) for x in status_channels ] report_inst . set_compiler_channels ( report_channels ) self . processes . extend ( [ status_inst , report_inst ] )
Compiles all status channels for the status compiler process
Compiles all status channels for the status compiler process