Source code for ewoksfluo.tests.tutorials.assert_results

from typing import Any
from typing import Dict
from typing import List
from typing import Tuple

try:
    from importlib.resources import files as resource_files
except ImportError:
    from importlib_resources import files as resource_files

from ewokscore import execute_graph
from ewoksorange.gui.workflows.owscheme import ows_to_ewoks

from ..utils import hdf5




def assert_without_qt(
    basename: str,
    inputs: List[Dict[str, Any]],
    expected_outputs: Dict[str, Dict[str, Any]],
    expected_hdf5: Tuple[str, Dict[str, Any]],
) -> None:
    """Execute workflow after converting it to an ewoks workflow"""
    from orangecontrib.ewoksfluo.categories.demo import tutorials

    filename = resource_files(tutorials).joinpath(basename)

    graph = ows_to_ewoks(filename)
    outputs = execute_graph(
        graph, inputs=inputs, outputs=[{"all": True}], merge_outputs=False
    )

    expected = expected_outputs
    label_to_id = {
        attrs["label"]: node_id for node_id, attrs in graph.graph.nodes.items()
    }
    expected = {label_to_id[k]: v for k, v in expected.items()}
    assert outputs == expected

    output_path, expected = expected_hdf5
    hdf5.assert_hdf5_content(output_path, expected)





def assert_with_qt(
    ewoks_orange_canvas,
    basename: str,
    inputs: List[Dict[str, Any]],
    expected_outputs: Dict[str, Dict[str, Any]],
    expected_hdf5: Tuple[str, Dict[str, Any]],
) -> None:
    """Execute workflow using the Qt widgets and signals"""
    from orangecontrib.ewoksfluo.categories.demo import tutorials

    filename = resource_files(tutorials).joinpath(basename)

    ewoks_orange_canvas.load_graph(str(filename), inputs=inputs)
    ewoks_orange_canvas.start_workflow()
    ewoks_orange_canvas.wait_widgets(timeout=30)

    outputs = dict(ewoks_orange_canvas.iter_output_values())
    assert outputs == expected_outputs

    output_path, expected = expected_hdf5
    hdf5.assert_hdf5_content(output_path, expected)