The lightweight, flexible, configurable and highly extensible framework for IoT devices (like raspberry pi ) that allows for capturing sensor data (and potentially other devices) and consuming it in different ways including sending it to the RADAR-base platform backend. The framework is highly decoupled and extensible. There are 4 major components in the framework -
- Sensors
- Converters
- Publishers
- Data Consumers
The only external dependency is a pub/sub broker or messaging queue that can be easily run using docker images either on the IoT device or elsewhere(like the cloud).
Here is it's architecture. The data flow is from left to right. To know more about each component please take a look at the Configuration section
.───────────.
┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ( )
│`───────────'│
│ Device Handlers │ Configuration│
│ │
│ │ │.───────────.│
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ( )
▲ `───────────'
┌──────────────────┴──────────────────────────┐
│ │
┌────────────────────────────────────────────────┐ ┌──────────────────────────────────┐ ┌──────────────────────────────────┐ ┌────────────────────────────────────────────┐ ┌─────┐ ┌──────────────────────────────────────────────────────────────────────┐
│ │ │ │ │ │ │ │ │ │ │ │
│ Sensor Handler │ │ Other Handlers... │ │ Converter │ │ Publisher │ │ │ │ Data Consumer │
│ │ │ │ │ │ │ │ │ │ │ │
└────────────────────────────────────────────────┘ └──────────────────────────────────┘ └──────────────────────────────────┘ └────────────────────────────────────────────┘ │ │ └──────────────────────────────────────────────────────────────────────┘
╔════════════════════════════════════════════════╗ ╔══════════════════════════════════╗ ┌──────────────────────────────────┐ ┌────────────────────────────────────────────┐ │ │ ╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳
║ ║ ║ ║ │ │ │ │ │ │ ╳ ┌──────────────────────────────────────────────────────────────────┐ ╳
║ ║ ║ ║ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │ │ ╳ │ Essentially the data consumer can be any application or system │ ╳
║ ║ ║ ║ │ │ │ │ │ │ │ │ ╳ │that can subscribe to topics/channels on the pub/sub system. This │ ╳
║ ║ ║ ║ │ │ Message Converter │ │ │ │ │ │ ╳ │ makes it language agnostic and helps connect external systems. │ ╳
║ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ║ ║ ┌───────────────────────┐ ║ │ Abstract Base class for │ │ │ Publisher │ │ │ │ ╳ └──────────────────────────────────────────────────────────────────┘ ╳
║ Sensor ║ ║ │ │ ║ │ │ serialisation and validation │ │ │ Abstract Base class for publishing │ │ │ ╳ ┌──────────────────────────────────────────────────────────────────┐ ╳
║ │Abstract Base class. Has implementation for│ ║ ║ │ │ ║ │ of data. │ │ │ records to the pub/sub broker or system │ │ │ │ ╳ │ Here we provide a general model which should be used when │ ╳
║ polling and flushing the data. The actual ║ ║ │ │ ║ │ │ │ │ │ │ │ │ ╳ │ developing such data consumer to provide flexibility and │ ╳
║ │ data is provided by the subclasses │ ║ ║ │ │ ║ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ │ │ │ │ │ │ ╳ │ extensibility in line with the other parts of the system. │ ╳
║ ║ ║ │ │ ║ │ ▲ │ │ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │ │ ╳ └──────────────────────────────────────────────────────────────────┘ ╳
║ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ ║ ║ │ Handler Specific │ ║ │ extends──┴──extends┐ │ │ ▲ │ │ P │ ╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳╳
║ ▲ ║ ║ │ Implementations │ ║ │ │ │ │ │ ┌extends─────┬─┴──────extends─┐ │ │ u │ ┌──────────────────────────────────────────────────────────────────────┐
║ ┌────extends───┬┴─────extends────┐ ║ ║ │ │ ║ │ │ │ │ │ │ │ │ │ │ b │ │ ┌──────────────────────────────────────────────────────────────────┐ │
║ │ │ │ ║ ║ │ │ ║ │┌────────────────┐ ┌───────────┐│ │ │ │ │ │ │ l │ │ │ Data Handler │ │
║ │ │ │ ║ ║ │ │ ║ ││ │ │ ││ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ i │ │ │ │ │
║┌─────────────┐┌─────────────┐ ┌─────────────┐║ ║ │ │ ║ ││ │ │ Other ││ │ │ │ │ │ │ │ │ │ s │ │ └──────────────────────────────────────────────────────────────────┘ │
║│ ││ │ │ │║ ║ │ │ ║ ││ AvroConverter │ │ Converter ││ │ │ Redis │ │ MQTT │ │ Other │ │ │ h │ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │
║│ Sensor 1 ││ Sensor 2 │ │ Sensor N │║ ║ └───────────────────────┘ ║ ││ │ │ ││ │ │Publisher│ │Publisher│ │Publisher│ │ │ e │ │ │ │
║│fun get_data ││fun get_data │ │fun get_data │║ ║ ║ ││ │ │ ││ │ │ │ │ │..... │ │ │ │ r │ │ │ │
║│ ││ │ │ │║ ║ ║ │└────────────────┘ └───────────┘│ │ └─────────┘ └─────────┘ └─────────┘ │ │ / │ │ Data Consumer │ │
║└─────────────┘└─────────────┘ └─────────────┘║ ║ ║ │ △ △ │ │ △ △ △ │ │ S │ │ │ Abstract base class for consuming the data. The actual ◁─┐ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ u │ │ processing of the consumed data is done by subclasses │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ b │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ s │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ c │ │ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ r │ │ ▲ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ i │ │ ┌──extends───────┬──────┴──extends────────┐ │ │
║ ║ ║ ║ │ └───────┬─────────┘ │ │ │ │ │ │ │ b │ │ │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ e │ │ ┌───────────┐ ┌──────────────┐ ┌──────────────┐ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ b │ │ │Radar Data │ │Visualisation │ │ Mobile App │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ r │ │ │ Consumer │ │Data Consumer │ │ Consumer │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ o │ │ │ │ │ │ ...... │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ k │ │ └───────────┘ └──────────────┘ └──────────────┘ │ │
║ ║ ║ ║ │ │ │ │ △ △ △ │ │ e │ │ ▣ ▣ ▣ │ │
║ ║ ║ ║ │ △ │ │ ┌──────────┐ ┌──────────┐ ┌───────────┐│ │ r │ │ └────────────────┴────┬───────────────────┘ │ │
║ ║ ║ ║ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │ │ │ │ │ │ ││ │ │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ Redis │ │ MQTT │ │ Other ││ │ o │ │ ◎ │ │
║ ║ ║ ║ │ │ │ │ │Connection│ │Connection│ │Connection ││ │ r │ │ .─────────────────────────────────────────────. │ │
║ ║ ║ ║ │ Schema Retriever │ │ │ │ │ │ │.....│ ││ │ │ │ ( External Services ) │ │
║ ║ ║ ║ │ │ │ │ └──────────┘ └──────────┘ └───────────┘│ │ s │ │ `─────────────────────────────────────────────' │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ y │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ s │ │ │ │
║ ║ ║ ║ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ │ │ └─extends────┴─┬────extends───┘ │ │ t │ │ │ │
║ ║ ║ ║ │ ▲ │ │ │ │ │ e │ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │
║ ║ ║ ║ │ ┌─extends──┴┬───extends┐ │ │ ▼ │ │ m │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │ │ │ │ │ │
║ ║ ║ ║ │ ┌────────┐ ┌────────┐ ┌────────┐│ │ │ │ │ │ │ Subscriber │ │ │
║ ║ ║ ║ │ │ │ │ │ │ ││ │ │ │ │ │ │ │Abstract base class for subscribing ▷──────────────┘ │
║ ║ ║ ║ │ │ │ │ │ │ ││ │ Connection │ │ │ │ │ to the pub/sub system │ │
║ ║ ║ ║ │ │ File │ │ URL │ │ Schema ││ │ │ Represents a connection to the pub/sub │ │ │ │ │ │
║ ║ ║ ║ │ │ System │ │ │ │Registry││ │ system │ │ │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ ││ │ │ │ │ │ │ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │
║ ║ ║ ║ │ │ │ │ │ │ ││ │ │ │ │ │ │ ▲ │
║ ║ ║ ║ │ └────────┘ └────────┘ └────────┘│ │ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │ │ │ ┌──extends───────┴──┬─────extends────┐ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ │ │ │
║ ║ ║ ║ │ │ │ │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
║ ║ ║ ║ │ │ │ │ │ │ │ │ Redis │ │ MQTT │ │ Other │ │
║ ║ ║ ║ │ │ │ │ │ │ │ │Subscriber│ │Subscriber│ │Subscriber│ │
║ ║ ║ ║ │ │ │ │ │ │ │ └──────────┘ └──────────┘ └──────────┘ │
║ ║ ║ ║ │ │ │ │ │ │ │ │
╚════════════════════════════════════════════════╝ ╚══════════════════════════════════╝ └──────────────────────────────────┘ └────────────────────────────────────────────┘ └─────┘ └──────────────────────────────────────────────────────────────────────┘
The template for configuration is located at config.yaml.template. Copy this to the config.yaml and modify as required.
Currently, the configuration can be divided in to 4 main components. Each of the components has some sensible defaults but it is recommended to understand this section thoroughly.
-
Sensors: Represented by the key
sensorsin the config file consists of an array of sensor configurations. Each sensor is configured as follows -- name: "your-sensor-name" # Name of your python module which contains the sensor class module: "your_package.your_module" # Name of the class of the sensor in the module class: "YourSensorClass" # topic/channel to publish the data to in pub/sub paradigm publishing_topic: "your-sensor-topic" # polling frequency in milliseconds poll_frequency_ms: 1000 # Flush size for flushing the records flush_size: 100 # Flush after [value] seconds if flush size is not reached flush_after_s: 1000
Currently, implementations for the following sensors are provided -
- Google Coral Environment Board Sensors: In the module - coral_enviro
By Default , No Sensors are added to the Configuration. This is because it is hardware dependent and thus we cannot have a default sensor config. If sensor config is not provided, the program will fail with an exception.
-
Converters: Represented by the key
converterin the config file. This is for Serialisation of the messages captured by the sensors and before publishing them. It comprises of the following fields -converter: name: 'avro' module: 'commons.message_converter' class: 'AvroConverter' validate_only: True schema_retriever: module: 'commons.schema' class: 'FileAvroSchemaRetriever' args: filepath: '/base/path/to/the/schemas' extension: '.avsc'
- If
validate_onlyisTrue, then no conversion of the message is performed. It is only validated against the schema. - The Default value of the converter is
None, so the messages are neither serialised nor validated unless explicitly configured.
-
Schema Retrievers: These are used for retrieving schemas to be used for validation and serialisation.
- Currently, support for Avro is provided out of the box with schema retrievers from Filesystem, URL or Confluent Schema Registry (Which is a part of the RADAR-base platform).
- Each schema retriever is has its own set of required arguments. These can be specified using the
argskey underschema_retriever. For example, theFileAvroSchemaRetrieverneeds a base path where all schemas are stored in the filesystem(filepath) and an extension of the files(extension) as shown in the above example - Right now, the name of the schema is taken from the sensor name as can be seen in class
SensorBasedSchemaNamingStrategyin the commons.schema module. - For more information take look at the Schema Retriever section
- If
-
Publisher: This is for configuring the pubsub broker/messaging queue for sending the captured sensor data. This provides extensibility to the framework so that other services, devices, etc can also capture the data which are outside of this framework. Some examples include mobile applications (like RADAR-base passive RMT), language agnostic data consumers (like using your existing libraries in other language like java to consume the messages), etc. MQTT is a widely used pubsub protocol for IoT devices and frameworks. Redis pubsub is another lightweight pub/sub system and perform well on ARM architectures(like Raspberry PI) too. Plus it provides other features which maybe useful in the future. Hence, there is an out of the box implementation for publishing the data to Redis Pub/Sub but can easily add MQTT(see the Extending section below). It can configured as follows-
publisher: module: 'pubsub.redis_publisher' class: 'RedisPublisher' connection: module: 'pubsub.redis_connection' class: 'RedisConnection' host: 'localhost' port: '6379' password: '' publisher_max_threads: 5
Most of the options are self-explanatory. The
publisher_max_threadsis the max workers for the ThreadPoolExecutor used for publishing messages. If you have large number of sensors and experience slow performance, then increasing the number of Threads for publisher may help. -
Others: Other config options include the following -
expose_config_endpoint: True root_logger_level: INFO scheduler_max_threads: 10
- The above mentioned values are the defaults. Only add these to the config file if need to update.
- The
expose_config_endpoint, if set toTrue, exposes an http endpoint for getting the config of the system. Could be useful if other systems need to use this config. scheduler_max_threadsis the max workers for the ThreadPoolExecutor used by the scheduler for polling sensor data. Increasing this may be beneficial if using a large number of sensors.- The
root_logger_leveldefine the log level for therootlogger which is used across the whole application.
Other than the defaults and the config.yaml file, The framework also supports configuration using the OS Environment Variables. For using these, the keys used in the above config files needs to be prefixed with radar_iot_. This is done to ensure the names never conflict with other environment variables.
For example, to set the root_logger_level value using environment variables, use -
radar_iot_root_logger_level=INFOWith so many config options, it's important to clarify the order in which the configuration is read. It is as follows-
- First the Default Config is loaded.
- This is overwritten with matching keys from the OS environment variables.
- This is further overwritten with values from the
config.yamlfile.
So in descending order of precedence,
config.yaml > OS environment vars > defaultThe Configuration is validated against the json-schema spec at configspec
Pre-requisites:
- It is recommended to use this on a Raspberry pi 3+.
- The requirement to run this is python 3.7+.
- It is recommended to run this on Raspbian Buster OS which comes pre-installed with python 3.7.
- It is also recommended to use
To install the dependencies, navigate to the root folder of the run the following-
python3 -m pip install -r Then, run the program using -
python3 main.pyNote: If using any of the grove pi hat and modules, you will need to install GrovePi library separately using the command below -
sudo bash scripts/install_grovepi.shFor more info, look at the GrovePi official docs
The framework is decoupled and it is easy to extend different components. In particular, there are 4 major components in the framework that can be easily extended -
- The sensor module
- The publisher module
- The Message Converter (Serialisation) module
- The Data uploader or data consumer module.
- Basic knowledge of IoT devices like Raspberry Pi and their hardware and interfacing like sensors.
- Basic knowledge of Python and docker.
- Optional but a basic knowledge about the RADAR-base platform and its components and client apps.
Depending on which part of the framework is being extended, the requirements will vary.
- Sensor module can be extended by adding new sensors by extending the
sensor.Sensorabstract base class (ABC) and implementing the appropriate abstract functions. - After creating the subclass, you just need to add it to the
config.yamlfile as specified in the Configuration section above.
For example, following is implementation of a test sensor. You just need to specify the get_data() method and the rest will be taken care by the framework.
import logging
from sensors import Sensor
logger = logging.getLogger('root')
class YourTestSensor(Sensor):
def __init__(self, name, topic, poll_freq_ms, flush_size, flush_after_s):
super().__init__(name, topic, poll_freq_ms, flush_size, flush_after_s)
def get_data(self):
logger.debug('test data')
# Your logic for getting data from the sensor
return 35.7Remember to pass all the required constructor values to the super class. Also notice the use of the root logger which was discussed earlier in the configuration section.
Additionally, you can also extend other methods of sensor used for polling, flushing, etc. These can be found in the super class Sensor
For already available sensor implementations, take a look at various sensors in the sensors package.
This can be extended by extending the Publisher abstract class in pubsub.publisher module.
You will need to provide implementation of the _publish method which will handle all the logic of publishing the messages. Ideally these should also convert the messages before publishing.
Take a look at RedisPublisher in pubsub.redis_publisher module for an example implementation.
Just extend the MessageConverter abstract class in commons.message_converter module.
You will need to provide implementation for convert (for a single message) and convert_all (for a list of messages) methods.
Take a look at AvroValidatedJsonConverter in the same module for an example implementation.
Please read the data uploader module documentation. This is currently provided in Kotlin programming language but can be created in any language desired (which has clients for subscribing to the pub/sub system).
There are 3 types of schema retrievers provided which are located in the schema module -
-
FileAvroSchemaRetriever: This will retrieve the schemas from the local filesystem. It can configured as follows-
schema_retriever: module: 'commons.schema' class: 'FileAvroSchemaRetriever' args: filepath: '/etc/schemas/avro/sensors' extension: '.avsc'
This will walk down the path in the
filepathspecified and load all the files with theextensionprovided. The example Avro schemas are located in the /etc/schemas/avro/sensors -
GithubAvroSchemaRetriever: This will retrieve the schemas from a Github Repository. It can configured as follows-
schema_retriever: module: 'commons.schema' class: 'GithubAvroSchemaRetriever' args: repo_owner: 'RADAR-base' repo_name: 'RADAR-Schemas' branch: 'sensors' basepath: 'commons/iot/sensor' extension: '.avsc' git_user: 'username' git_password: '*******' token: ''
Like the file retriever, this will also walk down the
basepathin the repository and retrieve schemas with theextensionprovided. If the repository is public, there is no requirement to specify thegit_userandgit_passwordbut it is still recommended as it increases the Github Api limits. You can also provide atoken(personal access token) inplace of username/password for better security. -
SchemaRegistrySchemaRetriever: This will retrieve schemas from the Confluent Schema Registry. It can be configured as follows. The
schema_registry_urlis a required argument.schema_retriever: module: 'commons.schema' class: 'SchemaRegistrySchemaRetriever' args: schema_registry_url: 'https://radar-cns-platform.rosalind.kcl.ac.uk/schema'