instrument = 'bifrost', 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)
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/999999/raw/999999_00084483.hdf:
E Violations
E ----------
E depends_on_missing @ /entry/instrument/114_sample_stack/goniometer_x_lower
E depends_on_missing @ /entry/instrument/114_sample_stack/goniometer_y_upper
E depends_on_missing @ /entry/instrument/114_sample_stack/rotation_stage
E depends_on_missing @ /entry/instrument/117_detector_tank_angle/detector_tank_angle_r0
E
E
E Summary
E -------
E depends_on_missing: 4/184
E depends_on_target_missing: 0/1222
E float_dataset_units_missing: 0/1931
E group_has_units: 0/960
E index_has_units: 0/523
E mask_has_units: 0/0
E non_numeric_dataset_has_units: 0/809
E NX_class_attr_missing: 0/960
E NX_class_is_legacy: 0/960
E transformation_depends_on_missing: 0/881
E transformation_offset_units_missing: 0/0
E units_invalid: 0/2490
E NXlog_has_value: 0/98
E detector_numbers are not unique: 0/45
E event_id is not subset of associated detector_numbers: 0/45
E Shape of pixel offsets does not correspond to detector_number: 0/45
E chopper_frequency_unit_invalid: 0/6
E dataset_units_check: 0/2490
E transformation_offset_units_invalid: 0/0
E transformation_value_units_invalid: 0/885
E
E Total: 4/14534
tests/chexus/chexus_test.py:25: AssertionError
Test: chexus|bifrost|run_chexus|
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
None
instrument = 'bifrost', 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 = 'BIFROST', 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 BIFROST
.tox/chexus/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
None
None
None
None
None
None
instrument = 'bifrost', 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 = 'BIFROST', 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") 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 BIFROST
.tox/chexus/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError
None
None