Source code for blissoda.demo.processors.id22

import os
from pprint import pprint
from typing import Any
from typing import Dict
from typing import Optional

from ...bliss_globals import current_session
from ...id22.stscan_processor import StScanProcessor
from ...utils import directories




class DemoStScanProcessor(StScanProcessor):
    def __init__(
        self,
        config: Optional[Dict[str, Any]] = None,
        defaults: Optional[Dict[str, Any]] = None,
        **deprecated_defaults: Dict[str, Any],
    ) -> None:
        defaults = self._merge_defaults(deprecated_defaults, defaults)

        if self._HAS_BLISS:
            root_dir = directories.get_processed_dir(
                current_session.scan_saving.filename
            )
            root_dir = os.path.join(root_dir, "demo", "id22", "config")

            defaults.setdefault(
                "_convert_workflow",
                os.path.join(root_dir, "convert.json"),
            )
            defaults.setdefault(
                "_rebinsum_workflow",
                os.path.join(root_dir, "rebinsum.json"),
            )
            defaults.setdefault(
                "_extract_workflow",
                os.path.join(root_dir, "extract.json"),
            )

        super().__init__(config=config, defaults=defaults)

    def _submit_job(self, workflow, inputs, convert_destination, **kw):
        print("\nSubmit workfow")
        print(workflow)
        print("Inputs:")
        pprint(inputs)
        print("Save for provenance:")
        pprint(convert_destination)
        print("Options:")
        pprint(kw)

    def _get_workflows_dir(self, dataset_filename: str) -> str:
        return os.path.join(directories.get_workflows_dir(dataset_filename), "id22")