import datetime
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
[docs]
@dataclass
class RunProfile:
duration: float = 0
success: bool = True
start_time: Optional[datetime.datetime] = None
end_time: Optional[datetime.datetime] = None
log_data: Dict = field(default_factory=dict)
stdout: Optional[str] = None
stderr: Optional[str] = None
def __getitem__(self, key: str) -> Any:
"""Gets an attribute by key.
:param key: Attribute name
:type key: str
:return: Attribute value
:rtype: Any
"""
return getattr(self, key)
[docs]
def asdict(self, exclude: Tuple = ()) -> Dict:
"""Converts the RunProfile to a dictionary, excluding specified keys.
:param exclude: Tuple of keys to exclude, defaults to ()
:type exclude: Tuple, optional
:return: Dictionary representation
:rtype: Dict
"""
return {k: v for k, v in asdict(self).items() if k not in exclude}
[docs]
def filtered_log_data(self, component_type: str = "encoder") -> Dict:
"""Filters log data by component type, excluding keys not relevant to the component.
:param component_type: Type of component ("encoder" or "decoder"), defaults to "encoder"
:type component_type: str, optional
:return: Filtered log data dictionary
:rtype: Dict
"""
if not self.log_data:
return {}
exclude_keys_by_type = {
"encoder": set(),
"decoder": {
"output_number_of_bytes",
"jplm_reported_bpp",
"jplm_reported_bpp_codestream",
"lambda",
},
}
exclude_keys = exclude_keys_by_type.get(component_type, set())
return {k: v for k, v in self.log_data.items() if k not in exclude_keys}
[docs]
@dataclass
class PooledMetrics:
mean: float = 0
median: float = 0
min: float = 0
max: float = 0
stddev: float = 0
cv: float = 0
def __getitem__(self, key: str) -> Any:
"""Gets an attribute by key.
:param key: Attribute name
:type key: str
:return: Attribute value
:rtype: Any
"""
return getattr(self, key)
[docs]
def asdict(self, exclude: Tuple = ()) -> Dict:
"""Converts the PooledMetrics to a dictionary, excluding specified keys.
:param exclude: Tuple of keys to exclude, defaults to ()
:type exclude: Tuple, optional
:return: Dictionary representation
:rtype: Dict
"""
return {k: v for k, v in asdict(self).items() if k not in exclude}
[docs]
@dataclass
class ExecutionMeasurements:
runs: Optional[List[RunProfile]] = None
pooled_metrics: Optional[PooledMetrics] = None
memory_metrics: Optional[PooledMetrics] = None
def __post_init__(self) -> None:
if not self.runs:
self.runs = []
[docs]
def convert_to_bytes(value: float, unit: str, multiplier: int = 1024) -> float:
"""Converts memory values to bytes based on the given unit.
:param value: Numeric value to convert
:type value: float
:param unit: Unit string (kbytes, mbytes, gbytes, tbytes, or bytes)
:type unit: str
:param multiplier: Base multiplier for conversion, defaults to 1024
:type multiplier: int, optional
:return: Value converted to bytes
:rtype: float
"""
unit = unit.lower()
if unit == "kbytes":
return value * multiplier
elif unit == "mbytes":
return value * multiplier ** 2
elif unit == "gbytes":
return value * multiplier ** 3
elif unit == "tbytes":
return value * multiplier ** 4
return value # Assume bytes by default
[docs]
def get_metrics_from_runs(runs: List[RunProfile]) -> Optional[ExecutionMeasurements]:
"""Computes pooled execution metrics from a list of run profiles.
:param runs: List of RunProfile instances
:type runs: List[RunProfile]
:return: ExecutionMeasurements with pooled metrics, or None if no successful runs
:rtype: Optional[ExecutionMeasurements]
"""
success_runs = [r for r in runs if r.success]
durations = np.array([r.duration for r in success_runs], dtype=np.float64)
memory_usages = []
for r in success_runs:
if "max_memory_usage" in r.log_data:
memory_data = r.log_data["max_memory_usage"]
value = memory_data.get("value", 0)
unit = memory_data.get("unit", "bytes")
memory_usages.append(convert_to_bytes(value, unit))
if durations.size == 0:
return None
mean = np.mean(durations)
median = np.median(durations)
min = np.min(durations)
max = np.max(durations)
stddev = np.std(durations)
cv = (stddev / mean) * 100 if mean != 0 else 0
if memory_usages:
memory_mean = np.mean(memory_usages)
memory_median = np.median(memory_usages)
memory_min = np.min(memory_usages)
memory_max = np.max(memory_usages)
memory_stddev = np.std(memory_usages)
memory_cv = (memory_stddev / memory_mean) * 100 if memory_mean != 0 else 0
memory_metrics = PooledMetrics(
mean=memory_mean, median=memory_median, min=memory_min, max=memory_max,
stddev=memory_stddev, cv=memory_cv
)
else:
memory_metrics = None
return ExecutionMeasurements(
runs=runs,
pooled_metrics=PooledMetrics(mean=mean, median=median, min=min, max=max, stddev=stddev, cv=cv),
memory_metrics=memory_metrics
)