From 93bb6ef0e30b6f6bf37ac7cd864eda76075ab90e Mon Sep 17 00:00:00 2001 From: Kai Chappell Date: Sun, 31 Aug 2025 17:24:16 +0000 Subject: [PATCH] Implement test logger --- src/py_dvt_ate/framework/__init__.py | 3 +- src/py_dvt_ate/framework/context.py | 2 +- src/py_dvt_ate/framework/logger.py | 222 +++++++++++++++++++++++++++ 3 files changed, 225 insertions(+), 2 deletions(-) create mode 100644 src/py_dvt_ate/framework/logger.py diff --git a/src/py_dvt_ate/framework/__init__.py b/src/py_dvt_ate/framework/__init__.py index 4b1a8b9..0861d66 100644 --- a/src/py_dvt_ate/framework/__init__.py +++ b/src/py_dvt_ate/framework/__init__.py @@ -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"] diff --git a/src/py_dvt_ate/framework/context.py b/src/py_dvt_ate/framework/context.py index 3669ec7..6d9fc34 100644 --- a/src/py_dvt_ate/framework/context.py +++ b/src/py_dvt_ate/framework/context.py @@ -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 diff --git a/src/py_dvt_ate/framework/logger.py b/src/py_dvt_ate/framework/logger.py new file mode 100644 index 0000000..0b33248 --- /dev/null +++ b/src/py_dvt_ate/framework/logger.py @@ -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