mfi-ddb 1.1.0

MFI DDB: Python library for streaming data

MFI DDB Library

Library to stream data to Digital Data Backend (DDB) for the MFI project.

Installation

using uv manager

Pre-requisite: Install uv manager

git clone --recurse-submodules https://github.com/cmu-mfi/mfi_ddb_library.git
cd mfi_ddb_library
uv sync

using pip/venv

Linux

git clone --recurse-submodules https://github.com/cmu-mfi/mfi_ddb_library.git
cd mfi_ddb_library
python -m venv .venv
source .venv/bin/activate
pip install .

Windows CMD

git clone --recurse-submodules https://github.com/cmu-mfi/mfi_ddb_library.git
cd mfi_ddb_library
python -m venv .venv
.venv\Scripts\activate.bat
pip install .

Concept

flowchart LR;
    A[Data Source] --> B[Data Adapter];
    D[Streamer] --> M[MQTT Client];
    B --> D;
    C[Topic Family] --> D;
    M-->X[MQTT Broker];    

    classDef highlight fill:#094d57
    class B,C,D highlight

MFI DDB Library gives tools to write

• "MQTT Client" which streams data from a "Data Source" to a "MQTT broker". The data source may not be generating MQTT messages directly as per MFI-DDB schema. The library provides a way to convert the data to MQTT messages and stream them to the broker.

To be able to do the above three major classes are provided:

• Data Adapter: These are the objects that represent the data that needs to be streamed. These objects are responsible for converting the data to mfi_ddb ingestible structure.
• Streamer: This is responsible for publishing MQTT messages to the broker. It uses the data from data adapters to stream the data. Streaming can be event driven using obeserver callback or polling based.
• Topic Family: These classes allow streamer to convert data from adapters to MQTT payload as required by respective topic branch (historian, blob, kv)

Usage

• Review the examples for usage.

• [WIP] mqtt_client_tutorial.ipynb gives a step by step tutorial to write your own MQTT client for mfi_ddb franework.

Available Classes

Data Adapters

• ROS
• ROS Files
• Local Files
• MQTT
• MTConnect
• gRPC
• key_value

To create a new data adapter, please follow the checklist and refer to the existing data adapters for implementation reference.

Use the right PR template by following instructions here.

Streamer

• Streamer

Topic Family

• BaseTopicFamily
• BlobTopicFamily
• KeyValueTopicFamily
• SpbTopicFamily

Streaming Metadata

When streaming data to the broker, the following metadata is recorded through the mfi-ddb stream:

Metadata	Description	Recorded as
location context	The location context of the data being streamed, which includes the enterprise, site, area, and device.	topic structure
attributes	Key-value pairs that provide additional information about the data being streamed. These are defined in the adapter yaml configuration file.	streamed on the same topic before data using the same topic family encoding
streaming configuration	The configuration of the data stream, which includes broker information, enterprise and site details.	streamed on the kv and blob at birth and death of data streaming
adapter configuration	The configuration of the adapter that is streaming the data, which includes all the components and their attributes	streamed on the kv and blob at birth and death of data streaming

Executable Modules

store_cfs.py

Example usage

python -m mfi_ddb.scripts.store_cfs path/to/mqtt.yaml path/to/cfs.yaml

Command-line arguments

usage: store_cfs.py [-h] mqtt_config_path cfs_config_path

Subscribe to a topic and save files to Cloud File Store (CFS) based on configuration.

positional arguments:
  mqtt_config_path  Path to the MQTT configuration file (e.g., mqtt.yaml).
  cfs_config_path   Path to the CFS configuration file (e.g., cfs.yaml).

stream_data.py

Example usage

Use a configuration directory:

$ python -m mfi_ddb.scripts.stream_data --data_adapter 'MQTT' --config_dir ./configs

Use specific configuration files:

$ python -m mfi_ddb.scripts.stream_data -d 'Local Files' --adapter_cfg ./configs/local_files.yaml --streamer_cfg ./configs/streamer.yaml

Enable polling mode with a specific rate (in Hz):

$ python -m mfi_ddb.scripts.stream_data -d 'MTConnect' -a ./configs/mtconnect.yaml -s ./configs/streamer.yaml -p True -r 2

Command-line arguments

usage: stream_data.py [-h] --data_adapter DATA_ADAPTER [--config_dir CONFIG_DIR]
                      [--adapter_cfg ADAPTER_CFG] [--streamer_cfg STREAMER_CFG] [--polling POLLING]
                      [--poll_rate POLL_RATE]

Stream data using MFI-DDB library.

optional arguments:
  -h, --help            show this help message and exit
  --data_adapter DATA_ADAPTER, -d DATA_ADAPTER
                        Type of data adapter to use. Supported: 'Local Files', 'MTConnect', 'MQTT',
                        'ROS', 'ROS Files'
  --config_dir CONFIG_DIR, -cd CONFIG_DIR
                        Directory containing the configuration files (local_files.yaml and mqtt.yaml).
                        If --streamer_cfg or --adapter_cfg are provided, this argument is ignored.
  --adapter_cfg ADAPTER_CFG, -a ADAPTER_CFG
                        Path to the local files adapter configuration file (local_files.yaml).
  --streamer_cfg STREAMER_CFG, -s STREAMER_CFG
                        Path to the Streamer configuration file (streamer.yaml).
  --polling POLLING, -p POLLING
                        Enable polling mode. Default is False.
  --poll_rate POLL_RATE, -r POLL_RATE
                        Polling rate in Hz. Default is 1 Hz, if --polling is set to True.

License

This project is licensed under the BSD-3-Clause License - see the LICENSE file for details.