Skip to main content

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?

More Documentation

Project details


Release history Release notifications

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for admobilize-malos, version 0.0.1
Filename, size File type Python version Upload date Hashes
Filename, size admobilize_malos-0.0.1-py3-none-any.whl (21.5 kB) File type Wheel Python version py3 Upload date Hashes View hashes
Filename, size admobilize-malos-0.0.1.tar.gz (7.6 kB) File type Source Python version None Upload date Hashes View hashes

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page