Skip to content

Base

Base classes for calculation processors that track their own performance metrics.

TrackableProcessor()

Bases: EntityProcessor

Mixin for entity processors that track calculation metrics.

Provides set_metric / get_metric / get_all_metrics so processors can record timing and other data that flows into benchmark results and solution metadata.

Initialize the processor with metadata storage.

Source code in src/oracle_paper/data/calculator/base.py 
18
19
20
21
def __init__(self):
	"""Initialize the processor with metadata storage."""
	super().__init__()
	self._processor_metadata: dict[str, Any] = {}

get_all_metrics()

Get all recorded metrics.

Source code in src/oracle_paper/data/calculator/base.py 
31
32
33
def get_all_metrics(self) -> dict[str, Any]:
	"""Get all recorded metrics."""
	return self._processor_metadata.copy()

get_metric(key)

Retrieve a metric value.

Source code in src/oracle_paper/data/calculator/base.py 
27
28
29
def get_metric(self, key: str) -> Any:
	"""Retrieve a metric value."""
	return self._processor_metadata.get(key)

set_metric(key, value)

Store a metric value.

Source code in src/oracle_paper/data/calculator/base.py 
23
24
25
def set_metric(self, key: str, value: Any) -> None:
	"""Store a metric value."""
	self._processor_metadata[key] = value

track_metric(metric_name)

Decorator to automatically track calculation time for a method.

Parameters:

Name Type Description Default
metric_name str

The name to store the metric under (e.g., "lagrange_time")

 required
Example

@track_metric("lagrange_calculation_time") def process(self, dataset): # ... calculation code ...

The decorated method must be part of a TrackableProcessor instance. The elapsed time will be stored via set_metric(metric_name, elapsed_time).

Source code in src/oracle_paper/data/calculator/base.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def track_metric(metric_name: str) -> Callable:
	"""
	Decorator to automatically track calculation time for a method.

	Args:
		metric_name: The name to store the metric under (e.g., "lagrange_time")

	Example:
		@track_metric("lagrange_calculation_time")
		def process(self, dataset):
			# ... calculation code ...

	The decorated method must be part of a TrackableProcessor instance.
	The elapsed time will be stored via set_metric(metric_name, elapsed_time).
	"""

	def decorator(func: Callable) -> Callable:
		@wraps(func)
		def wrapper(self, *args, **kwargs) -> Any:
			start_time = time.time()
			try:
				result = func(self, *args, **kwargs)
				return result
			finally:
				elapsed_time = time.time() - start_time
				# Only store if the instance supports metrics (TrackableProcessor)
				if hasattr(self, "set_metric"):
					self.set_metric(metric_name, elapsed_time)

		return wrapper

	return decorator