evokit.watch package

Submodules

evokit.watch.memory.PYMPLER_ASIZEOF_RECURSION_LIMIT: int = 5

Recursion limit of pympler.asizeof.asizeof.

class evokit.watch.memory.MemoryWatcherMetric

Bases: Enum

Metrics that can be measured by MemoryWatcher.

psutil_rss = 1

psutil_vms = 2

psutil_rss_plus_children = 3

psutil_vms_plus_children = 4

pympler_asizeof_algorithm = 5

guppy3_domisize_algorithm = 6

tracemalloc_total_current = 7

tracemalloc_total_peak = 8

tracemalloc_snapshot = 9

class evokit.watch.memory.MemoryWatcher

Bases: Watcher[C, dict[MemoryWatcherMetric, int | Snapshot]]

A highly evolved Watcher with the ability to explore entire memory footprints. Returns a

Memory profiling is tricky. For example, the tracemalloc module is only able to track memory allocations made inside the Python iterator and therefore does not work well with PyTorch tensors, NumPy arrays, and Cython modules.

To remedy this, this watcher supports several metrics (see MemoryWatcher.Metric). Only those given to MemoryWatcher.__init__() will be measured and reported.

Also note that, because this watcher is attached to an algorithm, it is also part of the algorithm’s memory footprint.

supported_metrics: set[MemoryWatcherMetric] = {MemoryWatcherMetric.guppy3_domisize_algorithm, MemoryWatcherMetric.psutil_rss, MemoryWatcherMetric.psutil_rss_plus_children, MemoryWatcherMetric.psutil_vms, MemoryWatcherMetric.psutil_vms_plus_children, MemoryWatcherMetric.pympler_asizeof_algorithm, MemoryWatcherMetric.tracemalloc_snapshot, MemoryWatcherMetric.tracemalloc_total_current, MemoryWatcherMetric.tracemalloc_total_peak}

metric_to_measure: dict[MemoryWatcherMetric, Callable[[Algorithm, MemoryWatcher], int | tuple[int, int] | Snapshot]] = {MemoryWatcherMetric.guppy3_domisize_algorithm: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.psutil_rss: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.psutil_rss_plus_children: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.psutil_vms: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.psutil_vms_plus_children: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.pympler_asizeof_algorithm: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.tracemalloc_snapshot: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.tracemalloc_total_current: <function MemoryWatcher.<lambda>>, MemoryWatcherMetric.tracemalloc_total_peak: <function MemoryWatcher.<lambda>>}

Map from each MemoryWatcherMetric to a handler.

__init__(events: Container[str], metrics: Iterable[MemoryWatcherMetric], stride: int = 1, watch_post_step: bool = False)

Parameters:

• events – See Watcher.

• metrics – Metrics to measure and report.

• handler – See Watcher.

• stride – See Watcher.

• watch_automatic_events – See Watcher.

Effect:

Cause tracemalloc to start tracing memory allocations. Call MemoryWatcher.close() to stop.

metrics: Iterable[MemoryWatcherMetric]

Metrics that are measured and reported by this watcher.

subscribe(subject: C) → None

close()

Stop tracemalloc from tracing memory allocation.

class evokit.watch.memory.AttributeMemoryWatcher

Bases: Watcher[C, dict[str, int]]

A MemoryWatcher that inspects select attributes.

The initialiser accepts a list of attribute names. For each name, that attribute of the Watcher.subject is measured with asizeof (from Pympler) and IdentitySet.domisize (from Guppy3).

__init__(events: Container[str], attributes: Iterable[str], stride: int = 1, watch_post_step: bool = False)

Parameters:

• events – See Watcher.

• attributes – Names of attributes to measure and report.

• handler – See Watcher.

• stride – See Watcher.

• watch_automatic_events – See Watcher.

attributes

A collection of Guppy3 IsoSets. Initialised by subscribe().

subscribe(subject: C) → None

class evokit.watch.visual.PrintableRecord

Bases: NamedTuple

PrintableRecord(time, event, value)

time: float

Alias for field number 0

event: str

Alias for field number 1

value: tuple[float, ...]

Alias for field number 2

evokit.watch.visual.plot(records: Sequence[WatcherRecord[tuple[float, ...]]] | Sequence[WatcherRecord[float]], show_generation: bool = False, use_line: bool = False, show_legend: bool = True, axes: Axes | None = None, *args: Any, **kwargs: Any)

Plot a sequence of WatcherRecord`s. Plot :attr:`WatcherRecord.value against WatcherRecord.time. Also set the X axis label.

Parameters:

• records – Sequence of records. Each WatcherRecord.value must only hold either float or a 1-tuple of type tuple[float].

• show_generation – If True, then also plot values collected at "STEP_BEGIN" and "POST_STEP" as bigger (s=50), special (marker="*") markers. Otherwise, plot them as any other values.

• use_line – If True, then plot a line plot. Otherwise, plot a scatter graph.

• args – Passed to matplotlib.plot().

• kwargs – Passed to matplotlib.plot().

Effects:

Plot to the current Matplotlib Axes.

Note

The parameter use_line is provided for convenience. Since some values might be nan, plotting and connecting only available data points could produce misleading plots.

evokit.watch.visual.plot_dict(records: Sequence[WatcherRecord[dict[Any, float]]], keys: Collection[Any] | None = None, show_generation: bool = False, show_legend: bool = True, use_line: bool = False, axes: Axes | None = None, *args: Any, **kwargs: Any)

class evokit.watch.watcher.WatcherRecord

Bases: Generic[T]

A record collected by an Watcher from an Algorithm. Also records the generation count and time of collection.

event: str

Event that triggered the handler.

generation: int

Generation count when the event event occurs.

value: T

Data collected in generation after event.

time: float

Time (by time.process_time()) when the event event occurs.

__init__(event: str, generation: int, value: T, time: float) → None

class evokit.watch.watcher.Watcher

Bases: Generic[C, T], Sequence[WatcherRecord[T]]

Observes and collect data from a running Algorithm.

The Watcher should be registered to an Algorithm, which then becomes the watcher’s subject. When an event fires in the subject, if that event is in the watcher’s events, then the handler will be called with the subject as argument. Results are collected as a sequence of WatcherRecords.

Call Algorithm.register() to register an Watcher to a Algorithm. Call report() to retrieve collected records.

For type checking purposes, the Watcher has two type parameter C and T. C is the type of the observed Algorithm; T is the type of .value in the reported WatcherRecord.

Tutorial: Collect and Plotting Statistics.

MANUAL_EVENT: str = 'MANUAL_TRIGGER'

__init__(events: Container[str], handler: Callable[[C], T], stride: int = 1, *, watch_post_step: bool = False, timer: Callable[[], float] = <built-in function process_time>)

Parameters:

• events – Events that trigger the handler.

• handler – Callable that takes the attached algorithm as input.

• stride – Collection interval. Only stride ^th event triggers handler.

• watch_post_step – If True, also watch the POST_STEP event. This event fires automatically after Algorithm.step().

subject: C | None

The attached Algorithm.

unsubscribe() → None

Unsubscribe this watcher from the subject. Reset this watcher to be registered with another algorithm.

Effect:

Reset the subject

Reset the accumulated stride

Reset all collected records

Remove self from subject‘s

Algorithm.watchers

update(event: str) → None

When the subject calls Algorithm.update(), the subject calls this method on every watcher registered to it.

When an event matches a key in handlers, call the corresponding value with the subject as argument. Store the result in records.

To trigger collections, call manual_update() instead of this method.

Raises:

RuntimeError – If no Algorithm is attached.

force_update(event: str = 'MANUAL_EVENT') → None

Manually trigger update(), bypassing all checks of whether event is observed.

Parameters:

event – Name of an event. Defaults to MANUAL_EVENT.

Raises:

RuntimeError – If no Algorithm is attached.

report() → list[WatcherRecord[T]]

Report collected records.

is_registered() → bool

Return if this watcher is observing an Algorithm.

purge() → None

Remove all collected records.

Effect:

Reset collected records to an empty list.

evokit.watch.watchers.create_fitness_watcher(events: list[str], stride: int = 1, *, watch_post_step: bool = False, timer: ~typing.Callable[[], float] = <built-in function process_time>) → Watcher[HomogeneousAlgorithm[Individual[Any]], tuple[float, ...]]

Return an Watcher that collects the best Individual.fitness of a HomogeneousAlgorithm.population.

See Watcher.__init__() for parameters.

evokit.watch.watchers.create_size_watcher(events: list[str], stride: int = 1, *, watch_post_step: bool = False, timer: ~typing.Callable[[], float] = <built-in function process_time>) → Watcher[HomogeneousAlgorithm[Individual[Any]], int]

Return an Watcher that collects the size of the population.

See Watcher.__init__() for parameters.

evokit.watch.watchers.create_cpu_watcher(events: list[str], stride: int = 1, *, watch_post_step: bool = False, timer: ~typing.Callable[[], float] = <built-in function process_time>) → Watcher[HomogeneousAlgorithm[Individual[Any]], float]

Return an Watcher that collects the CPU time of the process.

See Watcher.__init__() for parameters.

evokit.watch.watchers.create_rss_watcher(events: list[str], stride: int = 1, *, watch_post_step: bool = False, timer: ~typing.Callable[[], float] = <built-in function process_time>) → Watcher[HomogeneousAlgorithm[Individual[Any]], float]

Return an Watcher that collects the memory usage (RSS) of an algorithm.

See Watcher.__init__() for parameters.

Module contents