marpledata 3.0.1

Marple SDK for Python

Marple SDK

An SDK to interact with Marple DB & Insight

Installation and importing

Install the Marple SDK using your package manager:

• poetry add marpledata
• uv add marpledata
• pip install marpledata

The SDK currently exposes:

from marple import DB      # Marple DB
from marple import Insight # Marple Insight

Marple DB

To get started:

• Create a stream in the Marple DB UI
• Create an API token (in user settings)

If you are using a VPC or self-hosted version, pass a custom api_url to DB(...) (it should end in /api/v1).

Examples

Import a file and wait for it to import

This is the typical flow for importing a new file into Marple DB:

import time
from marple import DB

# Create a stream + API token in the Marple DB web application
STREAM = "Car data"
API_TOKEN = "<your api token>"
API_URL = "https://db.marpledata.com/api/v1"  # optional if using the default SaaS

db = DB(API_TOKEN, API_URL)

db.check_connection()

stream = db.get_stream(STREAM)
dataset = stream.push_file("examples_race.csv", metadata={"driver": "Mbaerto"})
# Wait at most 10s for the dataset to completely import and get the new state of the dataset
dataset = dataset.wait_for_import(timeout=10)

Filter datasets and get resampled data

# See previous example for setup
import re
from marple.db import Dataset

datasets = stream.get_datasets()  # Get all datasets in a specific Data Stream
# OR
# datasets = db.get_datasets()  # Get all datasets in the datapool

datasets = (
    datasets
    .where_metadata({"car_id": [1, 2], "track": "track_1"})
    .wait_for_import()
    .where_imported()
    .where_signal("car.speed", "max", greater_than=75)
    .where_signal("car.engine.temp", "mean", greater_than=30)
    .where_dataset("n_datapoints", greater_than=100000)
)

def custom_filter_function(dataset: Dataset) -> bool:
    return (
        dataset.metadata.get("weather") == "sunny"
        or dataset.get_signal("car.engine.NGear").stats.get("avg", 0) ** 2 > 16
    )

# Pass any function to filter the datasets on more complex conditions
datasets = datasets.where(custom_filter_function)

# Create an overview of the datasets as a pandas.DataFrame to save it to a CSV.
datasets.to_pandas().to_csv("all_datasets.csv")

# Get a dataframe per dataset of the matching signals which is resampled at a period of 0.17s.
# The regex patterns will match with car.wheel.rear.left.speed, car.wheel.rear.front.speed, ...
for dataset, data in datasets.get_data(
    signals=[
        "car.speed",
        "car.engine.temp",
        re.compile("car.wheel.*.speed"),
        re.compile("car.wheel.*.trq"),
    ],
    resample_rule="0.17s",
):
    machine_learning_model.train(data)

Delete a dataset that failed to import

datasets = stream.get_datasets()
datasets = datasets.where_dataset("import_status", equals="FAILED")

# datasets is of type DatasetList which is a subclass of list so you can do all normal list operations on it.
if len(datasets) > 0:
    datasets[0].delete()

Common operations

• List streams: db.get_streams()
• List datasets in a stream: stream.get_datasets()
• Upload a file to a file-stream: stream.push_file(file_path, metadata={...})
• Wait for a dataset to import: dataset.wait_for_import(timeout=60)
• Download original uploaded file: dataset.download(destination_folder=".")
• Download parquet for a signal: dataset.get_signal(signal_name).download(destination_folder=".")
• Get a resampled df of multiple signals: dataset.get_data(signals=[...], resample_rule="1s")

For live/realtime streams (creating and appending data):

• Create an empty dataset: db.add_dataset(stream_key, dataset_name, metadata=None)
• Upsert signal definitions: db.upsert_signals(stream_key, dataset_id, signals=[...])
• Append timeseries data: db.dataset_append(stream_key, dataset_id, data=df, shape="long"|"wide"|None)

Calling endpoints directly

For advanced use cases, you can call API endpoints directly:

db.get("/health")

Notes on DB API changes

• Methods like DB.push_file, DB.download_signal, and DB.update_metadata are deprecated.
• Prefer stream/dataset methods instead: stream.push_file, dataset.get_signal(...).download(), and dataset.update_metadata(...).
• These are still available for compatibility, but the examples above use the current API.

Marple Insight

Common operations

• List datasets in the workspace: insight.get_datasets()
• Get a Marple DB dataset (by dataset id): insight.get_dataset_mdb(dataset_id)
• List signals in a dataset: insight.get_signals(dataset_filter) / insight.get_signals_mdb(dataset_id)

Example: export a dataset (H5/MAT)

from marple import DB, Insight

INSIGHT_TOKEN = "<your api token>"
INSIGHT_URL = "https://insight.marpledata.com/api/v1"  # optional if using the default SaaS
DB_TOKEN = "<your api token>"
DB_URL = "https://db.marpledata.com/api/v1"  # optional if using the default SaaS
STREAM = "Car data"

insight = Insight(INSIGHT_TOKEN, INSIGHT_URL)
db = DB(DB_TOKEN, DB_URL)

dataset_id = db.get_datasets(STREAM)[0].id
insight_dataset = insight.get_dataset_mdb(dataset_id)

file_path = insight.export_data_mdb(
    dataset_id,
    format="h5",
    signals=["car.speed"],
    destination=".",
)
print("Wrote", file_path)

Calling endpoints directly

For advanced use cases, you can call API endpoints directly:

insight.get("/user/info")
insight.post("sources/search")