punchbowl.auto.control.util#

Attributes#

DEFAULT_SCALING

Functions#

get_database_session([get_engine, engine_kwargs, ...])

Sets up a session to connect to the MariaDB punchpipe database

update_file_state(session, file_id, new_state)

load_pipeline_configuration(→ dict)

load_quicklook_scaling(level, product, obscode, path, ...)

write_file(→ None)

_write_quicklook(pipeline_config, ...)

match_data_with_file_db_entry(data, file_db_entry_list)

get_files_in_time_window(...)

batched(iterable, n)

group_files_by_time(...)

replace_version(pattern, replacement, string)

replace_file_version_in_metadata(path, old_pattern, ...)

Module Contents#

punchbowl.auto.control.util.DEFAULT_SCALING = (5e-13, 5e-11)#
punchbowl.auto.control.util.get_database_session(get_engine=False, engine_kwargs={}, session_kwargs={}, credential_name: str = None)#

Sets up a session to connect to the MariaDB punchpipe database

Parameters:

  • get_engine (bool) – if true returns engine as second parameter

  • engine_kwargs (dict) – kwargs for the SqlAlchemy engine creation

  • session_kwargs (dict) – kwargs for the SqlAlchemy session creation

  • credential_name (str) – keyword for the credential block name in Prefect

Return type:

session or (session, engine) tuple depending on get_engine

punchbowl.auto.control.util.update_file_state(session, file_id, new_state)#
punchbowl.auto.control.util.load_pipeline_configuration(path: str = None) → dict#
punchbowl.auto.control.util.load_quicklook_scaling(level: str = None, product: str = None, obscode: str = None, path: str = None)#
punchbowl.auto.control.util.write_file(data: ndcube.NDCube, corresponding_file_db_entry, pipeline_config) → None#
punchbowl.auto.control.util._write_quicklook(pipeline_config: dict, corresponding_file_db_entry: punchbowl.auto.control.db.File, data: ndcube.NDCube)#
punchbowl.auto.control.util.match_data_with_file_db_entry(data: ndcube.NDCube, file_db_entry_list)#
punchbowl.auto.control.util.get_files_in_time_window(level: str, file_type: str, obs_code: str, start_time: datetime.datetime, end_time: datetime.datetime, session: sqlalchemy.orm.Session | None) → list[punchbowl.auto.control.db.File]#
punchbowl.auto.control.util.batched(iterable, n)#
punchbowl.auto.control.util.group_files_by_time(files: list[punchbowl.auto.control.db.File], max_duration_seconds: float = inf, max_per_group: int = inf) → list[list[punchbowl.auto.control.db.File]]#
punchbowl.auto.control.util.replace_version(pattern, replacement, string)#
punchbowl.auto.control.util.replace_file_version_in_metadata(path, old_pattern, new_version)#