Examples
A couple of examples demostrating library’s usage and capabilities. Each example displays configuration file config.json and also the main Python code.
Hello world
A minimal working example.
config.json
1{}
main.py
1from uun_iot import Gateway, IModule
2
3# Every user module has to implement interface IModule
4# That means specifying an arbitrary `id` string attribute identifying the module
5class HelloWorld(IModule):
6 id = "helloWorld"
7 def __init__(self):
8 print("Hello, world!")
9
10# Init function might seem cumbersome now, but it is a good idea to get used to it
11# Later, it will have a non-trivial functionality
12def init(config, uuclient):
13 # two arguments (not used right now):
14 # config: a configuration dictionary used to initialize the modules
15 # an UuClient instance which is useful for communication with uuApp servers
16 return HelloWorld()
17
18# Gateway is initialized with a config file and an init function.
19# Gateway is a context manager, meaning it can be usedin `with` statement
20# and you do not have to worry about starting and stopping the Gateway
21# The config file is an empty ('{}') JSON file for now
22with Gateway("config.json", init) as g:
23 # The body of the `with` statement is empty
24 # as we want to start and immediately exit
25 pass
26
27# Equivalent to this particular with statement is
28# g = Gateway("config.json", init)
29# g.start()
30# g.stop()
31
32# Output >>>
33# Hello world!
Events
The library has support for basic event listeners. Examples of such events are start and stop of the Gateway main object, timer ticks and configuration updates. For more information, see on() decorator.
Timers
Single timer
config.json
1{
2 "gateway": {
3 "moduleTimers": {
4 "timerModule": 1
5 }
6 }
7}
main.py
1import time
2from uun_iot import Gateway, IModule, on
3
4class TimerModule(IModule):
5 id="timerModule"
6
7 def __init__(self):
8 pass
9
10 # This is an `on` decorator, which lets you
11 # execute functions on given events
12 # One of these events is a `tick` event
13 @on("tick")
14 def periodical(self):
15 print("Timer with one second period.")
16
17def init(config, uuclient):
18 return TimerModule()
19
20with Gateway("config.json", init) as g:
21 # You can use an infinite while loop for now
22 # it will keep application running "in the background"
23 # BEWARE, it prevents proper app exit in some cases and is not very efficient
24 while True:
25 time.sleep(.1)
Multiple timers
config.json
1{
2 "gateway": {
3 "moduleTimers": {
4 "sendReceive": {
5 "send": 2,
6 "get": 1
7 }
8 },
9
10 "sendReceive": {
11 "ip": "192.168.0.1",
12 "endpoint": "v1/get"
13 }
14 }
15}
main.py
1import time
2from uun_iot import Gateway, Module, on
3
4# we will use the Module base class which already implements IModule
5class SendReceive(Module):
6 # notice, that we do not have to specify `id`, it is already set
7 # to snakeCased version of the module name ("sendReceive")
8
9 def __init__(self, config):
10 # here, we will use constructor of Module
11 # config is a parsed configuration dictionary
12 super().__init__(config=config)
13
14 # The `@on("tick")` decorator can receive another argument
15 # indicating a special timer ID to be linked to.
16 @on("tick", "get")
17 def get(self):
18 # The Module automatically copies config to `self._config`
19 ip = self._config["sendReceive"]["ip"]
20
21 # This is a shorthand to access configuration. Here, it is equivalent
22 # to `self._config["sendReceive"]["endpoint"].
23 # It helps mainly with nested paths and scopes in configuration, see next examples.
24 endpoint = self._c("endpoint")
25 print(f"Retrieving data from {ip}/{endpoint}.")
26
27 @on("tick", "send")
28 def send(self):
29 print(f"Sending data...")
30
31def init(config, uuclient):
32 # config is a parsed JSON config file passed by Gateway
33 gconfig = config["gateway"]
34 return SendReceive(gconfig)
35
36with Gateway("config.json", init) as g:
37 while True:
38 time.sleep(.1)
39
40# Output >>>
41# Retrieving data from 192.168.0.1/v1/get.
42# Sending data...
43# Retrieving data from 192.168.0.1/v1/get.
44# Retrieving data from 192.168.0.1/v1/get.
45# Sending data...
Start & stop
config.json
1{}
main.py
1from uun_iot import Gateway, Module, on
2import time
3import random
4import logging
5
6class OnStartStop(Module):
7 def __init__(self):
8 super().__init__()
9
10 @on("start")
11 def oneshot(self, evs):
12 runev, stopev = evs
13 # runev and stopev are threading.Event object which can be waited for
14 # the corresponding event is set on start and stop of Gateway, respectively
15 while runev.is_set():
16 print("Polling for serial connection...")
17 data = round(random.random()*100)
18
19 # Module automatically initializes uun_iot.utils.Storage
20 self._storage.append(data)
21 if data > 50:
22 time.sleep(1)
23 else:
24 time.sleep(.2)
25
26 @on("stop")
27 def on_stop(self):
28 print()
29 print("Ending polling. Here are results:")
30 print(self._storage.data)
31 print()
32
33
34def init(config, uuclient):
35 return OnStartStop()
36
37with Gateway("config.json", init) as g:
38 while True:
39 time.sleep(.1)
40
41# Output >>>
42# Polling for serial connection...
43# Polling for serial connection...
44# Polling for serial connection...
45# Polling for serial connection...
46# Polling for serial connection...
47# ^C
48# Ending polling. Here are results:
49# [98, 88, 11, 36, 39]
Config
Basic configuration
config.json
1{
2 "gateway": {
3 "userDatabase": {
4 "vicki": {
5 "password": "L7yxWo7{Cr",
6 "url": "api/getUser",
7 "timeout": 5
8 },
9 "miller": {
10 "password": "8EASt!+",
11 "url": "api/getUser",
12 "timeout": 10
13 }
14 },
15 "someOtherModule": {
16 "data": "value"
17 }
18 }
19}
main.py
1from uun_iot import Gateway, Module, on
2from uun_iot.modules.Module import ConfigScopeEnum
3import time
4
5class UserDatabase(Module):
6 def __init__(self, config):
7 super().__init__(config=config)
8 print("Printing value from other module's configuration:")
9 print(self._c("someOtherModule/data", scope=ConfigScopeEnum.GATEWAY))
10
11 @on("start")
12 def oneshot(self, runev):
13 password = self._c(f"vicki/password")
14 print(f"Vicki tried password '{password}'.")
15 print(password == self._config["userDatabase"]["vicki"]["password"])
16
17def init(config, uuclient):
18 gconfig = config["gateway"]
19 return UserDatabase(gconfig)
20
21with Gateway("config.json", init) as g:
22 pass
23
24# Output >>>
25# Printing value from other module's configuration:
26# value
27# Vicki tried password 'L7yxWo7{Cr'.
28# True
Storage
Typically, you would like to use a data storage in your module. This is exactly what baseclass Module provides. It offers a _storage object of uun_iot.utils.Storage.
Basic storage
config.json
1{
2 "gateway": {
3 "moduleBackupStorage": {
4 "storageTest": "backup/storageTest.json"
5 },
6 "moduleTimers": {
7 "storageTest": 0.5
8 }
9 }
10}
main.py
1from uun_iot import Gateway, Module, on
2from uun_iot import utils
3import random
4
5import time
6
7class StorageTest(Module):
8 def __init__(self, config):
9 super().__init__(config=config)
10
11 @on("tick")
12 def tick(self):
13 print("Measuring data...")
14 data = round(random.random()*100)
15 self._storage.append(data, backup=True)
16 print(self._storage.data)
17
18def init(config, uuclient):
19 gconfig = config["gateway"]
20 return StorageTest(gconfig)
21
22with Gateway("config.json", init) as g:
23 while True:
24 time.sleep(.1)
25
26# Output >>>
27# Measuring data...
28# [84]
29# Measuring data...
30# [84, 35]
31# Measuring data...
32# [84, 35, 83]
33# Measuring data...
34# [84, 35, 83, 42]
35# Measuring data...
36# [84, 35, 83, 42, 31]
37
38# (exit & start again) Output >>>
39# Measuring data...
40# [84, 35, 83, 42, 31, 91]
41# Measuring data...
42# [84, 35, 83, 42, 31, 91, 0]
43# Measuring data...
44# [84, 35, 83, 42, 31, 91, 0, 53]
45# Measuring data...
46# [84, 35, 83, 42, 31, 91, 0, 53, 24]
47# Measuring data...
48# [84, 35, 83, 42, 31, 91, 0, 53, 24, 6]
49
Backup of storage to file
config.json
1{
2 "gateway": {
3 "moduleBackupStorage": {
4 "storageTestUnlimited": "backup/storageTestUnlimited.json",
5 "storageTestLimited": {
6 "path": "backup/storageTestLimited.json",
7 "limit": 3
8 }
9 }
10 }
11}
main.py
1from uun_iot import Gateway, Module, on
2import time
3
4class StorageTestUnlimited(Module):
5 def __init__(self, config):
6 super().__init__(config=config)
7
8 @on("start")
9 def oneshot(self, events):
10 runev, stopev = events
11 while runev.is_set():
12 print("1Polling for serial connection...")
13 time.sleep(.3)
14 data = 53.8
15 self._storage.append(data, backup=True)
16 print(self._storage.data)
17
18class StorageTestLimited(Module):
19 def __init__(self, config):
20 super().__init__(config=config)
21
22 @on("start")
23 def oneshot(self, events):
24 runev, stopev = events
25 while runev.is_set():
26 print("2Polling for serial connection...")
27 time.sleep(.2)
28 data = 12.1
29 self._storage.append(data, backup=True)
30 print(self._storage.data)
31
32def init(config, uuclient):
33 gconfig = config["gateway"]
34 return [StorageTestUnlimited(gconfig), StorageTestLimited(gconfig)]
35
36with Gateway("config.json", init) as g:
37 while True:
38 time.sleep(.1)
39
40# Output >>>
41# 1Polling for serial connection...
42# 2Polling for serial connection...
43# [12.1]
44# 2Polling for serial connection...
45# [53.8]
46# 1Polling for serial connection...
47# [12.1, 12.1]
48# 2Polling for serial connection...
49# [53.8, 53.8]
50# 1Polling for serial connection...
51# [12.1, 12.1, 12.1]
52# 2Polling for serial connection...
53# [12.1, 12.1, 12.1]
54# 2Polling for serial connection...
55# [53.8, 53.8, 53.8]
56# 1Polling for serial connection...
57# [12.1, 12.1, 12.1]
58# 2Polling for serial connection...
59# [53.8, 53.8, 53.8, 53.8]
60# 1Polling for serial connection...
61# [12.1, 12.1, 12.1]
62# 2Polling for serial connection...
63# [12.1, 12.1, 12.1]
64# 2Polling for serial connection...
65# [53.8, 53.8, 53.8, 53.8, 53.8]
66# 1Polling for serial connection...
67# [12.1, 12.1, 12.1]