None
Test: chexus|loki|run_chexus|
None
None
None
instrument = 'loki', coda_proposal_id = '999999'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
> path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
tests/chexus/chexus_test.py:16:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/chexus/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/chexus/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
None
None
None
None
None
None
None
None
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
> path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
tests/chexus/chexus_test.py:16:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/chexus/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/chexus/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
None
None
None
None
None
None
None
None
None
None
None
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
report = chexus.report(results=results)
logging.info(report)
logging.info(chexus.make_fileinfo(path))
if chexus.has_violations(results):
> raise AssertionError(f"Validation failed for {path}: \n" + report)
E AssertionError: Validation failed for /ess/data/coda/2025/977695/raw/977695_00081009.hdf:
E Violations
E ----------
E depends_on_missing @ /entry/instrument/thermometer_box_ts1
E depends_on_missing @ /entry/instrument/thermometer_box_ts2
E depends_on_missing @ /entry/instrument/thermometer_box_ts3
E depends_on_missing @ /entry/instrument/thermometer_box_ts4
E depends_on_missing @ /entry/instrument/thermometer_box_ts5
E depends_on_missing @ /entry/instrument/thermometer_box_ts6
E depends_on_missing @ /entry/instrument/thermometer_box_ts7
E depends_on_missing @ /entry/instrument/thermometer_box_ts8
E depends_on_target_missing @ /entry/instrument/loki_detector_0/transformations/trans_1 (depends_on target /entry/instrument/detector_carriage/target_value is not a transformation)
E
E
E Summary
E -------
E depends_on_missing: 8/56
E depends_on_target_missing: 1/119
E float_dataset_units_missing: 0/541
E group_has_units: 0/287
E index_has_units: 0/181
E mask_has_units: 0/0
E non_numeric_dataset_has_units: 0/264
E NX_class_attr_missing: 0/287
E NX_class_is_legacy: 0/287
E transformation_depends_on_missing: 0/58
E transformation_offset_units_missing: 0/0
E units_invalid: 0/1096
E NXlog_has_value: 0/124
E detector_numbers are not unique: 0/9
E event_id is not subset of associated detector_numbers: 0/9
E Shape of pixel offsets does not correspond to detector_number: 0/9
E chopper_frequency_unit_invalid: 0/4
E dataset_units_check: 0/1096
E transformation_offset_units_invalid: 0/0
E transformation_value_units_invalid: 0/59
E
E Total: 9/4486
tests/chexus/chexus_test.py:25: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
report = chexus.report(results=results)
logging.info(report)
logging.info(chexus.make_fileinfo(path))
if chexus.has_violations(results):
> raise AssertionError(f"Validation failed for {path}: \n" + report)
E AssertionError: Validation failed for /ess/data/coda/2025/977695/raw/977695_00080871.hdf:
E Violations
E ----------
E depends_on_missing @ /entry/instrument/thermometer_box_ts1
E depends_on_missing @ /entry/instrument/thermometer_box_ts2
E depends_on_missing @ /entry/instrument/thermometer_box_ts3
E depends_on_missing @ /entry/instrument/thermometer_box_ts4
E depends_on_missing @ /entry/instrument/thermometer_box_ts5
E depends_on_missing @ /entry/instrument/thermometer_box_ts6
E depends_on_missing @ /entry/instrument/thermometer_box_ts7
E depends_on_missing @ /entry/instrument/thermometer_box_ts8
E
E
E Summary
E -------
E depends_on_missing: 8/54
E depends_on_target_missing: 0/109
E float_dataset_units_missing: 0/512
E group_has_units: 0/271
E index_has_units: 0/175
E mask_has_units: 0/0
E non_numeric_dataset_has_units: 0/250
E NX_class_attr_missing: 0/271
E NX_class_is_legacy: 0/271
E transformation_depends_on_missing: 0/53
E transformation_offset_units_missing: 0/0
E units_invalid: 0/1043
E NXlog_has_value: 0/118
E detector_numbers are not unique: 0/9
E event_id is not subset of associated detector_numbers: 0/9
E Shape of pixel offsets does not correspond to detector_number: 0/9
E chopper_frequency_unit_invalid: 0/4
E dataset_units_check: 0/1043
E transformation_offset_units_invalid: 0/0
E transformation_value_units_invalid: 0/54
E
E Total: 8/4255
tests/chexus/chexus_test.py:25: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
> path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
tests/chexus/chexus_test.py:16:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/chexus/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/chexus/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
reader = chexus.read_hdf5(path, locking=False)
group = next(reader)
validators = chexus.validators.base_validators(has_scipp=True)
results = chexus.validate(group, validators=validators)
logging.info(chexus.report(results=results))
logging.info(chexus.make_fileinfo(path))
> assert not chexus.has_violations(results)
E AssertionError: assert not True
E + where True =
E + where
tests/chexus/chexus_test.py:23: AssertionError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
> path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
tests/chexus/chexus_test.py:16:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/chexus/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/chexus/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
instrument = 'loki', coda_proposal_id = '977695'
pathfinder =
def test_run_chexus(
instrument: str,
coda_proposal_id: str | None,
pathfinder: Callable[..., Path],
) -> None:
> path = pathfinder(proposal_id=coda_proposal_id, instrument=instrument)
tests/chexus/chexus_test.py:16:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/chexus/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/chexus/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError