Querying ROS data with SDKs

This guide explains how to query and decode ROS data with the official ReductStore client SDKs. It expands on the SDK workflow introduced in the ROS applications guide.

note

This guide uses Python only as a representative example; the same workflow and ReductROS options are available with the other official SDKs.

ReductROS availability

Server-side ROS decoding requires the ReductROS extension, which is available with ReductStore Pro. The public Play server used in this guide already provides the extension.

Shared demo resources

The Play server and its reductstore token are shared demo resources. Do not upload private data to this server or use the demo token in production.

Query and save images

The example queries filtered infrared images from the public ROS dataset, samples one image every 30 seconds, and saves the results as JPEG files. It uses the ReductROS extension to decode ROS messages and encode raw image data as JPEG.

The script performs the following steps:

• Connect to the public Play server with the reductstore token and open the orion bucket.
• Query the /right_ir/rotated/image_raw entry for the previous hour.
• Apply the strict range 100 < gps_z < 120 to the stored gps_z label before decoding payloads.
• Use $each_t to keep at most one matching image every 30 seconds, reducing decoding work and network traffic.
• Ask ReductROS through #ext.ros.extract to decode each CDR image using its $schema attachment and encode the raw data field as JPEG.
• Base64-decode each result and save it as orion-images/<timestamp>.jpg.

import asyncio
import base64
import json
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path

from reduct import Client


SERVER_URL = 'https://play.reduct.store/'
API_TOKEN = 'reductstore'
BUCKET_NAME = 'orion'
ENTRY_NAME = '/right_ir/rotated/image_raw'
OUTPUT_DIR = Path(os.environ.get('OUTPUT_DIR', 'orion-images'))


def last_hour_query(now=None):
    stop = now or datetime.now(timezone.utc)
    start = stop - timedelta(hours=1)
    condition = {
        '$and': [
            {'&gps_z': {'$gt': 100}},
            {'&gps_z': {'$lt': 120}},
        ],
        '$each_t': '30s',
        '#ext': {
            'ros': {
                'extract': {
                    'encode': {
                        'data': 'jpeg',
                    }
                },
            }
        },
    }
    return start, stop, condition


def save_jpeg(message, output_path):
    image = message[0]
    jpeg = base64.decodebytes(image['data'].encode('ascii'))
    output_path.write_bytes(jpeg)
    return {key: value for key, value in image.items() if key != 'data'}


async def main():
    start, stop, condition = last_hour_query()
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    async with Client(SERVER_URL, api_token=API_TOKEN) as client:
        bucket = await client.get_bucket(BUCKET_NAME)

        print(f"Run query: {start} to {stop} condition: {json.dumps(condition)}")
        async for record in bucket.query(
            ENTRY_NAME,
            start=start,
            stop=stop,
            when=condition,
        ):
            message = json.loads((await record.read_all()).decode('utf-8'))
            output_path = OUTPUT_DIR / f'{record.timestamp}.jpg'
            image_info = save_jpeg(message, output_path)

            print(record.entry, record.timestamp, record.labels['gps_z'])
            print(f'Saved JPEG to {output_path}')
            print(f'Image parameters: {json.dumps(image_info, sort_keys=True)}')


if __name__ == '__main__':
    asyncio.run(main())

To run this implementation, clone the documentation repository, enter the example directory, install the Python SDK, and execute the script:

git clone https://github.com/reductstore/website.git
cd website/docs/ros/example
pip install reduct-py
python query_orion.py

See Python bucket queries for the Python API used in this implementation, comparison operators for additional label conditions, and the ReductROS examples for more query patterns.