ExdFileInterface Guide

ExdFileInterface gives you full control over how file structure and channel values are mapped to protobuf messages. Use it when you need multiple groups, custom attributes, or maximum performance.

Installation

pip install ods-exd-api-box

Abstract Methods

from ods_exd_api_box import ExdFileInterface, exd_api

class MyHandler(ExdFileInterface):

    @classmethod
    def create(cls, file_path: str, parameters: str) -> ExdFileInterface:
        """Factory method. Receives the file path and a parameters string.
        Raise NotMyFileError if this handler cannot process the file."""

    def close(self) -> None:
        """Release any resources (file handles, caches, etc.)."""

    def fill_structure(self, structure: exd_api.StructureResult) -> None:
        """Populate the StructureResult protobuf with groups, channels,
        attributes, data types, and row counts."""

    def get_values(self, request: exd_api.ValuesRequest) -> exd_api.ValuesResult:
        """Return channel values for the requested group, channels, start
        offset, and limit."""

File Rejection with NotMyFileError

When your plugin cannot handle a given file, raise NotMyFileError. You can do this in two places:

1. In create() — when the file extension or header doesn’t match:

   from ods_exd_api_box import NotMyFileError
   
   @classmethod
   def create(cls, file_path: str, parameters: str) -> ExdFileInterface:
       if not file_path.endswith(".tdms"):
           raise NotMyFileError(f"File '{file_path}' is not a TDMS file.")
       return cls(file_path, parameters)
   

2. In fill_structure() — when deeper inspection reveals the file is invalid:

   def fill_structure(self, structure: exd_api.StructureResult) -> None:
       if not self._is_valid_content():
           raise NotMyFileError(f"File '{self.file_path}' has invalid content.")
       # ... populate structure ...
   

The library catches NotMyFileError and returns a gRPC FAILED_PRECONDITION status to the caller.

Parsing Parameters

The parameters argument is a string passed by the ODS server. Use ParamParser to convert it to a dictionary:

from ods_exd_api_box.utils import ParamParser

params = ParamParser.parse_params(parameters)
# Supports: "key1=value1;key2=value2", JSON strings, and Base64-encoded values
value = params.get("my_key", "default")

Adding Attributes with AttributeHelper

AttributeHelper.add() populates protobuf ContextVariables from a Python dictionary:

from ods_exd_api_box.utils import AttributeHelper

# Add file-level attributes
AttributeHelper.add(structure.attributes, {"name": "My Measurement"})

# Add channel-level attributes
AttributeHelper.add(channel.attributes, {
    "description": "Voltage measurement",
    "unit_string": "V",
})

Supported value types: str, int, float, bool, and datetime objects.

Complete Example

This is the test example from tests/external_data_file.py:

from ods_exd_api_box import ExdFileInterface, exd_api
from ods_exd_api_box.utils import AttributeHelper, ParamParser


class ExternalDataFile(ExdFileInterface):
    """Handler for .exd_api_test files."""

    @classmethod
    def create(cls, file_path: str, parameters: str) -> ExdFileInterface:
        if not file_path.endswith(".exd_api_test"):
            from ods_exd_api_box.exceptions import NotMyFileError
            raise NotMyFileError(
                f"File '{file_path}' is not handled by ExternalDataFile."
            )
        return cls(file_path, parameters)

    def __init__(self, file_path: str, parameters: str):
        params = ParamParser.parse_params(parameters)
        params.get("example_param", "default_value")
        self.file_path = file_path

    def close(self):
        pass

    def fill_structure(self, structure: exd_api.StructureResult) -> None:
        # Add file-level attributes
        AttributeHelper.add(
            attributes=structure.attributes,
            properties={"name": "Raw Layer_00001"},
        )

        # Add groups with channels, data types, units, and attributes
        # (typically read from the actual file)
        # ...

    def get_values(self, request: exd_api.ValuesRequest) -> exd_api.ValuesResult:
        # Read data for the requested channels and row range
        # Return a ValuesResult with the appropriate typed arrays
        # ...
        return exd_api.ValuesResult()

The __main__ Block

if __name__ == "__main__":
    from ods_exd_api_box import serve_plugin

    serve_plugin(
        file_type_name="EXD-API-TEST",
        file_type_factory=ExternalDataFile.create,
        file_type_file_patterns=["*.exd_api_test"],
    )

Parameters:

Parameter	Description
file_type_name	Identifier for the file type (used in the registry)
file_type_factory	Your create classmethod — called for each new file
file_type_file_patterns	Glob patterns to match filenames (e.g. ["*.tdms", "*.TDMS"])

serve_plugin() registers your factory in FileHandlerRegistry, parses CLI / env configuration, and starts the gRPC server. See Server Options for all available settings.

Protobuf Structure Reference

StructureResult

StructureResult
├── identifier       (set by the library)
├── name             (set by the library — filename)
├── attributes       (ContextVariables — file-level metadata)
└── groups[]
    ├── name
    ├── id
    ├── total_number_of_channels
    ├── number_of_rows
    ├── attributes   (ContextVariables — group-level metadata)
    └── channels[]
        ├── name
        ├── id
        ├── data_type    (DT_DOUBLE, DT_STRING, DT_DATE, ...)
        ├── unit_string
        └── attributes   (ContextVariables — channel-level metadata)

ValuesRequest

Field	Description
handle	Connection handle from Open
group_id	Which group to read from
channel_ids	List of channel IDs to retrieve
start	Row offset (0-based)
limit	Maximum number of rows to return

ValuesResult

ValuesResult
├── id           (group_id echo)
└── channels[]
    ├── id       (channel_id echo)
    └── values
        ├── data_type
        └── <typed_array>   (double_array, string_array, long_array, ...)

Info Endpoint (Extension)

Note: This is an extension to the standard ASAM ODS EXD-API interface. Standard-compliant clients do not need to use it, but it provides a convenient way to inspect what file types a running plugin supports.

The server exposes a built-in discovery endpoint through the standard GetStructure RPC. Calling GetStructure with a handle whose uuid starts with "exd://info" — without calling Open first — returns a StructureResult describing all registered file handlers.

How to call it

import json
import grpc
from ods_exd_api_box import exd_api, exd_grpc

with grpc.insecure_channel("localhost:50051") as channel:
    service = exd_grpc.ExternalDataReaderStub(channel)

    info = service.GetStructure(
        exd_api.StructureRequest(handle=exd_api.Handle(uuid="exd://info")),
        None,
    )

    for file_type_name, variable in info.attributes.variables.items():
        handler_info = json.loads(variable.string_array.values[0])
        print(handler_info["name"])      # e.g. "MDF4"
        print(handler_info["patterns"])  # e.g. ["*.mdf4"]

Response format

Each registered handler appears as an entry in attributes.variables, keyed by the file type name. The value is a ContextVariableValue containing a single JSON string in string_array.values[0]:

{
  "name": "MDF4",
  "patterns": ["*.mdf4"]
}

JSON field	Description
name	The file type name as registered with FileHandlerRegistry / serve_plugin()
patterns	Glob patterns passed as file_type_file_patterns in serve_plugin()

An empty attributes.variables map means no file handlers have been registered yet.

Next Steps

• Server Options — CLI arguments, environment variables, TLS
• Docker Deployment — containerize your plugin
• Real-World Plugins — see asam_ods_exd_api_nptdms for a production example