Recording

import pathlib
import zipfile
from datetime import datetime

import blickfeld_qb2


def stream_recording(zip_archive: zipfile.ZipFile) -> None:
    # Create token factory using an application key
    token_factory = blickfeld_qb2.TokenFactory(
        application_key_secret="application-key-for-qb2-xxxxxxxxx"
    )

    # Open a secure connection to Qb2
    with blickfeld_qb2.Channel("qb2-xxxxxxxxx", token=token_factory) as channel:
        # Use recording service
        recording_service = blickfeld_qb2.toolkit.services.Record(channel=channel)

        # Request a recording stream from Qb2
        for response in recording_service.stream(
            # The type of point cloud to record, e.g. the full frame or only the foreground
            point_cloud_type=blickfeld_qb2.percept_pipeline.data.PointCloudType.POINT_CLOUD_TYPE_FULL,
            # Option to disable the states stream in the recording to reduce data
            disable_states=False,
            # Option to disable streaming of .vtu and .pcd formats to reduce data
            disable_vtu_and_pcd=False
        ):
            # The recording service sets a flag if data is potentially lost
            if response.data_loss:
                print("Potential data loss e.g. due to low network bandwidth")

            # Create a new file inside the zip archive
            with zip_archive.open(response.file.path, mode="w") as zf:
                # Write the data to the file
                zf.write(response.file.binary)


# Set the name of the recording
recording_name = f"recording_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.zip"

# The recording will be saved under the recording name
output_zip = pathlib.Path(recording_name)

# Create a zipfile at the set path to store the data from the stream
with zipfile.ZipFile(output_zip, mode="w", compression=zipfile.ZIP_STORED) as zip_archive:
    # Start the recording stream
    print("Start recording ...")
    try:
        stream_recording(zip_archive)

    # Stop the recording by pressing Ctrl+C
    except KeyboardInterrupt:
        zip_size_mb = sum([zinfo.file_size for zinfo in zip_archive.filelist]) / 1_000_000
        print(f"Finished recording {recording_name} with size {round(zip_size_mb, ndigits=1)} MB.")

|-📁 recording/		                      # Archive containing the recorded data
| |-📁 states/			                  # Output of the states stream
|   |-states_timestamp-1.json		      # States captured at timestamp 1
|   |-states_timestamp-2.json
|   |- ...
|
| |-📁 raw/			                      # Point clouds in Frame format
|   |-frame_timestamp-1.pb		          # Frame at timestamp 1
|   |-frame_timestamp-2.pb
|   |- ...
|
| |-📁 metadata/                          # Configurations saved as .json
|   |-toolkitRecordStreamRequest.json     # RecordStreamRequest that was sent to start the recording
|   |-systemScanPattern.json              # The ScanPattern of the Qb2
|   |-systemFirmware.json                 # The Firmware running on the Qb2
|   |-perceptPipelineReferenceFrame.json  # The background point cloud as a Frame
|   |-perceptPipelinePerception.json      # The Perception settings of the Qb2
|   |-perceptPipelineDataSource.json      # The configured DataSource
|   |-hardwareIdentification.json         # The product Identification
|   |-coreProcessingAcceleration.json     # The measured Acceleration
|
| |-frame_frame-id-1.vtu                  # Point cloud at frame-id-1 as .vtu (1)
| |-frame_frame-id-1.pcd                  # Point cloud at frame-id-1 as .pcd (2)
| |-frame_frame-id-2.vtu
| |-frame_frame-id-2.pcd
| |- ...