Skip to content
GitLab
Commit 8a82c649 authored by payno's avatar payno
Browse files

remove test_workflow_concept. Those have been moved to est/test/test_ewoks

parent c0586855
Pipeline #51036 failed with stages
in 7 minutes and 59 seconds
Hide whitespace changes
Inline Side-by-side
orangecontrib/est/test/test_workflow_concept.py deleted 100644 → 0
View file @ c0586855
# coding: utf-8
# /*##########################################################################
#
# Copyright (c) 2016-2017 European Synchrotron Radiation Facility
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# ###########################################################################*/
__authors__ = ["H. Payno"]
__license__ = "MIT"
__date__ = "15/03/2021"
import unittest
from est.core.utils import DownloadDataset
from esrf2orange3.owsconvert import ows_to_esrf, esrf_to_ows
from est.io.utils.information import InputInformation
from silx.io.url import DataUrl
from esrftaskgraph.graph import TaskGraph
from silx.io.utils import h5py_read_dataset
import tempfile
import shutil
import os
import h5py
try:
import PyMca5
except ImportError:
has_pymca = False
else:
has_pymca = True
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
@unittest.skipIf(has_pymca is False, "PyMca5 is not installed")
class TestOwsToGraph(unittest.TestCase):
"""Test conversion of a .ows using est classes to an esrf graph"""
def setUp(self) -> None:
self.folder = tempfile.mkdtemp()
file_pymca = "ows_files/example_pymca.ows"
file_larch = "ows_files/example_larch.ows"
for file_ in (file_larch, file_pymca):
DownloadDataset(dataset=file_, output_folder=self.folder, timeout=2.0)
self.file_pymca = os.path.join(self.folder, "example_pymca.ows")
self.file_larch = os.path.join(self.folder, "example_larch.ows")
input_file1 = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
self.output_file = os.path.join(self.folder, "output.h5")
self.input_information = InputInformation(
spectra_url=DataUrl(
file_path=input_file1, scheme="PyMca", data_path="Column 1"
),
channel_url=DataUrl(
file_path=input_file1, scheme="PyMca", data_path="Column 2"
),
)
def tearDown(self) -> None:
shutil.rmtree(self.folder)
def testTranslationLarch(self):
"""Insure we can translate a workflow using pymca to a graph and
execute it"""
self.input_information.energy_unit = "keV"
graph = ows_to_esrf(self.file_larch)
# provide input information to dict
# this should be done by changing the .ows or something like
# that but not at this level because we are already on the
# instance of graph
graph.graph.nodes["input"]["inputs"] = self.input_information.to_dict()
graph.graph.nodes["output"]["inputs"] = {"output_file": self.output_file}
self.assertTrue(isinstance(graph, TaskGraph))
self.assertEqual(len(graph.graph.nodes()), 5)
self.assertEqual(len(graph.graph.edges()), 4)
graph.run({"root_uri": tempfile.mkdtemp()})
# check the generated process file
self.assertTrue(os.path.exists(self.output_file))
with h5py.File(self.output_file, mode="r") as h5f:
self.assertTrue("scan1" in h5f)
scan1_grp = h5f["scan1"]
for process_name in ("xas_process_1", "xas_process_2", "xas_process_3"):
self.assertTrue(process_name in scan1_grp)
process_2 = scan1_grp["xas_process_2"]
from est.core.process.larch.autobk import Larch_autobk
self.assertEqual(
h5py_read_dataset(process_2["program"]), Larch_autobk.program_name()
)
self.assertTrue("results" in process_2)
from silx.io.dictdump import h5todict
results = h5todict(h5file=self.output_file, path="scan1/xas_process_2/results")
self.assertTrue(results not in (None, {}))
def testTranslationPymca(self):
"""Insure we can translate a workflow using pymca to a graph and
execute it"""
self.input_information.energy_unit = "eV"
graph = ows_to_esrf(self.file_pymca)
# provide input information to dict
# this should be done by changing the .ows or something like
# that but not at this level because we are already on the
# instance of graph
graph.graph.nodes["input"]["inputs"] = self.input_information.to_dict()
graph.graph.nodes["output"]["inputs"] = {"output_file": self.output_file}
self.assertTrue(isinstance(graph, TaskGraph))
self.assertEqual(len(graph.graph.nodes()), 6)
self.assertEqual(len(graph.graph.edges()), 5)
graph.run({"root_uri": tempfile.mkdtemp()})
# check the generated process file
self.assertTrue(os.path.exists(self.output_file))
with h5py.File(self.output_file, mode="r") as h5f:
self.assertTrue("scan1" in h5f)
scan1_grp = h5f["scan1"]
for process_name in (
"xas_process_1",
"xas_process_2",
"xas_process_3",
"xas_process_4",
):
self.assertTrue(process_name in scan1_grp)
process_2 = scan1_grp["xas_process_4"]
from est.core.process.pymca.ft import PyMca_ft
self.assertEqual(
h5py_read_dataset(process_2["program"]), PyMca_ft.program_name()
)
self.assertTrue("results" in process_2)
from silx.io.dictdump import h5todict
results = h5todict(h5file=self.output_file, path="scan1/xas_process_4/results")
self.assertTrue(results not in (None, {}))
class TestGraphToOws(unittest.TestCase):
"""Insure we can convert an esrf graph using est tasks can be converted
to orange widget"""
def setUp(self) -> None:
self.folder = tempfile.mkdtemp()
self.output_ows_file = os.path.join(self.folder, "output.ows")
self.pymca_workflow_dict = {
"directed": True,
"multigraph": False,
"graph": {"name": "PyMca processing"},
"nodes": [
{
"inputs": {},
"class": "est.core.process.io.ReadXasObject",
"id": "input",
},
{
"inputs": {},
"class": "est.core.process.io.DumpXasObject",
"id": "output",
},
{
"inputs": {},
"class": "est.core.process.pymca.normalization.PyMca_normalization",
"id": "normalization",
},
{
"inputs": {},
"class": "est.core.process.pymca.exafs.PyMca_exafs",
"id": "exafs",
},
{
"inputs": {},
"class": "est.core.process.pymca.k_weight.PyMca_k_weight",
"id": "k weight",
},
{
"inputs": {},
"class": "est.core.process.pymca.ft.PyMca_ft",
"id": "fourier transform",
},
],
"links": [
{
"arguments": {"xas_obj": "xas_obj"},
"source": "input",
"target": "normalization",
},
{
"arguments": {"xas_obj": "xas_obj"},
"source": "normalization",
"target": "exafs",
},
{
"arguments": {"xas_obj": "xas_obj"},
"source": "exafs",
"target": "k weight",
},
{
"arguments": {"xas_obj": "xas_obj"},
"source": "k weight",
"target": "fourier transform",
},
{
"arguments": {"xas_obj": "xas_obj"},
"source": "fourier transform",
"target": "output",
},
],
}
self.larch_workflow_dict = {
"directed": True,
"multigraph": False,
"graph": {"name": "Larch processing"},
"nodes": [
{
"inputs": {},
"class": "est.core.process.io.ReadXasObject",
"id": "input",
},
{
"inputs": {},
"class": "est.core.process.io.DumpXasObject",
"id": "output",
},
{
"inputs": {},
"class": "est.core.process.larch.pre_edge.Larch_pre_edge",
"id": "pre edge",
},
{
"inputs": {},
"class": "est.core.process.larch.autobk.Larch_autobk",
"id": "autobk",
},
{
"inputs": {},
"class": "est.core.process.larch.xftf.Larch_xftf",
"id": "xftf",
},
],
"links": [
{
"arguments": {"xas_obj": "xas_obj"},
"source": "input",
"target": "pre edge",
},
{
"arguments": {"xas_obj": "xas_obj"},
"source": "pre edge",
"target": "autobk",
},
{
"arguments": {"xas_obj": "xas_obj"},
"source": "autobk",
"target": "xftf",
},
{
"arguments": {"xas_obj": "xas_obj"},
"source": "xftf",
"target": "output",
},
],
}
def tearDown(self) -> None:
shutil.rmtree(self.folder)
def testTranslation(self):
for graph in (self.larch_workflow_dict, self.pymca_workflow_dict):
with self.subTest(graph=graph):
esrf_to_ows(graph, self.output_ows_file)
self.assertTrue(os.path.exists(self.output_ows_file))
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment