Monitor CO2 concentration over time with Elasticsearch
Out of curiosity I bought a AIRCO2NTROL MINI CO2 meter from TFA Dostmann a few months ago to see just how much a single person (me) increases the CO2 concentration in a room. CO2 concentration is measured in parts per million and a normal value outside is about 410 ppm. Looking online you often find that you should keep the concentration below 1000-1200 ppm or your cognitive functions will start to degrade. It only took a few hours in my room to reach these levels. Reading the display and seeing the red dot was enough to know when I need to let in fresh air, but now I also wanted to know the CO2 concentration over time.
The CO2 meter I bought is powered over USB and you can read the CO2 level and temperature over the USB connection. Smarter people than me reverse-engineered the protocol and another awesome person create a nice python library that I then use within a small python script that reads the values continuously and sends them to an Elasticsearch Index. The python script is executed from a systemd service and the systemd service is startet automatically from udev when the CO2 meter is plugged into my Intel NUC.
Creating the Elasticsearch index template
Before sending data to Elasticsearch we should create an index template to correctly set the type of the three fields we want to send. The co2 field will contain an integer, temperature will be available as a float and the timestamp we create will use epoch_millis. Creating the index template can be done with curl for example. Replace the domain with your own and the Basic Auth with a user that is allowed to create index templates, the default elastic admin user for example.
curl --location --request PUT 'https://elasticsearch.domain.tld:443/_template/co2-custom-mapping' \
--header 'Content-Type: application/json' \
--header 'Authorization: Basic XXXXXXXXX' \
--data-raw '{
"index_patterns": "co2",
"order": 2,
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0
},
"mappings": {
"properties": {
"timestamp": {
"type": "date",
"format": "epoch_millis"
},
"co2": {
"type": "integer"
},
"temperature": {
"type": "float"
}
}
}
}'The python script
The script has two dependencies, requests and CO2Meter, these can be installed by pip: pip install requests git+https://github.com/heinemml/CO2Meter. First we set the variable sensor to the path where the USB CO2 meter is mounted at. Normally this would be /dev/hidraw0 but as you will see further down, I create a symlink to /dev/co2monitor because when plugged in, my meter didn't always get hidraw0, sometimes it was hidraw1. We can then use the function get_data() on the sensor to receive a dictionary with the two values for co2 and temperature.
After getting the dictionary I run some sanity checks on the data. I discard the data if either co2 or temperature are missing, or if the value for co2 and temperature are outside of what the technical documentation says it can measure.
Then I add a timestamp (Unix time in milliseconds) to the dictionary, create a JSON from the dictionary and send it to Elasticsearch. The /co2/ inside the URL is the index I want to send the data to. You need to replace the domain with your own and elastic-username-here / elastic-password-here with an user that has create_index and index permission on the co2 index.
#!/usr/bin/env python3
import json
import time
import sys
import requests
from CO2Meter import CO2Meter
sensor = CO2Meter("/dev/co2monitor")
while True:
time.sleep(5)
try:
data = sensor.get_data()
except OSError:
print("co2 monitor unplugged")
sys.exit()
if "co2" not in data:
print("co2 not in data")
continue
if "temperature" not in data:
print("temperature not in data")
continue
if not 0 <= data["co2"] <= 3000:
print("co2 not in range")
continue
if not 0 <= data["temperature"] <= 50:
print("temperature not in range")
continue
data.update({"timestamp": int(round(time.time() * 1000))})
data_json = json.dumps(data)
url = "https://elasticsearch.domain.tld/co2/_doc"
headers = {"Content-Type": "application/json"}
try:
requests.post(
url=url,
data=data_json,
headers=headers,
auth=("elastic-username-here", "elastic-password-here"),
timeout=10,
)
except Exception:
print("No Connection to ElasticSearch")Running the script with systemd
To run the script I created the following systemd service. The service runs the script inside a python virtual environment that has the two dependencies installed. It also runs as the user co2 that you have to create beforehand. As always, when I create a systemd service, I add as much sandboxing as possible. Using PrivateDevices=true is not possible here, since we need to access /dev/co2monitor, but we can close down all access to /dev and allow only access to /dev/co2monitor with DevicePolicy=closed and DeviceAllow=/dev/co2monitor rw. The Environment="PYTHONUNBUFFERED=true" allows the python output to be immediately visible inside jounalctl, without a delay from buffering a bunch of messages first.
[Unit]
Description=co2 and temp monitor
[Service]
Type=exec
User=co2
Group=co2
Environment="PYTHONUNBUFFERED=true"
SyslogIdentifier=co2-monitor
ExecStart=/srv/co2/python3-virtualenv/bin/python3 /srv/co2/monitor.py
# filesystem access
ProtectSystem=strict
ProtectHome=true
PrivateTmp=true
ProtectControlGroups=true
ProtectKernelModules=true
ProtectKernelTunables=true
DevicePolicy=closed
DeviceAllow=/dev/co2monitor rw
# network
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
# misc
NoNewPrivileges=true
PrivateUsers=true
RestrictRealtime=true
ProtectKernelLogs=true
LockPersonality=true
ProtectHostname=true
RemoveIPC=true
RestrictSUIDSGID=true
As you can see, the systemd service has no [install] section and will not be started at boot, so we need to start it another way. Using udev we can do a couple of things. We can create a symlink for the device to always have the same name, we can set the correct permissions to allow the user group co2 to access the device and we can start the systemd service when the CO2 meter ist plugged in. Fot that, create the file /etc/udev/rules.d/90-co2monitor.rules with the following content:
SUBSYSTEMS=="usb", KERNEL=="hidraw*", ATTRS{idVendor}=="04d9", ATTRS{idProduct}=="a052", GROUP="co2", MODE="0660" ACTION=="add", TAG+="systemd", ENV{SYSTEMD_WANTS}="co2.service" SYMLINK+="co2monitor"
The first few options identify the CO2 meter, GROUP="co2", MODE="0660" sets the correct permissions so our user co2 can access the device, SYMLINK+="co2monitor" creates a symlink to /dev/co2monitor and TAG+="systemd", ENV{SYSTEMD_WANTS}="co2.service" starts our systemd service.
Visualizing the data
Using Elasticsearch we can easily create graphs from our data. In Kibana under Visualize create a line graph with the following options. Do the same for temperature.
Then create a Dashboard and add both graphs to it. The result should look like this.
