import os
import h5py
import numpy as np
import pytest
import silx.resources
from ewoksorange.tests.utils import execute_task
from Orange.data import StringVariable
from ewoksxas.converters.orange import Converter
from ewoksxas.tasks.read_scans import ReadScans
[docs]
def test_read_scans():
"""Test reading a single scan with one counter from HDF5."""
file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
assert os.path.exists(file_path)
# Create input table with one scan.
converter = Converter()
converter.add_meta("Filename", [file_path], var_type=StringVariable)
converter.add_meta("Scan Name", ["1.1"], var_type=StringVariable)
input_table = converter.to_table()
inputs = {
"Data": input_table,
"x": "instrument/energy_enc/data",
"counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
"metadata": [
{
"name": "Emono",
"path": "instrument/positioners/Emono",
}
],
}
outputs = execute_task(ReadScans, inputs)
data = outputs["Data"]
assert len(data) == 1
assert len(data.domain.attributes) > 0
assert "Filename" in [m.name for m in data.domain.metas]
assert "Scan Name" in [m.name for m in data.domain.metas]
assert "Counter" in [m.name for m in data.domain.metas]
assert "Emono" in [m.name for m in data.domain.metas]
# Check Counter value.
counter_idx = [m.name for m in data.domain.metas].index("Counter")
assert data.metas[0, counter_idx] == "mu_trans"
# Check Emono value (scalar in HDF5).
with h5py.File(file_path, "r") as h5:
expected_emono = h5["1.1/instrument/positioners/Emono"][()]
emono_idx = [m.name for m in data.domain.metas].index("Emono")
assert data.metas[0, emono_idx] == expected_emono
[docs]
def test_read_scans_from_table():
"""Test reading multiple scans with one counter from HDF5."""
file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
converter = Converter()
converter.add_meta("Filename", [file_path, file_path], var_type=StringVariable)
converter.add_meta("Scan Name", ["1.1", "2.1"], var_type=StringVariable)
input_table = converter.to_table()
inputs = {
"Data": input_table,
"x": "instrument/energy_enc/data",
"counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
"metadata": [],
}
outputs = execute_task(ReadScans, inputs)
data = outputs["Data"]
assert len(data) == 2
assert "Filename" in [m.name for m in data.domain.metas]
assert "Scan Name" in [m.name for m in data.domain.metas]
assert "Counter" in [m.name for m in data.domain.metas]
[docs]
def test_read_scans_multiple_counters():
"""Test reading scans with multiple counters creates row per scan*counter."""
file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
converter = Converter()
converter.add_meta("Filename", [file_path], var_type=StringVariable)
converter.add_meta("Scan Name", ["1.1"], var_type=StringVariable)
input_table = converter.to_table()
inputs = {
"Data": input_table,
"x": "instrument/energy_enc/data",
"counters": [
{"name": "mu_trans", "path": "measurement/mu_trans"},
{"name": "mu_ref", "path": "measurement/mu_ref"},
],
"metadata": [],
}
outputs = execute_task(ReadScans, inputs)
data = outputs["Data"]
# 1 scan * 2 counters = 2 rows.
assert len(data) == 2
counter_idx = [m.name for m in data.domain.metas].index("Counter")
counter_names = [data.metas[i, counter_idx] for i in range(len(data))]
assert "mu_trans" in counter_names
assert "mu_ref" in counter_names
[docs]
def test_read_scans_multiple_scans_multiple_counters():
"""Test reading multiple scans with multiple counters."""
file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
converter = Converter()
converter.add_meta("Filename", [file_path, file_path], var_type=StringVariable)
converter.add_meta("Scan Name", ["1.1", "2.1"], var_type=StringVariable)
input_table = converter.to_table()
inputs = {
"Data": input_table,
"x": "instrument/energy_enc/data",
"counters": [
{"name": "mu_trans", "path": "measurement/mu_trans"},
{"name": "mu_ref", "path": "measurement/mu_ref"},
],
"metadata": [],
}
outputs = execute_task(ReadScans, inputs)
data = outputs["Data"]
# 2 scans * 2 counters = 4 rows.
assert len(data) == 4
[docs]
def test_read_scans_custom_grid():
"""Test reading scans with a custom X grid specification."""
file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
converter = Converter()
converter.add_meta("Filename", [file_path], var_type=StringVariable)
converter.add_meta("Scan Name", ["1.1"], var_type=StringVariable)
input_table = converter.to_table()
# Custom grid: 7000 to 7100 with 11 points.
inputs = {
"Data": input_table,
"x": "instrument/energy_enc/data",
"counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
"metadata": [],
"x_interp_grid": [7000.0, 7100.0, 11],
}
outputs = execute_task(ReadScans, inputs)
data = outputs["Data"]
# Verify number of points (features) corresponds to npoints.
assert len(data.domain.attributes) == 11
# Verify X values (features) are as expected.
expected_x = np.linspace(7000.0, 7100.0, 11)
feature_names = [attr.name for attr in data.domain.attributes]
# Orange features are usually named by index or value, let's check the values.
# Our Converter.add_features uses the X values as names if possible or serial names.
# Let's just check the data length for now and if we can extract names.
for i, x_val in enumerate(expected_x):
assert float(feature_names[i]) == x_val