A collection of Micropython libraries for the Raspberry Pi Pico.
- libraries - Various Micropython libraries including an async WebSockets client for Home Assistant (see README)
- sensor - Async sensor routines for Home Assistant to report the status of various I2C/GPIO sensors
- infodisplay - Informational display for Home Assistant (Pimoroni Pico Display Pack 2.0/2.8)
- thermostat - Integration with a Home Assistant thermostat (Pimoroni Pico Display Pack 2.0/2.8)
This repository offers a standard async module layout to simplify running, configuring and deploying multiple disparate async modules in parallel, using a lightweight dependency injection system (servicefactory.py).
Async modules are laid out in a standard way: they have a static create method accepting a provider parameter, and an async instance start method:
class MyModule:
def create(provider):
# Factory method
return MyModule()
async def start(self):
# Async workThe async start method is expected to run forever, and will be retried if it exits (e.g. throws an exception, or returns). The start method should also be expected to gracefully cancel.
To deploy an example async module which prints to the console:
- Copy main.py to
main.py - Copy servicefactory.py to
lib/servicefactory.py - Create
config.pywith the following:
import mymodule
config = dict()- Create
mymodule.pywith the following:
import asyncio
class MyModule:
def create(provider):
return MyModule()
async def start(self):
print('Started')
while True:
await asyncio.sleep(30)
print('30s passed')- Run
main.py. You should see the following:
Started
30s passed
30s passed
...
To dep
77E0
loy another async module, simply define it with a static create and async start method, and import it into config.py.
It is possible to inject config and services in the async module's create method. To prepare, make the following changes to config.py:
import mymodule
config = dict(
mymodule = dict(wait_ms = 1_000)
)And copy the following adapted version of mymodule.py:
import asyncio
class MyModule:
def __init__(self, nic, wait_ms):
self.nic = nic
self.wait_ms = wait_ms
def create(provider):
config = provider['config']['mymodule']
return MyModule(provider['nic'], config['wait_ms'])
async def start(self):
print('Started')
while True:
await asyncio.sleep_ms(self.wait_ms)
connected = self.nic.isconnected()
net = 'connected' if connected else 'no network'
print('%ims passed (%s)' % (self.wait_ms, net))This newer version injects config to determine how long to wait, and also the network card to print the network status (both config and nic are basic services registered directly in main.py).
All async modules are made available to the dependency injection container by default, accessible via their fully qualified name. Consider the following module:
class MyModule:
def create(provider):
return MyModule()
async def start(self):
pass
async def mymethod(self):
raise NotImplementedErrorThis can be accessed automatically by another module's creation method by referring to it as mymodule.MyModule:
class MyOtherModule:
def __init__(self, mymodule):
self.mymodule = mymodule
CREATION_PRIORITY = 1
def create(provider):
return MyOtherModule(provider['mymodule.MyModule'])
async def start(self):
# Do something with the injected dependency
await self.mymodule.mymethod()Note that the second module has the following static property:
CREATION_PRIORITY = 1This is a hint to the dependency injection system about what order to initialize dependencies in; higher means later (making this explicit keeps the service factory simple and fast).
An async module may also register its own dependencies with custom names. This can be done by adding to the dependency injection container in an async module's create method:
def create(provider):
provider['myservice'] = MyService()
return MyModule()
async def start(self):
await asyncio.Event().wait()This allows manipulation of the dependency injection container prior to construction of the module. If the start method doesn't ever need to be run, the create method can simply not return:
def create(provider):
provider['myservice'] = MyService()
async def start(self):
raise NotImplementedErrorIn the above example, start will never be called, and myservice will be added to the dependency injection container. This allows async modules to manipulate the service container without changing the dependency injection container in main.py.