Python
The pre-built python package blickfeld_qb2 is available on pypi. It contains all the necessary sources to communicate securely to a Qb2 device.
Python is recommended for rapid prototyping, data evaluation and tools. Usually, the performance is sufficient for live point cloud streaming & processing as e.g. it directly provides efficient memory-mapped numpy arrays as output. In combination with Open3D, it can cover a lot of application scenarios. |
In this guide the Blickfeld Qb2 Python library will be first installed and a secure connection to a Qb2 device will be established. Then the version of the currently installed firmware is fetched from the device. Finally we take a look at asynchronous usage of the Blickfeld library.
Installing Blickfeld Qb2 Library
The python package to control the Qb2 device is called
and can be installed from the PyPI repository with the respective command shown below:blickfeld_qb2
$ pip3 install -U blickfeld_qb2
After the installation the blickfeld_qb2
library can be imported in Python.
import blickfeld_qb2
Connecting to Qb2
To interact with a Qb2 device, a gRPC connection must be established. Either the IP address or the hostname of the Qb2 device can be used to establish this connection.
The implementation uses the gRPC-
in combination with the Channel
from the TokenFactory
library.blickfeld_qb2
The resulting connection is encrypted by default. The
additionally authenticates a configured account on the Qb2.TokenFactory
import blickfeld_qb2
# Create token factory using an application key
token_factory = blickfeld_qb2.TokenFactory(
application_key_secret="application-key-for-qb2-xxxxxxxxx"
)
# Open a secure & authenticated connection to Qb2
with blickfeld_qb2.Channel(
fqdn_or_ip="qb2-xxxxxxxxx",
token=token_factory
) as channel:
# Call methods of services here
# ..
pass
A matching application key for the account which should be authenticated has to be created before. If unsure please read the documentation about application keys in order to understand how to do this first. If the application key is invalid, the first API call on the channel will result in a gRPC-error with code 5 / In the example, replace |
The can be passed additionally to authenticate a particular device. If not supplied, the authentication only checks if it is a Blickfeld device.
|
Unauthenticated Connection
If user-management is not enabled on Qb2, an unauthenticated connection can be established. The resulting connection will still be encrypted. Here only the gRPC
from the Channel
library is used.blickfeld_qb2
import blickfeld_qb2
# Open a secure connection to Qb2
with blickfeld_qb2.Channel(fqdn_or_ip="qb2-xxxxxxxxx") as channel:
# Call methods of services
# ..
Establishing unauthenticated connections is deprecated and will be disabled for upcoming Qb2 firmware releases. |
Simple API Requests
The code snippet below provides an example of connecting to Qb2 with the hostname
and retrieving information about its firmware version.
In this case, the qb2-xxxxxxxxx
service from Firmware
namespace is used over the gRPC channel making a request blickfeld_qb2.system.services
to receive a response containing information about the currently installed firmware.get_status()
import blickfeld_qb2
# Create token factory using an application key
token_factory = blickfeld_qb2.TokenFactory(
application_key_secret="application-key-for-qb2-xxxxxxxxx"
)
# Open a secure & authenticated connection to Qb2
with blickfeld_qb2.Channel(
fqdn_or_ip="qb2-xxxxxxxxx",
token=token_factory
) as channel:
# Request the firmware status
response = blickfeld_qb2.system.services.Firmware(channel).get_status()
# Extract the firmware version
firmware_version = response.status.installed_firmware_info
# Print the firmware version
print(firmware_version)
# NOTE: A channel can be reused for other requests. It is not required to create a channel per request.
Every response to the service call is an object that typically comprises several nested objects following the protocol schema defined in the corresponding sections of the documentation. In this scenario, the information about the installed firmware version can be obtained via the field
.response.status.installed_firmware_info
The print output should resemble the example shown below:
FirmwareInfo(label='AEGIS v2.0.5', version=Version(major=2, patch=5, revision='3d7b970b', is_release=True), allowed_downgrade_version=Version(major=1, minor=12, patch=6))
Asynchronous Requests
If necessary, the Blickfeld library can also be used in a fully asynchronous mode. In the example shown below we introduce a coroutine
, which opens an asynchronous gRPC channel and utilizes an asynchronous request get_firmware_version()
to obtain the same information about the Qb2 firmware status.async_get_status()
import asyncio
import blickfeld_qb2
async def main():
# Create token factory using an application key
token_factory = blickfeld_qb2.TokenFactory(
application_key_secret="application-key-for-qb2-xxxxxxxxx"
)
# Open a secure, authenticated & asynchronous connection to Qb2
with blickfeld_qb2.Channel(
fqdn_or_ip="qb2-xxxxxxxxx",
token=token_factory
) as channel:
# Request the firmware status asynchronously and await for the response
response = await blickfeld_qb2.system.services.Firmware(channel).async_get_status()
# Get the firmware version from the response
firmware_version = response.status.installed_firmware_info
# Print the firmware version
print(firmware_version)
# NOTE: A channel can be reused for other requests. It is not required to create a channel per request.
# run couroutine on the event loop
asyncio.run(main())
To make an asynchronous request over the asynchronous channel, the desired method name should be prefixed with .
|
This coroutine can be further executed on the event loop. The print output resemble the example shown below:
FirmwareInfo(label='JAKOB v1.5.0', version=Version(major=1, minor=5, revision='e87dfda0', is_release=True), allowed_downgrade_version=Version(minor=26))