Implement test logger
This commit is contained in:
@@ -5,5 +5,6 @@ and runtime context management for DVT characterisation tests.
|
||||
"""
|
||||
|
||||
from py_dvt_ate.framework.context import ITest, TestContext
|
||||
from py_dvt_ate.framework.logger import ITestLogger, TestLogger
|
||||
|
||||
__all__ = ["ITest", "TestContext"]
|
||||
__all__ = ["ITest", "ITestLogger", "TestContext", "TestLogger"]
|
||||
|
||||
@@ -17,7 +17,7 @@ from py_dvt_ate.data.models import TestStatus
|
||||
|
||||
if TYPE_CHECKING:
|
||||
# Avoid circular imports while maintaining type checking
|
||||
from py_dvt_ate.framework.logger import ITestLogger # type: ignore[import-not-found]
|
||||
from py_dvt_ate.framework.logger import ITestLogger
|
||||
from py_dvt_ate.instruments.factory import InstrumentSet
|
||||
|
||||
|
||||
|
||||
222
src/py_dvt_ate/framework/logger.py
Normal file
222
src/py_dvt_ate/framework/logger.py
Normal file
@@ -0,0 +1,222 @@
|
||||
"""Test logger for recording measurements and events.
|
||||
|
||||
This module provides the logging infrastructure for DVT tests. The test logger
|
||||
records time-series measurements, scalar results with limits, and event messages
|
||||
during test execution.
|
||||
"""
|
||||
|
||||
import time
|
||||
from abc import ABC, abstractmethod
|
||||
from datetime import datetime
|
||||
from uuid import UUID
|
||||
|
||||
from py_dvt_ate.data.models import Measurement
|
||||
from py_dvt_ate.data.repository import ITestRepository
|
||||
|
||||
|
||||
class ITestLogger(ABC):
|
||||
"""Abstract interface for test data logging.
|
||||
|
||||
Provides methods for logging measurements, results, and events during
|
||||
test execution. Implementations are responsible for persisting this
|
||||
data to the appropriate storage backend.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def log_measurement(
|
||||
self,
|
||||
parameter: str,
|
||||
value: float,
|
||||
unit: str,
|
||||
conditions: dict[str, float] | None = None,
|
||||
) -> None:
|
||||
"""Log a time-series measurement with environmental conditions.
|
||||
|
||||
Used for logging raw measurements taken during the test. These are
|
||||
stored as time-series data for later analysis and plotting.
|
||||
|
||||
Args:
|
||||
parameter: Measurement parameter name (e.g., "v_out", "i_q").
|
||||
value: Measured value.
|
||||
unit: Unit of measurement (e.g., "V", "A", "°C").
|
||||
conditions: Optional environmental conditions at time of measurement:
|
||||
- "temperature": Chamber temperature (°C)
|
||||
- "input_voltage": DUT input voltage (V)
|
||||
- "load_current": DUT load current (A)
|
||||
|
||||
Example:
|
||||
logger.log_measurement(
|
||||
"v_out", 3.301, "V",
|
||||
conditions={"temperature": 25.0, "input_voltage": 5.0}
|
||||
)
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def log_result(
|
||||
self,
|
||||
parameter: str,
|
||||
value: float,
|
||||
unit: str,
|
||||
lower_limit: float | None = None,
|
||||
upper_limit: float | None = None,
|
||||
) -> None:
|
||||
"""Log a scalar test result with pass/fail limits.
|
||||
|
||||
Used for logging calculated or derived results that will be evaluated
|
||||
against specification limits. These appear in test reports and determine
|
||||
overall pass/fail status.
|
||||
|
||||
Args:
|
||||
parameter: Result parameter name (e.g., "temp_co", "load_reg").
|
||||
value: Calculated or measured value.
|
||||
unit: Unit of measurement (e.g., "ppm/°C", "%", "mV").
|
||||
lower_limit: Optional lower limit for pass/fail evaluation.
|
||||
upper_limit: Optional upper limit for pass/fail evaluation.
|
||||
|
||||
Example:
|
||||
logger.log_result(
|
||||
"temp_co", 23.5, "ppm/°C",
|
||||
lower_limit=-50.0, upper_limit=50.0
|
||||
)
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def log_event(self, message: str, level: str = "INFO") -> None:
|
||||
"""Log a test event or message.
|
||||
|
||||
Used for logging informational messages, warnings, and errors during
|
||||
test execution. Useful for debugging and understanding test flow.
|
||||
|
||||
Args:
|
||||
message: Event message text.
|
||||
level: Log level ("DEBUG", "INFO", "WARNING", "ERROR").
|
||||
|
||||
Example:
|
||||
logger.log_event("Waiting for thermal stability", level="INFO")
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def flush(self) -> None:
|
||||
"""Flush any buffered data to storage.
|
||||
|
||||
Forces any buffered measurements or results to be written to the
|
||||
underlying storage backend. Called automatically at end of test,
|
||||
but can be called manually for long-running tests.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class TestLogger(ITestLogger):
|
||||
"""Concrete test logger implementation using repository pattern.
|
||||
|
||||
Buffers measurements in memory and writes them in batches to a
|
||||
repository for efficiency. Results and events are written immediately.
|
||||
|
||||
Attributes:
|
||||
run_id: UUID of the test run this logger is associated with.
|
||||
repository: Data repository for persisting measurements and results.
|
||||
measurement_buffer: In-memory buffer of measurements awaiting write.
|
||||
buffer_size: Number of measurements to buffer before auto-flush.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
run_id: UUID,
|
||||
repository: ITestRepository,
|
||||
buffer_size: int = 100,
|
||||
):
|
||||
"""Initialise test logger.
|
||||
|
||||
Args:
|
||||
run_id: UUID of the test run to associate logs with.
|
||||
repository: Repository for persisting data.
|
||||
buffer_size: Number of measurements to buffer before auto-flush.
|
||||
Default 100 provides good balance of performance
|
||||
and memory usage.
|
||||
"""
|
||||
self.run_id = run_id
|
||||
self.repository = repository
|
||||
self.buffer_size = buffer_size
|
||||
self.measurement_buffer: list[Measurement] = []
|
||||
|
||||
def log_measurement(
|
||||
self,
|
||||
parameter: str,
|
||||
value: float,
|
||||
unit: str,
|
||||
conditions: dict[str, float] | None = None,
|
||||
) -> None:
|
||||
"""Log a time-series measurement with environmental conditions.
|
||||
|
||||
Measurements are buffered in memory and written to the repository
|
||||
in batches for efficiency.
|
||||
"""
|
||||
conditions = conditions or {}
|
||||
measurement = Measurement(
|
||||
timestamp=time.time(),
|
||||
parameter=parameter,
|
||||
value=value,
|
||||
unit=unit,
|
||||
temperature=conditions.get("temperature", 0.0),
|
||||
input_voltage=conditions.get("input_voltage", 0.0),
|
||||
load_current=conditions.get("load_current", 0.0),
|
||||
)
|
||||
self.measurement_buffer.append(measurement)
|
||||
|
||||
# Auto-flush when buffer is full
|
||||
if len(self.measurement_buffer) >= self.buffer_size:
|
||||
self.flush()
|
||||
|
||||
def log_result(
|
||||
self,
|
||||
parameter: str,
|
||||
value: float,
|
||||
unit: str,
|
||||
lower_limit: float | None = None,
|
||||
upper_limit: float | None = None,
|
||||
) -> None:
|
||||
"""Log a scalar test result with pass/fail limits.
|
||||
|
||||
Results are written immediately to the repository (not buffered).
|
||||
"""
|
||||
self.repository.save_result(
|
||||
run_id=self.run_id,
|
||||
parameter=parameter,
|
||||
value=value,
|
||||
unit=unit,
|
||||
lower_limit=lower_limit,
|
||||
upper_limit=upper_limit,
|
||||
)
|
||||
|
||||
def log_event(self, message: str, level: str = "INFO") -> None:
|
||||
"""Log a test event or message.
|
||||
|
||||
Events are currently logged to console. Future versions may
|
||||
persist events to the repository.
|
||||
"""
|
||||
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||||
print(f"[{timestamp}] {level:7s} {message}")
|
||||
|
||||
def flush(self) -> None:
|
||||
"""Flush buffered measurements to repository.
|
||||
|
||||
Writes all buffered measurements to the repository in a single
|
||||
batch operation, then clears the buffer.
|
||||
"""
|
||||
if self.measurement_buffer:
|
||||
self.repository.save_measurements(
|
||||
run_id=self.run_id,
|
||||
measurements=self.measurement_buffer,
|
||||
)
|
||||
self.measurement_buffer.clear()
|
||||
|
||||
def __del__(self) -> None:
|
||||
"""Ensure buffered data is flushed on logger destruction."""
|
||||
try:
|
||||
self.flush()
|
||||
except Exception:
|
||||
# Ignore errors during cleanup
|
||||
pass
|
||||
Reference in New Issue
Block a user