Skip to content
This repository was archived by the owner on Nov 8, 2024. It is now read-only.

Commit

Permalink
Test writing pixel data
Browse files Browse the repository at this point in the history
  • Loading branch information
@jl-wynen
jl-wynen committed Nov 4, 2024
1 parent e426d9e commit 69362e5
Show file tree
Hide file tree
Showing 2 changed files with 178 additions and 6 deletions.
121 changes: 115 additions & 6 deletions tests/horace_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
# Copyright (c) 2024 Scipp contributors (https://github.com/scipp)

import os
from dataclasses import replace
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
Expand Down Expand Up @@ -108,16 +109,16 @@ def sample() -> SqwIXSample:
def experiment_template() -> SqwIXExperiment:
return SqwIXExperiment(
run_id=0,
efix=sc.scalar(1.0, unit='meV'),
efix=sc.scalar(1.2, unit='meV'),
emode=EnergyMode.direct,
en=sc.array(dims=['energy_transfer'], values=[], unit='meV'),
psi=sc.scalar(0.0, unit='rad'),
en=sc.array(dims=['energy_transfer'], values=[-0.1, 0.3, 0.5], unit='meV'),
psi=sc.scalar(0.4, unit='rad'),
u=sc.vector([1.0, 0.0, 0.0], unit='1/angstrom'),
v=sc.vector([0.0, 1.0, 0.0], unit='1/angstrom'),
omega=sc.scalar(0.0, unit='rad'),
omega=sc.scalar(-0.01, unit='rad'),
dpsi=sc.scalar(0.0, unit='rad'),
gl=sc.scalar(0.0, unit='rad'),
gs=sc.scalar(0.0, unit='rad'),
gl=sc.scalar(1.2, unit='rad'),
gs=sc.scalar(0.6, unit='rad'),
filename='experiment1.nxspe',
filepath='/data',
)
Expand Down Expand Up @@ -252,3 +253,111 @@ def test_horace_roundtrip_experiment(
np.testing.assert_equal(loaded.gs.squeeze(), expected.gs.to(unit='rad').value)
assert loaded.filename == expected.filename
assert loaded.filepath == expected.filepath


def test_horace_roundtrip_pixels_single_experiment(
matlab: Any,
dnd_metadata: SqwDndMetadata,
null_instrument: SqwIXNullInstrument,
sample: SqwIXSample,
experiment_template: SqwIXExperiment,
tmp_path: Path,
) -> None:
path = tmp_path / "roundtrip_pixels_single_experiment.sqw"
n_pixels = 7
# Chosen numbers can all be represented in float32 to allow exact comparisons.
u1 = np.arange(n_pixels) + 0.0
u2 = np.arange(n_pixels) + 1.0
u3 = np.arange(n_pixels) + 3.0
u4 = np.arange(n_pixels) * 2
irun = np.full(n_pixels, 0)
idet = (np.arange(n_pixels) / 3).astype(int)
ien = np.full(n_pixels, 1)
values = 20 * np.arange(n_pixels)
variances = values / 2

with (
Sqw.build(path, title="Pixel test file")
.add_default_instrument(null_instrument)
.add_default_sample(sample)
.add_empty_dnd_data(dnd_metadata)
.register_pixel_data(
n_pixels=n_pixels, n_dims=4, experiments=[experiment_template]
)
.create()
) as sqw:
sqw.write_pixel_data(
np.c_[u1, u2, u3, u4, irun, idet, ien, values, variances], run=0
)

loaded = matlab.read_horace(os.fspath(path))
np.testing.assert_equal(loaded.pix.u1.squeeze(), u1)
np.testing.assert_equal(loaded.pix.u2.squeeze(), u2)
np.testing.assert_equal(loaded.pix.u3.squeeze(), u3)
np.testing.assert_equal(loaded.pix.dE.squeeze(), u4)
np.testing.assert_equal(loaded.pix.run_idx.squeeze(), irun)
np.testing.assert_equal(loaded.pix.detector_idx.squeeze(), idet)
np.testing.assert_equal(loaded.pix.energy_idx.squeeze(), ien)
np.testing.assert_equal(loaded.pix.signal.squeeze(), values)
np.testing.assert_equal(loaded.pix.variance.squeeze(), variances)


def test_horace_roundtrip_pixels_two_experiments(
matlab: Any,
dnd_metadata: SqwDndMetadata,
null_instrument: SqwIXNullInstrument,
sample: SqwIXSample,
experiment_template: SqwIXExperiment,
tmp_path: Path,
) -> None:
path = tmp_path / "roundtrip_pixels_two_experiments.sqw"
n_pixels = 7
# Chosen numbers can all be represented in float32 to allow exact comparisons.
u1 = np.arange(n_pixels) + 0.0
u2 = np.arange(n_pixels) + 1.0
u3 = np.arange(n_pixels) + 3.0
u4 = np.arange(n_pixels) * 2
irun = np.full(n_pixels, 0)
idet = (np.arange(n_pixels) / 3).astype(int)
ien = np.full(n_pixels, 1)
values = 20 * np.arange(n_pixels)
variances = values / 2

experiments = [
replace(experiment_template, run_id=0, filename="experiment_1.nxspe"),
replace(experiment_template, run_id=1, filename="experiment_2.nxspe"),
]

with (
Sqw.build(path, title="Pixel test file")
.add_default_instrument(null_instrument)
.add_default_sample(sample)
.add_empty_dnd_data(dnd_metadata)
.register_pixel_data(
n_pixels=n_pixels * len(experiments), n_dims=4, experiments=experiments
)
.create()
) as sqw:
sqw.write_pixel_data(
np.c_[u1, u2, u3, u4, irun, idet, ien, values, variances], run=0
)
sqw.write_pixel_data(
np.c_[u1, u2, u3, u4, irun, idet, ien, values, variances] + 1000, run=1
)

loaded = matlab.read_horace(os.fspath(path))
np.testing.assert_equal(loaded.pix.u1.squeeze(), np.r_[u1, u1 + 1000])
np.testing.assert_equal(loaded.pix.u2.squeeze(), np.r_[u2, u2 + 1000])
np.testing.assert_equal(loaded.pix.u3.squeeze(), np.r_[u3, u3 + 1000])
np.testing.assert_equal(loaded.pix.dE.squeeze(), np.r_[u4, u4 + 1000])
np.testing.assert_equal(loaded.pix.run_idx.squeeze(), np.r_[irun, irun + 1000])
np.testing.assert_equal(loaded.pix.detector_idx.squeeze(), np.r_[idet, idet + 1000])
np.testing.assert_equal(loaded.pix.energy_idx.squeeze(), np.r_[ien, ien + 1000])
np.testing.assert_equal(loaded.pix.signal.squeeze(), np.r_[values, values + 1000])
np.testing.assert_equal(
loaded.pix.variance.squeeze(), np.r_[variances, variances + 1000]
)


# TODO indirect
# TODO ndims
63 changes: 63 additions & 0 deletions tests/write_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,69 @@ def test_writes_expdata(
)


@pytest.mark.parametrize("byteorder", ["native", "little", "big"])
def test_writes_pixel_data(
byteorder: Literal["native", "little", "big"], buffer: _BytesBuffer | _PathBuffer
) -> None:
experiment_template = SqwIXExperiment(
run_id=-1,
efix=sc.scalar(1.2, unit='meV'),
emode=EnergyMode.direct,
en=sc.array(dims=['energy_transfer'], values=[3.0], unit='meV'),
psi=sc.scalar(1.2, unit='rad'),
u=sc.vector([0.0, 1.0, 0.0]),
v=sc.vector([1.0, 1.0, 0.0]),
omega=sc.scalar(1.4, unit='rad'),
dpsi=sc.scalar(0.0, unit='rad'),
gl=sc.scalar(3, unit='rad'),
gs=sc.scalar(-0.5, unit='rad'),
filename="",
filepath='/data',
)
experiments = [
dataclasses.replace(experiment_template, run_id=0, filename="f1"),
dataclasses.replace(experiment_template, run_id=1, filename="f2"),
]

n_pixels = 7
# Chosen numbers can all be represented in float32 to allow exact comparisons.
u1 = np.arange(n_pixels) + 0.0
u2 = np.arange(n_pixels) + 1.0
u3 = np.arange(n_pixels) + 3.0
u4 = np.arange(n_pixels) * 2
irun = np.full(n_pixels, 0)
idet = (np.arange(n_pixels) / 3).astype(int)
ien = np.full(n_pixels, 1)
values = 20 * np.arange(n_pixels)
variances = values / 2

builder = Sqw.build(buffer.get(), byteorder=byteorder)
builder = builder.register_pixel_data(
n_pixels=n_pixels * 2, n_dims=4, experiments=experiments
)
with builder.create() as sqw:
sqw.write_pixel_data(
np.c_[u1, u2, u3, u4, irun, idet, ien, values, variances], run=0
)
sqw.write_pixel_data(
np.c_[u1, u2, u3, u4, irun, idet, ien, values, variances] + 1000, run=1
)
buffer.rewind()

with Sqw.open(buffer.get()) as sqw:
loaded = sqw.read_data_block(("pix", "data_wrap"))

np.testing.assert_equal(loaded[:, 0], np.r_[u1, u1 + 1000])
np.testing.assert_equal(loaded[:, 1], np.r_[u2, u2 + 1000])
np.testing.assert_equal(loaded[:, 2], np.r_[u3, u3 + 1000])
np.testing.assert_equal(loaded[:, 3], np.r_[u4, u4 + 1000])
np.testing.assert_equal(loaded[:, 4], np.r_[irun, irun + 1000])
np.testing.assert_equal(loaded[:, 5], np.r_[idet, idet + 1000])
np.testing.assert_equal(loaded[:, 6], np.r_[ien, ien + 1000])
np.testing.assert_equal(loaded[:, 7], np.r_[values, values + 1000])
np.testing.assert_equal(loaded[:, 8], np.r_[variances, variances + 1000])


@pytest.mark.parametrize("byteorder", ["native", "little", "big"])
def test_writes_data_metadata(
byteorder: Literal["native", "little", "big"], buffer: _BytesBuffer | _PathBuffer
Expand Down

0 comments on commit 69362e5

Please sign in to comment.