bhf_smarthealth.log.log_config#

This module configures Logging objects used within the pipeline logging system

Functions#

generate_log_timestamp_utc([timespec])

Generate a timestamp for 'now' in UTC zero offset

build_log_paths(→ dict[str, str])

Build a dictionary of logging output paths for each output location specified in outputs.keys() and the corresponding part-path in outputs.values()

default_log_format(→ str)

setup_global_logger(, log_path, level)

Initialises a global logger with File Handler to capture Snakemake workflow outputs.

setup_rule_log() → logging.Logger)

Initialises a logger with File Handler to capture outputs from Snakemake rules.

Module Contents#

bhf_smarthealth.log.log_config.generate_log_timestamp_utc(timespec: str = 'seconds')[source]#

Generate a timestamp for ‘now’ in UTC zero offset

Parameters:

timespec (str, optional) – The optional argument timespec specifies the number of additional terms of the time to include. Valid options are ‘auto’, ‘hours’, ‘minutes’, ‘seconds’, ‘milliseconds’ and ‘microseconds’., by default “seconds”

Returns:

String of UTC zero offset timestamp for ‘now

Return type:

str

bhf_smarthealth.log.log_config.build_log_paths(date_to_process: datetime.date, outputs: dict[str, str], run_id: str, data_type: bhf_smarthealth.models.DataStreamNames | None = None) → dict[str, str][source]#

Build a dictionary of logging output paths for each output location specified in outputs.keys() and the corresponding part-path in outputs.values()

Parameters:
  • date_to_process (date) – Date object of date being processed

  • outputs (dict[str, str]) – Dictionary containing the log OUTPUT name as keys and the corresponding path value

  • run_id (str) – String identifer of a unique pipeline run

  • data_type (DataStreamNames | None, optional) – StrEnum of data stream being processed. If specified, a “data_type” subdir will be appended to each log path. By default None

Returns:

Dictionary of log names in the form {OUTPATH}_{LOGTYPE}_LOG as keys with path to log as the value.

Return type:

dict[str, str]

bhf_smarthealth.log.log_config.default_log_format() → str[source]#
bhf_smarthealth.log.log_config.setup_global_logger(format: str = default_log_format(), log_path: pathlib.Path | str = os.getcwd(), level: int | str = logging.INFO) → None[source]#

Initialises a global logger with File Handler to capture Snakemake workflow outputs.

Parameters:
  • format (str, optional) – Format options for the logged outputs, by default default_log_format()

  • log_path (str, optional) – Path to output the log file to, by default the current working directory

  • level (int, optional) – Logging level, by default logging.INFO

bhf_smarthealth.log.log_config.setup_rule_log(log_file: pathlib.Path, level: int | str = logging.INFO, format: str = default_log_format()) → logging.Logger[source]#

Initialises a logger with File Handler to capture outputs from Snakemake rules.

Parameters:
  • log_file (Path) – Path of the file to log to

  • level (int | str, optional) – Logging level, by default logging.INFO

  • format (str, optional) – Format options for the logged outputs, by default default_log_format()

Returns:

logger – A configured Logger object with a File Handler

Return type:

logging.Logger