Cameras and 3D Reconstruction

Cameras

CameraService (client.camera in Python) talks to a physical camera that is linked to a Camera config in the scene (Setting Up a Scene) by its camera_id. Data is saved on the server; the RPCs return status and file names, not pixels.

RPC	Description
IsCameraOnline	online is true if the camera responds.
UpdateCameraIntrinsicFromDevice	Read fx, fy, cx, cy and distortion_coeffs from the device and store them.
GetCameraRGBDImage	Capture an RGB-D image; saved as save_file_name if not empty.
GetCameraPointCloud	Capture a point cloud in the camera frame, or in the world frame with in_world_frame=True; saved as save_file_name if not empty.
ListSavedCameraImages / ListSavedCameraPointClouds	File names saved for a camera.
DeleteSavedCameraImage / DeleteSavedCameraPointCloud	Delete a saved file. Return a bare common.Status.

All requests identify the camera by camera_name (the scene config name).

The example examples/python/camera_example.py needs a Flexiv robot with RDK remote mode enabled and a camera with ID 9 enabled in the AI console. Edit the serial number and camera ID for your setup.

print("Checking if the camera is online...")
try:
    response = client.camera.stub.IsCameraOnline(CameraNameRequest(camera_name=camera_name))
    print(f"Camera online: {response.online}")
    if not response.online:
        return
except grpc.RpcError as e:
    print(f"Camera online check failed or environment not running a server yet: {e}")
    return

# 6. Update camera intrinsics from device
print("Updating camera intrinsics from device...")
try:
    update_response = client.camera.stub.UpdateCameraIntrinsicFromDevice(CameraNameRequest(camera_name=camera_name))
    if update_response.status.code == 0:
        print(f"Successfully updated camera intrinsics for: {camera_name}")
    else:
        print(f"Failed to update camera intrinsics for: {camera_name}")
except grpc.RpcError as e:
    print(f"Update camera intrinsics failed: {e}")
    return

# 7. Save the camera's RGB and depth image to disk
print("Saving camera RGBD image...")
image_file_name = "example_rgbd_image"
try:
    rgbd_response = client.camera.stub.GetCameraRGBDImage(
        CameraRGBDImageRequest(camera_name=camera_name, save_file_name=image_file_name)
    )
    if rgbd_response.status.code == 0:
        print(f"Saved RGBD image as: {image_file_name}")
    else:
        print(f"Failed to get RGBD image: {rgbd_response.status.message}")
except grpc.RpcError as e:
    print(f"Get camera RGBD image failed: {e}")

time.sleep(1)  # Wait a moment to ensure the image is saved before listing

# 8. Save the camera's point cloud to disk
print("Saving camera point cloud...")
point_cloud_file_name = "example_point_cloud"
try:
    pcd_response = client.camera.stub.GetCameraPointCloud(
        CameraPointCloudRequest(
            camera_name=camera_name, in_world_frame=True, save_file_name=point_cloud_file_name
        )
    )
    if pcd_response.status.code == 0:
        print(f"Saved point cloud as: {point_cloud_file_name}")
    else:
        print(f"Failed to get point cloud: {pcd_response.status.message}")
except grpc.RpcError as e:
    print(f"Get camera point cloud failed: {e}")

3D reconstruction

ReconstructionService (client.reconstruction) fuses depth data from one or more cameras into a point cloud or mesh and can add the result to the scene as an object.

Workflow:

1. Save a ReconstructionConfig (name, point_cloud_size, voxel_size in m, voxel_bucket_count) with AddReconstructionConfig.

2. StartReconstruction(result_format, publish_frequency, reconstruction_config_name). result_format is, for example, "mesh". reconstruction_config_name is optional: leave it empty to use the server’s default parameters, so step 1 is optional too.

3. AddCameraToReconstruction(camera_names=[...]).

4. Call GetReconstructionResult(header, add_to_scene, replace_existing_object) as often as you need. With add_to_scene=True the result becomes a scene object named by header; replace_existing_object replaces the previous result instead of appending to it.

5. RemoveCameraFromReconstruction, StopReconstruction, and optionally RemoveReconstructionConfig. It takes a whole ReconstructionConfig, but only name is used.

ListReconstructionObjects lists results (name and format). GetReconstructionConfig and ListReconstructionConfigs query saved configurations.

Note

Most ReconstructionService RPCs return a bare common.Status. Check .code on the returned message directly.

The examples print grpc.RpcError instead of raising it. Production code should check both layers; see Error Handling.

print("Adding reconstruction config...")
reconstruction_config_name = "example_reconstruction_config"
reconstruction_config = ReconstructionConfig(
    name=reconstruction_config_name,
    point_cloud_size=1000000,
    voxel_size=0.005,
    voxel_bucket_count=10000,
)
try:
    status = client.reconstruction.stub.AddReconstructionConfig(reconstruction_config)
    if status.code != 0:
        print(f"Failed to add reconstruction config: {status.message}")
        return
except grpc.RpcError as e:
    print(f"Add reconstruction config failed: {e}")
    return

# Run a reconstruction using the camera
print("Starting reconstruction...")
try:
    status = client.reconstruction.stub.StartReconstruction(
        StartReconstructionRequest(
            result_format="mesh",
            publish_frequency=1.0,
            reconstruction_config_name=reconstruction_config_name,
        )
    )
    if status.code != 0:
        print(f"Failed to start reconstruction: {status.message}")
        return
except grpc.RpcError as e:
    print(f"Start reconstruction failed: {e}")
    return

print("Adding camera to reconstruction...")
try:
    status = client.reconstruction.stub.AddCameraToReconstruction(
        CameraNamesRequest(camera_names=["camera_1"])
    )
    if status.code != 0:
        print(f"Failed to add camera to reconstruction: {status.message}")
        return
except grpc.RpcError as e:
    print(f"Add camera to reconstruction failed: {e}")
    return

print("Fetching reconstruction result...")
counter = 0
while counter < 20:
    try:
        result = client.reconstruction.stub.GetReconstructionResult(
            ReconstructionResultRequest(
                header=make_header("example_reconstruction"),
                add_to_scene=True,
                replace_existing_object=True,
            )
        )
        if result.status.code != 0:
            print(f"Failed to get reconstruction result: {result.status.message}")
        else:
            print("Reconstruction result saved successfully")
    except grpc.RpcError as e:
        print(f"Get reconstruction result failed: {e}")
        break
    counter+=1
    time.sleep(1)

print("Listing available reconstruction objects...")
try:
    response = client.reconstruction.stub.ListReconstructionObjects(empty_pb2.Empty())
    if response.status.code != 0:
        print(f"Failed to list reconstruction objects: {response.status.message}")
    else:
        for obj in response.objects:
            print(f"  - {obj.name} ({obj.result_format})")
except grpc.RpcError as e:
    print(f"List reconstruction objects failed: {e}")

print("Removing camera from reconstruction...")
try:
    status = client.reconstruction.stub.RemoveCameraFromReconstruction(
        CameraNamesRequest(camera_names=["camera_1"])
    )
    if status.code != 0:
        print(f"Failed to remove camera from reconstruction: {status.message}")
except grpc.RpcError as e:
    print(f"Remove camera from reconstruction failed: {e}")

print("Stopping reconstruction...")
try:
    client.reconstruction.stub.StopReconstruction(empty_pb2.Empty())
except grpc.RpcError as e:
    print(f"Stop reconstruction failed: {e}")

print("Removing reconstruction config...")
try:
    status = client.reconstruction.stub.RemoveReconstructionConfig(reconstruction_config)
    if status.code != 0:
        print(f"Failed to remove reconstruction config: {status.message}")
except grpc.RpcError as e:
    print(f"Remove reconstruction config failed: {e}")