instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00084400.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
Test: nexusfiles-scipp|loki|load_raw_file|
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00084260.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00084127.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00083998.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki', coda_proposal_id = '999999'
pathfinder =
@pytest.fixture(scope="session")
def coda_nexus_file_path(
instrument: str, coda_proposal_id: str, pathfinder: Callable[..., Path]
) -> Path:
> return pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
instrument = 'LoKI', proposal_id = '999999', n = 1, inspect = 20
def get_latest_nexus_paths(
instrument: str, proposal_id: str | None, n: int = 1, inspect: int = 20
) -> list[Path]:
"""
Get the latest file paths manually by sorting the files in the directory.
Parameters
----------
instrument:
The instrument name.
proposal_id:
SciCat proposal ID for CODA.
If ``None``, the environment variable ``OVERRIDE_RAW_DATA_DIR`` must be set.
n:
The number of files to return.
inspect:
The number of files to inspect before returning.
Returns
-------
:
Paths to the n latest files for the given instrument and proposal.
"""
instrument = NEXUS_INSTRUMENT_NAME[instrument]
path = _coda_raw_data_path(proposal_id)
files_found = []
logging.info(
"Scanning the %d latest files in %s for instrument %s",
inspect,
path,
instrument,
)
recent_files = sorted(path.glob("*.hdf"))[-inspect::][::-1]
logging.info("Recent files: %s", recent_files)
for file in recent_files:
logging.info("Inspecting file: %s", file)
try:
with snx.File(file, "r", locking=False) as f:
instrument_in_file = f["/entry/instrument/name"][()]
if instrument_in_file == instrument:
logging.info(
"File %s matches requested instrument %s. Adding to list. "
"Last modified: %s",
file,
instrument,
datetime.fromtimestamp(
os.stat(file).st_mtime, tz=local_timezone()
),
)
files_found.append(file)
if len(files_found) == n:
logging.info("Found %d files. Returning.", n)
return files_found
else:
logging.info(
"File instrument is %s, but %s was requested. Skipping.",
instrument_in_file,
instrument,
)
except OSError as e:
logging.warning(
"File %s could not be opened. Reason: %s",
file,
e,
)
continue
logging.error("No files found for instrument %s", instrument)
> raise RuntimeError(f"No files found for instrument {instrument}")
E RuntimeError: No files found for instrument LoKI
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00083739.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00083606.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00083473.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00083347.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00083214.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00083081.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00082955.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/999999/raw/999999_00082822.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
@pytest.fixture(scope="session")
def coda_nexus_file_path(
instrument: str, coda_proposal_id: str, pathfinder: Callable[..., Path]
) -> Path:
> return pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
instrument = 'LoKI', proposal_id = '977695', n = 1, inspect = 20
def get_latest_nexus_paths(
instrument: str, proposal_id: str | None, n: int = 1, inspect: int = 20
) -> list[Path]:
"""
Get the latest file paths manually by sorting the files in the directory.
Parameters
----------
instrument:
The instrument name.
proposal_id:
SciCat proposal ID for CODA.
If ``None``, the environment variable ``OVERRIDE_RAW_DATA_DIR`` must be set.
n:
The number of files to return.
inspect:
The number of files to inspect before returning.
Returns
-------
:
Paths to the n latest files for the given instrument and proposal.
"""
instrument = NEXUS_INSTRUMENT_NAME[instrument]
path = _coda_raw_data_path(proposal_id)
files_found = []
logging.info(
"Scanning the %d latest files in %s for instrument %s",
inspect,
path,
instrument,
)
recent_files = sorted(path.glob("*.hdf"))[-inspect::][::-1]
logging.info("Recent files: %s", recent_files)
for file in recent_files:
logging.info("Inspecting file: %s", file)
try:
with snx.File(file, "r", locking=False) as f:
instrument_in_file = f["/entry/instrument/name"][()]
if instrument_in_file == instrument:
logging.info(
"File %s matches requested instrument %s. Adding to list. "
"Last modified: %s",
file,
instrument,
datetime.fromtimestamp(
os.stat(file).st_mtime, tz=local_timezone()
),
)
files_found.append(file)
if len(files_found) == n:
logging.info("Found %d files. Returning.", n)
return files_found
else:
logging.info(
"File instrument is %s, but %s was requested. Skipping.",
instrument_in_file,
instrument,
)
except OSError as e:
logging.warning(
"File %s could not be opened. Reason: %s",
file,
e,
)
continue
logging.error("No files found for instrument %s", instrument)
> raise RuntimeError(f"No files found for instrument {instrument}")
E RuntimeError: No files found for instrument LoKI
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00082359.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00082359.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00082293.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00082173.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00082047.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00081909.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00081639.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00081519.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00081393.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00081261.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00081129.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00081009.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00080871.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00080739.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00080601.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00080451.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00080313.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
@pytest.fixture(scope="session")
def coda_nexus_file_path(
instrument: str, coda_proposal_id: str, pathfinder: Callable[..., Path]
) -> Path:
> return pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
instrument = 'LoKI', proposal_id = '977695', n = 1, inspect = 20
def get_latest_nexus_paths(
instrument: str, proposal_id: str | None, n: int = 1, inspect: int = 20
) -> list[Path]:
"""
Get the latest file paths manually by sorting the files in the directory.
Parameters
----------
instrument:
The instrument name.
proposal_id:
SciCat proposal ID for CODA.
If ``None``, the environment variable ``OVERRIDE_RAW_DATA_DIR`` must be set.
n:
The number of files to return.
inspect:
The number of files to inspect before returning.
Returns
-------
:
Paths to the n latest files for the given instrument and proposal.
"""
instrument = NEXUS_INSTRUMENT_NAME[instrument]
path = _coda_raw_data_path(proposal_id)
files_found = []
logging.info(
"Scanning the %d latest files in %s for instrument %s",
inspect,
path,
instrument,
)
recent_files = sorted(path.glob("*.hdf"))[-inspect::][::-1]
logging.info("Recent files: %s", recent_files)
for file in recent_files:
logging.info("Inspecting file: %s", file)
try:
with snx.File(file, "r", locking=False) as f:
instrument_in_file = f["/entry/instrument/name"][()]
if instrument_in_file == instrument:
logging.info(
"File %s matches requested instrument %s. Adding to list. "
"Last modified: %s",
file,
instrument,
datetime.fromtimestamp(
os.stat(file).st_mtime, tz=local_timezone()
),
)
files_found.append(file)
if len(files_found) == n:
logging.info("Found %d files. Returning.", n)
return files_found
else:
logging.info(
"File instrument is %s, but %s was requested. Skipping.",
instrument_in_file,
instrument,
)
except OSError as e:
logging.warning(
"File %s could not be opened. Reason: %s",
file,
e,
)
continue
logging.error("No files found for instrument %s", instrument)
> raise RuntimeError(f"No files found for instrument {instrument}")
E RuntimeError: No files found for instrument LoKI
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00080044.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00079883.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00079743.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00079610.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00079470.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00079330.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00079204.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00079064.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00078917.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00078770.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00078630.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00078462.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00078350.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00078210.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki'
coda_nexus_file_path = PosixPath('/ess/data/coda/2025/977695/raw/977695_00077892.hdf')
def test_load_raw_file(instrument: str, coda_nexus_file_path: Path) -> None:
# Catch the warnings while loading the file to avoid the test from stopping after
# the first warning.
# Note that we need action="default" to make sure that the warnings are not
# promoted as errors.
with warnings.catch_warnings(action="default", record=True) as recorded_warnings:
with sx.File(coda_nexus_file_path, "r", locking=False) as f:
dg = f[()]
# Log the warnings to the test output.
for w in recorded_warnings:
logging.warning("%s: %s", w.category.__name__, w.message)
assert dg["entry"]["instrument"]["name"] == NEXUS_INSTRUMENT_NAME[instrument]
# Re-issue the warnings. We do this instead of asserting that the list of warnings
# is empty to make sure that the filtering of warnings in the pyproject.toml is
# working correctly.
for w in recorded_warnings:
> warnings.warn(
message=w.message,
category=w.category,
source=w.source,
stacklevel=1,
)
E UserWarning: Failed to load /entry/instrument/beam_monitor_2 as NXmonitor: Could not determine signal field or dimensions. Falling back to loading HDF5 group children as scipp.DataGroup.
tests/nexusfiles-scipp/fileloading_test.py:32: UserWarning
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
@pytest.fixture(scope="session")
def coda_nexus_file_path(
instrument: str, coda_proposal_id: str, pathfinder: Callable[..., Path]
) -> Path:
> return pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
instrument = 'LoKI', proposal_id = '977695', n = 1, inspect = 20
def get_latest_nexus_paths(
instrument: str, proposal_id: str | None, n: int = 1, inspect: int = 20
) -> list[Path]:
"""
Get the latest file paths manually by sorting the files in the directory.
Parameters
----------
instrument:
The instrument name.
proposal_id:
SciCat proposal ID for CODA.
If ``None``, the environment variable ``OVERRIDE_RAW_DATA_DIR`` must be set.
n:
The number of files to return.
inspect:
The number of files to inspect before returning.
Returns
-------
:
Paths to the n latest files for the given instrument and proposal.
"""
instrument = NEXUS_INSTRUMENT_NAME[instrument]
path = _coda_raw_data_path(proposal_id)
files_found = []
logging.info(
"Scanning the %d latest files in %s for instrument %s",
inspect,
path,
instrument,
)
recent_files = sorted(path.glob("*.hdf"))[-inspect::][::-1]
logging.info("Recent files: %s", recent_files)
for file in recent_files:
logging.info("Inspecting file: %s", file)
try:
with snx.File(file, "r", locking=False) as f:
instrument_in_file = f["/entry/instrument/name"][()]
if instrument_in_file == instrument:
logging.info(
"File %s matches requested instrument %s. Adding to list. "
"Last modified: %s",
file,
instrument,
datetime.fromtimestamp(
os.stat(file).st_mtime, tz=local_timezone()
),
)
files_found.append(file)
if len(files_found) == n:
logging.info("Found %d files. Returning.", n)
return files_found
else:
logging.info(
"File instrument is %s, but %s was requested. Skipping.",
instrument_in_file,
instrument,
)
except OSError as e:
logging.warning(
"File %s could not be opened. Reason: %s",
file,
e,
)
continue
logging.error("No files found for instrument %s", instrument)
> raise RuntimeError(f"No files found for instrument {instrument}")
E RuntimeError: No files found for instrument LoKI
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
@pytest.fixture(scope="session")
def coda_nexus_file_path(
instrument: str, coda_proposal_id: str, pathfinder: Callable[..., Path]
) -> Path:
> return pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
instrument = 'LoKI', proposal_id = '977695', n = 1, inspect = 20
def get_latest_nexus_paths(
instrument: str, proposal_id: str | None, n: int = 1, inspect: int = 20
) -> list[Path]:
"""
Get the latest file paths manually by sorting the files in the directory.
Parameters
----------
instrument:
The instrument name.
proposal_id:
SciCat proposal ID for CODA.
If ``None``, the environment variable ``OVERRIDE_RAW_DATA_DIR`` must be set.
n:
The number of files to return.
inspect:
The number of files to inspect before returning.
Returns
-------
:
Paths to the n latest files for the given instrument and proposal.
"""
instrument = NEXUS_INSTRUMENT_NAME[instrument]
path = _coda_raw_data_path(proposal_id)
files_found = []
logging.info(
"Scanning the %d latest files in %s for instrument %s",
inspect,
path,
instrument,
)
recent_files = sorted(path.glob("*.hdf"))[-inspect::][::-1]
logging.info("Recent files: %s", recent_files)
for file in recent_files:
logging.info("Inspecting file: %s", file)
try:
with snx.File(file, "r", locking=False) as f:
instrument_in_file = f["/entry/instrument/name"][()]
if instrument_in_file == instrument:
logging.info(
"File %s matches requested instrument %s. Adding to list. "
"Last modified: %s",
file,
instrument,
datetime.fromtimestamp(
os.stat(file).st_mtime, tz=local_timezone()
),
)
files_found.append(file)
if len(files_found) == n:
logging.info("Found %d files. Returning.", n)
return files_found
else:
logging.info(
"File instrument is %s, but %s was requested. Skipping.",
instrument_in_file,
instrument,
)
except OSError as e:
logging.warning(
"File %s could not be opened. Reason: %s",
file,
e,
)
continue
logging.error("No files found for instrument %s", instrument)
> raise RuntimeError(f"No files found for instrument {instrument}")
E RuntimeError: No files found for instrument LoKI
.tox/nexusfiles-scipp-loki/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError