AdMobilize Malos Library
Project description
A simple Python coroutine based driver for communicating with malos-vision.
License
This application follows the GNU General Public License, as described in the LICENSE file.
Installing
The package is available on PyPI, so you can easily install via pip:
$ pip install matrix-io-malos
Running the CLI client
The library includes a simple command line client to start reading data from your MALOS service right away.
# Get the malosclient help screen $ malosclient --help # Get IMU data to STDOUT from a locally running MALOS service $ malosclient IMU # Get HUMIDITY data to STDOUT from a remotely running MALOS service $ malosclient -h 192.168.0.100 HUMIDITY # Get FACE detection data using a serialized driver config file $ malosclient --driver-config-file ~/driver_config.proto VISION
Using the MalosDriver
To use the MALOS driver works as an async generator so in your code you can do the following:
import asyncio import sys from matrix_io.malos.driver import IMU_PORT, UV_PORT from matrix_io.proto.malos.v1 import driver_pb2 from matrix_io.proto.malos.v1 import sense_pb2 from matrix_io.malos.driver import MalosDriver async def imu_data(imu_driver): async for msg in imu_driver.get_data(): print(sense_pb2.Imu().FromString(msg)) await asyncio.sleep(1.0) async def uv_data(uv_driver): async for msg in uv_driver.get_data(): print(sense_pb2.UV().FromString(msg)) await asyncio.sleep(1.0) async def status_handler(driver): type_mapping = { driver_pb2.Status.MESSAGE_TYPE_NOT_DEFINED: "Not Defined", driver_pb2.Status.STARTED: "Started", driver_pb2.Status.STOPPED: "Stopped", driver_pb2.Status.CONFIG_RECEIVED: "Config Received", driver_pb2.Status.COMMAND_EXECUTED: "Command Executed", driver_pb2.Status.STATUS_CRITICAL: "Critical", driver_pb2.Status.STATUS_ERROR: "Error", driver_pb2.Status.STATUS_WARNING: "Warning", driver_pb2.Status.STATUS_INFO: "Info", driver_pb2.Status.STATUS_DEBUG: "Debug" } async for msg in driver.get_status(): print(type_mapping[msg.type]) if msg.uuid: print("UUID: {}".format(msg.uuid)) if msg.message: print("MESSAGE: {}".format(msg.message)) await asyncio.sleep(1.0) # Driver configuration driver_config = driver_pb2.DriverConfig() # Create the drivers imu_driver = MalosDriver('localhost', IMU_PORT) uv_driver = MalosDriver('localhost', UV_PORT) # Create loop and initialize keep-alive loop = asyncio.get_event_loop() loop.run_until_complete(imu_driver.configure(driver_config)) loop.run_until_complete(uv_driver.configure(driver_config)) loop.create_task(imu_driver.start_keep_alive()) loop.create_task(uv_driver.start_keep_alive()) # Initialize data and error handlers loop.create_task(imu_data(imu_driver)) loop.create_task(uv_data(uv_driver)) loop.create_task(status_handler(imu_driver)) loop.create_task(status_handler(uv_driver)) try: loop.run_forever() except KeyboardInterrupt: print('Shutting down. Bye, bye !', file=sys.stderr) finally: loop.stop() asyncio.gather(*asyncio.Task.all_tasks()).cancel() loop.run_until_complete(loop.shutdown_asyncgens()) loop.close()
Who can answer questions about this library?
- Heitor Silva <heitor.silva@admobilize.com>
- Maciej Ruckgaber <maciek.ruckgaber@admobilize.com>
More Documentation
…
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Filename, size | File type | Python version | Upload date | Hashes |
---|---|---|---|---|
Filename, size admobilize_malos-0.0.2-py3-none-any.whl (21.5 kB) | File type Wheel | Python version py3 | Upload date | Hashes View |
Filename, size admobilize-malos-0.0.2.tar.gz (7.6 kB) | File type Source | Python version None | Upload date | Hashes View |
Close
Hashes for admobilize_malos-0.0.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 288c0e56f09da20efec12099f018b218b6a15eab5e69e52c584a504910800a7b |
|
MD5 | 4495c33d704c937c24c405a5a45d40c1 |
|
BLAKE2-256 | 9d6d6bafb8ceaa6afe0d9e7209b3b7504cf359eb1f9f9ec7e3df07716afcfd26 |