DMSC Integration Testing

Last updated: September 15, 2025 23:36:07

Test: nexusfiles-scipp|estia|estia_read_component_position|source_position

None

None

None

None

None

None

None

None

None

None

None

None

None

None

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7fc6bf4b7560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f452e5bb560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f1452a0cf40>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f0e6466b560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f37ef23f560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f0a13203560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f4e75214f40>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f3393a7f560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7fa6ce557560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f2a3983b560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f4809af0f40>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7fd28b88f560>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7fa732a8f740>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7fe98a28f4c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f7ae102f4c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f47a88dcf40>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f62660bb4c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f029a080f40>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f4e034142c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f5a832082c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f2592dc42c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f26371dc2c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7efe57fac2c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7feecb2382c0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7fbf580200e0>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f649af04220>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7fc3df0a8040>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f199a710040>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f5a0fad0040>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f9e994b4040>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7f7cd2440040>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'estia', coda_proposal_id = '977695'
pathfinder = . at 0x7feda3408040>

@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-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:46:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/testing/nexusfiles/setup_fixtures.py:34: in
"manual": lambda *args, **kwargs: coda.get_latest_nexus_paths(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'ESTIA', 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 ESTIA

.tox/nexusfiles-scipp-estia/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

None

None

None