lenskit.logging#

Logging, progress, and resource records.

Functions

trace(logger, *args, **kwargs)

Emit a trace-level message, if LensKit tracing is enabled.

class lenskit.logging.LoggingConfig#

Bases: object

Configuration for LensKit logging.

This class is intended as a convenience for LensKit applications to set up a useful logging and progress reporting configuration; if unconfigured, LensKit will emit its logging messages directly to structlog and/or logging, which you can configure in any way you wish.

set_verbose(verbose=True)#

Enable verbose logging.

Note

It is better to only call this method if your application’s verbose option is provided, rather than passing your verbose option to it, to allow the LK_LOG_LEVEL environment variable to apply in the absence of a configuration option.

Parameters:

verbose (bool | int) – The level of verbosity. Values of True or 1 turn on DEBUG-level logs, and 2 or greater turns on tracing.

log_file(path, level=None)#

Configure a log file.

Parameters:
apply()#

Apply the configuration.

class lenskit.logging.Progress(*args, **kwargs)#

Bases: object

Base class for progress reporting. The default implementations do nothing.

Parameters:
  • args (Any)

  • kwargs (Any)

update(advance=1, **kwargs)#

Update the progress bar.

Parameters:
finish()#

Finish and clean up this progress bar. If the progresss bar is used as a context manager, this is automatically called on context exit.

lenskit.logging.item_progress(label, total, fields=None)#

Create a progress bar for distinct, counted items.

Parameters:
  • label (str) – The progress bar label.

  • total (int) – The total number of items.

  • fields (dict[str, str | None] | None) – Additional fields to report with the progress bar (such as a current loss). These are specified as a dictionary mapping field names to format strings (the pieces inside {...} in str.format()), and the values come from extra kwargs to Progress.update(); mapping to None use default str formatting.

Return type:

Progress

class lenskit.logging.Task(label, *, file=None, parent=None, reset_hwm=None, task_id=<factory>, parent_id=None, subprocess=False, status=TaskStatus.PENDING, start_time=None, finish_time=None, duration=None, cpu_time=None, peak_memory=None, peak_gpu_memory=None, subtasks=<factory>, **data)#

Bases: BaseModel

A task for logging and resource measurement.

A task may be top-level (have no parent), or it may be a subtask. By default, new tasks have the current active task as their parent. Tasks are not active until they are started (using a task as a context manager automatically does this, which is the recommended process).

The task-tracking mechanism is currently designed to support large tasks, like training a model or running batch inference; it is not yet designed for fine-grained spans like you would see in OpenTelemetry or Eliot.

Note

The notion of the “active task” does not yet support multi-threaded tasks.

Parameters:
  • file (PathLike[str] | None) – A file to save the task when it is finished.

  • parent (Task | UUID | None) – The parent task. If unspecified, uses the currently-active task.

  • reset_hwm (bool | None) – Whether to reset the system resource high-water-marks at the start of this task. Only effective on Linux, but allows for measurement of the peak memory use of this task specifically. If unspecified, it resets the HWM if there is no parent.

  • label (str)

  • task_id (UUID)

  • parent_id (UUID | None)

  • subprocess (bool)

  • status (TaskStatus)

  • start_time (float | None)

  • finish_time (float | None)

  • duration (float | None)

  • cpu_time (float | None)

  • peak_memory (int | None)

  • peak_gpu_memory (int | None)

  • subtasks (Annotated[list[Annotated[Task, SerializeAsAny()]], BeforeValidator(func=~lenskit.logging.tasks._dict_extract_values, json_schema_input_type=PydanticUndefined)])

  • data (Any)

task_id: UUID#

The task ID.

parent_id: UUID | None#

The parent task ID.

subprocess: bool#

Whether this task is a subprocess of its parent. Subprocess task CPU times are not included in the parent task times.

label: str#

Human-readable task label.

status: TaskStatus#

The task’s current status.

start_time: float | None#

The task start time (UNIX timestamp).

finish_time: float | None#

The task completion time (UNIX timestamp).

duration: float | None#

Task duration in seconds. Measured using time.perf_counter(), so it may disagree slightly with the difference in start and finish times.

cpu_time: float | None#

CPU time consumed in seconds.

peak_memory: int | None#

Peak memory usage (max RSS) in bytes. Only available on Unix; individual task peak memory use is only reliable on Linux (MacOS will report the max memory used since the process was started).

peak_gpu_memory: int | None#

Peak PyTorch GPU memory usage in bytes.

subtasks: Annotated[list[SerializeAsAny[Task]], BeforeValidator(_dict_extract_values)]#

This task’s subtasks.

static current()#

Get the currently-active task.

Return type:

Task | None

static root()#

Get the root task.

Return type:

Task | None

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)#

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:
  • self (BaseModel) – The BaseModel instance.

  • context (Any) – The context.

Return type:

None

save_to_file(path, monitor=True)#

Save this task to a file, and re-save it when finished.

Parameters:
start()#

Start the task.

finish(status=TaskStatus.FINISHED)#

Finish the task.

Parameters:

status (TaskStatus)

update()#

Update the task’s resource measurements and save the file (if one is set).

monitor_refresh()#

Refresh method called by the monitor backend.

add_subtask(task)#

Add or update a subtask.

Parameters:

task (Task)

update_resources()#

Update the resource measurements. Returns the current measurement.

This method is called by update(), with an exclusive lock held.

Return type:

ResourceMeasurement

lenskit.logging.get_logger(*args, **initial_values)#

Only calls structlog.get_logger, but has the correct type hints.

Warning

Does not check whether – or ensure that – you’ve configured structlog for standard library logging!

See standard-library for details.

Added in version 20.2.0.

Parameters:
  • args (Any)

  • initial_values (Any)

Return type:

BoundLogger

lenskit.logging.trace(logger, *args, **kwargs)#

Emit a trace-level message, if LensKit tracing is enabled. Trace-level messages are more fine-grained than debug-level messages, and you usually don’t want them.

Parameters:

Modules

config

Logging pipeline configuration.

monitor

LensKit background monitoring.

processors

LensKit logging processors and converters.

progress

resource

Measure resource consumption.

tasks

Abstraction for recording tasks.

tracing

Extended logger providing TRACE support.

worker

Support for logging from worker processes.