koa_middleware.store

Classes

CalibrationStore(orm_class[, cache_dir, ...])

A CalibrationStore is used to manage storing, and retrieving calibrations.
class koa_middleware.store.CalibrationStore(orm_class: type[CalibrationORM], cache_dir: str | None = None, local_database_filename: str | None = None, remote_database_url: str | None = None, calibrations_url: str | None = None, use_cached: bool | None = None)[source]

Bases: object

A CalibrationStore is used to manage storing, and retrieving calibrations.

calibration_in_cache(calibration: CalibrationORM) → str | None[source]

Check if the calibration file is already cached locally.

Parameters:

calibration (CalibrationORM) – The ORM instance to check.

Returns:

str | None – The local file path if the calibration is cached, otherwise None.

close()[source]

Close both databases by calling engine.dipose() on the local and remote DB.

download_calibration(calibration: CalibrationORM) → str[source]

Download the calibration from the remote URL (under development).

Parameters:

calibration (CalibrationORM) – The ORM instance representing the calibration to download.

Returns:

str – The local file path of the downloaded calibration file.

get_calibration(input, selector: CalibrationSelector, use_cached: bool | None = None, **kwargs) → tuple[CalibrationORM, str][source]

Select the best calibration based on the input data and a selection rule, download if not already cached.

Parameters:

  • input – Input data product.

  • selector (CalibrationSelector) – A CalibrationSelector instance that defines the selection rule.

  • use_cached (bool | None) – If True, return the cached calibration if available.

  • kwargs – Additional parameters to pass to Selector.select().

Returns:

  • str – The local file path of the calibration file.

  • CalibrationORM – The ORM instance and local filepath.

get_calibration_by_id(calibration_id: str) → tuple[CalibrationORM, str][source]
get_local_filepath(calibration) → str[source]

Get the local filepath of a calibration ORM object.

Parameters:

calibration (CalibrationORM) – The ORM instance.

Returns:

str – The local file path of the calibration.

get_missing_local_entries() → list[CalibrationORM][source]

Identify entries in the remote DB that are missing from the local DB, using the LAST_UPDATED field as a reference. under development

Returns:

list[CalibrationORM] – List of missing CalibrationORM objects.

init_cache(local_database_filename: str | None = None)[source]
init_remote_db(remote_database_url: str | None = None)[source]
register_local_calibration(calibration) → CalibrationORM[source]

Register a calibration that is now stored in the appropriate calibrations directory by adding it to the local SQLLite DB.

NOTE: This method is responsible for calling model.save(), so we must consider the input being a datamodel. Consider alternative approach.

Parameters:

calibration (DataModel) – The calibration object to register.

Returns:

CalibrationORM – The ORM instance representing the registered calibration.

sync_from_remote() → list[CalibrationORM][source]

Download entries present in the remote DB which are missing from the local DB based on LAST_UPDATED. under development