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

For release notes, see CHANGELOG.md. To publish a release, see RELEASING.md.

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:

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)

Add signals to an imported dataset

Upload additional signals (for example derived channels) on an existing dataset. DataFrames, Arrow tables, and on-disk parquet use LAKE_ARROW_SCHEMA: columns time (int64 nanoseconds) plus value and/or value_text. Times must overlap the dataset time range. Use overwrite=True to replace an existing signal name; a conflict without overwrite raises SignalsAlreadyExistError.

A Series, or a DataFrame without a time column, takes its times from a DatetimeIndex or TimedeltaIndex. Indexed DataFrames must still have a value and/or value_text column.

# Single signal: wait until available before reading
speed = dataset.get_signal("car.speed").get_data()
signal = dataset.add_signal(
    "car.speed_kmh",
    speed * 3.6,
    metadata={"unit": "km/h"},
).wait_until_available()

# Batch: returns IDs as soon as upload completes (no wait)
ids = dataset.add_signals([
    {"name": "car.speed_kmh", "data": speed * 3.6, "metadata": {"unit": "km/h"}},
], overwrite=True, concurrency=4)
signals = dataset.get_signals(signal_ids=ids)

Data assembled from scratch uses the explicit schema columns instead:

import pandas as pd
from marple.db import LAKE_ARROW_SCHEMA  # time + value and/or value_text

samples = pd.DataFrame({"time": [t0, t0 + 1_000_000_000], "value": [1.0, 2.0]})
dataset.add_signal("car.custom", samples)

Processing scripts

Write a process(dataset) function, store it, and try it on any imported dataset. This runs on the server and writes to that dataset.

source = """
from marple.db import Dataset

def process(dataset: Dataset) -> None:
    speed = dataset.get_signal("car.speed").get_data()
    dataset.add_signal("car.speed_kmh", speed * 3.6, metadata={"unit": "km/h"})
"""

script = db.create_script("speed_kmh", source)
dataset = stream.get_dataset(path="lap.csv")
# or: dataset = stream.push_file("lap.csv").wait_for_import()
dataset.run(script)

Pass a .py path or pathlib.Path to create_script / script.update instead of source text. Iterate with script.update(script=...) then dataset.run(script) again (or dataset.run(script, source=...) to save and run in one step). New signals written by the script may need signal.wait_until_available(...) before you read them.

When the script looks right, attach it to the stream with stream.update(scripts=...). That replaces the pipeline (pass [] to detach all). New uploads then run those scripts after ingest.

stream = stream.update(scripts=[script.id])

For files already imported, rerun aliasing and the stream's script pipeline:

dataset = dataset.rerun_processing().wait_for_import()

Use script.update(...) to change source or metadata, and dataset.get_debug_messages() for the latest ingestion's debug log (not the sandbox job log from dataset.run).

Upload large files

stream.push_file(...) starts an ingestion and lets the Marple DB API choose the best upload mode. Depending on the deployment and file size, the SDK can upload through the API server, upload directly to Azure Blob Storage, use a single presigned URL, or split the file into multipart uploads.

The default upload_mode="auto" is recommended for most users. Increase concurrency when uploading large files over a fast connection:

dataset = stream.push_file(
    "large_export.csv",
    metadata={"source": "testbench"},
    concurrency=8,
).wait_for_import(timeout=180)

If your network, proxy, or firewall blocks direct storage URLs, force the SDK to upload through the Marple DB API server:

dataset = stream.push_file(
    "large_export.csv",
    upload_mode="server",
).wait_for_import(timeout=180)

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 (file stream, default): stream.push_file(file_path, metadata={...}, concurrency=4)
• Custom lake ingest (file stream): stream.add_dataset(...) then dataset.add_signal(...) / add_signals([...])
• Add signals to a dataset: dataset.add_signal(name, data, ...) / dataset.add_signals([...]) — data matches LAKE_ARROW_SCHEMA (time + value and/or value_text)
• Wait until a signal is available: signal.wait_until_available(timeout=60)
• Fetch signals by ID: dataset.get_signals(signal_ids=[...], refresh=False)
• 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")
• Delete a stream: stream.delete() or db.delete_stream(stream_key)
• Delete a dataset: dataset.delete() or db.delete_dataset(dataset_id, dataset_path)
• Delete signals: signal.delete(), dataset.delete_signal(signal_id) / dataset.delete_signals(signal_ids), or db.delete_signals(dataset_id, dataset_path, signal_ids)
• Run a processing script: dataset.run(script) or db.run_script(dataset_id, script)
• Create a processing script: db.create_script(name, script)
• Set script pipeline (replaces the full list): stream.update(scripts=[script.id])
• Rerun aliasing + scripts: dataset.rerun_processing().wait_for_import() or stream.rerun_processing([dataset.id])
• Read ingest debug logs: dataset.get_debug_messages()

For live streams:

• Create an empty dataset: stream.add_dataset(dataset_name, metadata=None)
• Upsert signal definitions: dataset.upsert_signals(signals=[...])
• Append timeseries data: dataset.append(data=df, shape="long"|"wide"|None)
• Cool to cold storage: dataset.cool() then dataset.wait_for_import() until FINISHED

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 compatibility paths.
• Prefer stream/dataset methods instead: stream.push_file, dataset.get_signal(...).download(), and dataset.update_metadata(...).
• Use DataStream.push_file for new upload code because it exposes the current upload controls, including concurrency and upload_mode.

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")