Defining an Interpreter

Defining an Interpreter#

Overview#

All equipment captures, structures, and stores data in different ways. Therefore, an Interpreter is responsible for transforming data from lab equipment into a standardised format. Each piece of equipment requires a specific interpreter to handle unique data structures and formats. This document outlines the steps and best practices for creating a new interpreter.

Requirements#

This section briefly describes the programmatic features that constitute an interpreter.

  1. AbstractInterpreter: Each interpreter must subclass the AbstractInterpreter.

  2. Methods: Implement two main methods:

    • measurement(): Process-specific measurements are taken when a measurement is taken.

    • metadata(): Extract and process metadata when the experiment starts.

  3. Error Handling: Use the provided error handling system to ensure resilience.

Defining an interpreter#

This section provides a step-by-step guide on how an interpreter can be defined. It must be noted that the interpreter is by its nature unique to the equipment and, therefore, can differ considerably between types. These guidelines provide the best practices for creating an adapter that will provide the best outputs but is dependent on the developer.

1. Subclass the AbstractInterpreter#

Begin by creating a new class that inherits from AbstractInterpreter. The Interpreter can take any arguments from the parent equipment adapter that may be specific to the data. When the Interpreter is initialised in the EquipmentAdapter, an ErrorHolder may be provided, which must also be set as a parameter. The measurement_manager instance is a singleton that assists with transforming measurements, which will be discussed later.

from core.adapters.equipment_adapter import AbstractInterpreter
from core.measurement_terms.manager import measurement_manager

class CustomEquipmentInterpreter(AbstractInterpreter):
    def __init__(self, custom_argument, error_holder=None):
        super().__init__(error_holder=error_holder)
        self._custom_argument = custom_argument

2. Implement measurement() Method#

The measurement() method is the only required function because it processes the measurement data and transforms it into a standardised structure. The specifics of this function depend on the data variable’s structure and contents. However, the output must be in line with the influx object structure. The measurement_manager instance that has already been imported can help transform specific measurements. However, the developer must map local terms in the data and the measurement objects. As seen below and in the end-end example, the measurement_manager has a dot operator notation (measurement_manager.OD), which will return an instance handling a specific measurement. These instances can then be used to transform() the measurement value and access the standard term for the measurement that can be the key of the output dictionary.

    def measurement(self, data):
        measurements = {}
        update = {
            "tags": {"project": "indpensim"},
            "fields": measurements,
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }
        for name,value in data:
            # User defined mapping
            measurement = self._get_measurement(name)
            measurement_term = measurement.term
            value = measurement.transform(value)
            if measurement_term not in measurements:
            measurements[measurement_term] = measurement_data
        return update

3. Implement metadata() Method (Optional)#

The metadata is called when an experiment starts. It is designed to provide any metadata that the equipment may be able to export when an experiment begins. The metadata() method takes the data from the InputModule and parses metadata from the equipment. This information may include starting parameters such as starting temperature, the calibration of the equipment or specifics regarding the experiment, such as target measurements. For example, the experiment needs to run at 37 degrees. The first step in most cases would be to call the base implementation of metadata, which will simply return a timestamp and experiment ID initial dictionary (this won’t be done if you want a custom experiment ID). From this, any equipment-specific data can be assigned to this dictionary. In some cases, member variables for the interpreter may be set and accessed when a measurement is taken (see end-end example for a more detailed explanation). It must be noted that many pieces of equipment (especially continuous measurement equipment) may not capture any information when an experiment starts. Therefore, the metadata function is often optional and will fall back to the abstract implementation, which returns a timestamp and a unique generated experiment ID.

def metadata(self, data):
    metadata_dict = super().metadata(data)

    # Handle specific metadata fields
    metadata_dict["device_name"] = data.get("device_name")

    return metadata_dict

4. Implement simulate() Method ((Optional))#

End-to-end testing of adapters can be challenging because it will likely require either developing mock equipment or physically running it. The simulate() method is an optional function that uses pre-recorded data to mock a run. This enables testing of the adapter and client tools without requiring real equipment. Unlike normal operation, simulate() is not indefinite and completes once the mock data is processed. This method requires data from the equipment in its native format, specific to each adapter. For example, a file-based adapter (as seen in the end-end example) uses a simulate() function that copies segments from a read_file to a write_file. In contrast, a database-based (seen below) adapter periodically polls a database where the equipment writes data, allowing simulate() to insert measurements and metadata directly into the database. The implementation and structure of simulate depend on the adapter’s needs. For example, a developer may process simulation data in either the adapter.py or interpreter.py file; the goal is to initiate the appropriate InputModule with the mock data.

import time
def simulate(self, metadata, measurements, wait):
    # _db variable is a wrapper for a database.
    self._db.add_metadata(metadata)
    time.sleep(wait)
    for measurement in measurements:
        self._db.add_measurement(measurement)
        time.sleep(wait)

5. Handling errors and raising exceptions#

The adapter system is designed to run continuously, persisting across experiments and downtime. Since many components, such as Input and Output modules, operate in separate threads, exceptions cannot be raised traditionally. Instead, the program must handle errors as they arise without terminating. Error handling is critical in interpreter code due to potential issues from unprocessable inputs or unexpected data formats. To avoid disruptions, interpreters use an ErrorHolder to manage custom LEAF exceptions, which capture specific error types and severity levels for the adapter environment. When an error occurs in the interpreter, it is recommended to use a LEAF exception and pass it to _handle_exception() rather than raising a standard exception. This approach allows errors to be logged and tracked without halting the program. Severity levels (INFO, WARNING, ERROR, CRITICAL) are used to initiate appropriate responses based on error type and severity. The example below shows the InterpreterError exception added to the ErrorHolder using _handle_exception(). This exception prevents the futile processing of bad inputs without terminating the adapter. It signals that an issue has occurred and will attempt to be rectified in the background.

from core.error_handler.exceptions import InterpreterError
from core.error_handler.exceptions import SeverityLevel

def measurement(self, data):
    if data is None:
        exception = InterpreterError("Measurement called without any data.",
                                      severity=SeverityLevel.WARNING)
        self._handle_exception(exception)
        return
Contents