Skip to main content

Remote Sensors API framework

Project description

Remote Sensors API protocol library

This library translates raw bytearrays into the Remote Sensors API models.

This implementation uses the XBee Python library. And translates the incoming device messages into a Remote Sensors Request and Response.

This library uses similar syntax than Sanic or Flask.

Usage

import remote_sensors

from digi.xbee.devices import XBeeDevice

import my_settings as settings


@remote_sensors.bind_to_device()
@remote_sensors.request()
def access_log(request: remote_sensors.Request, *args, **kwargs):
    """Simple Access Logger method."""
    logger.info(f'Request: {request.method} - {request.uri}')


@remote_sensors.bind_to_device()
@remote_sensors.request(method='REGISTRATION')
def handle_registration(request: remote_sensors.Request, *args, **kwargs):
    """Handles the registration request."""
    response = remote_sensors.Response(status='ACK',
                                       transaction=request.transaction)

    # do something about it :D
    payload = {
        'request': request.as_dict(),
        'response': response.as_dict(),
        'completed': False,
    }

    logger.debug(f'Replying with {response.status} to {sensor_message.addresses}')

    # send the response to the device
    return response


@remote_sensors.bind_to_device()
@remote_sensors.request(method='END_REGISTRATION', pass_message=True)
def handle_registration_end(request: remote_sensors.Request, sensor_message: remote_sensors.SensorRequest,
                            *args, **kwargs):
    """Handles the end of registration."""
    response = remote_sensors.Response(status='ACK',
                                       transaction=request.transaction)

    # do something about it :D
    payload = {
        'device': sensor_message.as_dict(),
        'request': request.as_dict(),
        'response': response.as_dict(),
        'completed': True,
    }

    logger.debug(f'Device ({sensor_message.addresses}) has finished registration')

    return response


device = XBeeDevice(settings.SERIAL_PORT, settings.BAUD_RATE)

def main():
    device.open()
    remote_sensors.app.init(device)

    while True:
        time.sleep(1)


if __name__ == '__main__':
    try:
        main()
    finally:
        if device.is_open():
            device.close()

Contact

Arnulfo Solis Ramirez

Project details


Download files

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

Filename, size & hash SHA256 hash help File type Python version Upload date
remote_sensors-0.1.1.tar.gz (10.2 kB) Copy SHA256 hash SHA256 Source None

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 SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page