Source code for blissoda.demo.processors.streamline

import os

from ...streamline.scanner import StreamlineScanner
from .utils.streamline import DemoStreamlineScannerMixIn




class DemoStreamlineScanner(DemoStreamlineScannerMixIn, StreamlineScanner):
    _TEST_DATA_PATHS_WITHOUT_CALIB = (
        "/results/integrate/integrated/q",
        "/results/difflab6_integrate_2th/integrated/2th",
    )
    _TEST_DATA_PATHS_WITH_CALIB = (
        "/results/integrate/integrated/q",
        "/results/difflab6_integrate_2th/integrated/2th",
    )
    _TEST_ASCII_SUFFIXES_WITHOUT_CALIB = tuple()
    _TEST_ASCII_SUFFIXES_WITH_CALIB = tuple()



    def calib(self, *args, **kwargs):
        self._requires_workflow()
        super().calib(*args, **kwargs)




    def run(self, *args, **kwargs):
        self._requires_workflow()
        super().run(*args, **kwargs)


    def _requires_workflow(self):
        if not os.path.exists(self.workflow):
            raise RuntimeError(
                "the workflow file no longer exists, execute 'init_workflow' again"
            )