Thanks to numerous posts on the forum here, I’ve puzzled together a solution to calculate a NowCast prediction for PM10 and PM2.5.
The following is based on an hourly sensor reading which I obtain near my location from a scientific measurement station. It reports a reading every exact hour.
This is obtained through a REST integration and saved into a sensor aqi_current_pm10 and aqi_current_pm2_5` respectively.
In order to prevent duplicate values, I also capture REST response reporting time as timestamp attribute.
Example:
state: 53.94944
AQI: 25
device_class: pm10
friendly_name: AQI Current PM10
icon: mdi:molecule
state_class: measurement
timestamp: 1705417200 # Tue 16-Jan-24 15:00:00 UTC
unit_of_measurement: µg/m³
Once 12 hours of timestamped data is populated in Home Assistant, we can start the NowCast calculation.
We can extract the last 12 hours of data using the following SQL integration:
sql:
- name: "AQI History PM10"
query: >-
SELECT json_group_array(json_object('h', hour, 'v', ROUND(state, 3))) AS states
FROM (SELECT CAST(TRUNC((STRFTIME('%s', 'now') - A.shared_attrs ->> '$.timestamp') / 3600) AS int) AS hour,
CAST(S.state AS FLOAT) AS state
FROM states S
INNER JOIN state_attributes A ON S.attributes_id = A.attributes_id
INNER JOIN states_meta M ON S.metadata_id = M.metadata_id
WHERE M.entity_id = 'sensor.aqi_current_pm10'
AND A.shared_attrs ->> '$.timestamp' IS NOT NULL
GROUP BY A.shared_attrs ->> '$.timestamp'
ORDER BY A.shared_attrs ->> '$.timestamp' DESC LIMIT 12);
column: "states"
This query looks up the last 12 history states of the aqi_current_pm10 sensor and groups it per unique timestamp to filter out duplicate registrations if any.
It then generates a JSON array of JSON objects with both the hour (T-minus) and value.
The output of the sensor looks like this, an hourly mapping of the last 12 hours and each value reported:
[
{ "h": 0, "v": 53.94944 },
{ "h": 1, "v": 50.94944 },
{ "h": 2, "v": 48.13483 },
{ "h": 3, "v": 56.23595 },
{ "h": 4, "v": 69.98315 },
{ "h": 5, "v": 69.05056 },
{ "h": 6, "v": 73.02809 },
{ "h": 7, "v": 133.2022 },
{ "h": 8, "v": 144.4775 },
{ "h": 9, "v": 113.6461 },
{ "h": 10, "v": 92.84831 },
{ "h": 11, "v": 83.91573 }
]
We can now run a Jinja2 template for the NowCast calculation.
If hours do not have a value, they will be missing from the result and not calculated into the weighted average as described in the above algorithm.
The Air Quality Index calculation implementation has kindly been posted by emoses.
We can now pull this data in with the following Jinja2 template:
(Added additional comments in-line to explain step-by-step)
- sensor:
- name: "NowCast PM10"
attributes:
type: "pm10"
device_class: AQI
state: >
{# Macro to map value to AQI range #}
{% macro aqi(val, val_l, val_h, aqi_l, aqi_h) %}
{{(((aqi_h-aqi_l)/(val_h - val_l) * (val - val_l)) + aqi_l)|round(0)}}
{% endmacro %}
{# Get JSON measurements array into a list #}
{% set measurements = states('sensor.aqi_history_pm10') | from_json %}
{# Get minimum and maximum measurement values #}
{% set min = measurements | min(attribute='v') %}
{% set max = measurements | max(attribute='v') %}
{# Calculated the scaled weight: (max - min) / max #}
{% set weight = (max.v - min.v) / max.v %}
{# If weight < 0.5, set it to 0.5 #}
{% if weight < 0.5 %}
{% set weight = 0.5 %}
{% endif %}
{# Enumerate all the hourly data: weighted results and weighted weights #}
{% set data = namespace(results=[],weights=[]) %}
{% for measurement in measurements %}
{% if measurement.h < 12 %}
{% set data.results = data.results + [measurement.v * (weight ** measurement.h)] %}
{% set data.weights = data.weights + [weight ** measurement.h] %}
{% endif %}
{% endfor %}
{# Get scaled PM10 value #}
{% set v = ((data.results | sum) / (data.weights | sum)) | round(0) %}
{# Map PM10 value to AQI index range #}
{% if v <= 54 %}
{{aqi(v, 0, 54, 0, 50)}}
{% elif 54 < v <= 154 %}
{{aqi(v, 55, 154, 51, 100)}}
{% elif 154 < v <= 254 %}
{{aqi(v, 154, 254, 101, 150)}}
{% elif 254 < v <= 354 %}
{{aqi(v, 254, 354, 151, 200)}}
{% elif 354 < v <= 424 %}
{{aqi(v, 354, 424, 201, 300)}}
{% elif 424 < v <= 604 %}
{{aqi(v, 424, 604, 301, 500)}}
{% else %}
500
{% endif %}
state_class: measurement
unique_id: nowcast_pm10
Thats it! This sensor now allows you to get a NowCast PM10 value based on local measurements.
For completeness, here is the NowCast PM2.5 sensor as well (AQI mappings differ, both SQL sensor and current sensor are the same approach):
- sensor:
- name: "NowCast PM2.5"
attributes:
type: "pm25"
device_class: AQI
state: >
{% macro aqi(val, val_l, val_h, aqi_l, aqi_h) %}
{{(((aqi_h-aqi_l)/(val_h - val_l) * (val - val_l)) + aqi_l)|round(0)}}
{% endmacro %}
{% set measurements = states('sensor.aqi_history_pm2_5') | from_json %}
{% set min = measurements | min(attribute='v') %}
{% set max = measurements | max(attribute='v') %}
{% set weight = (max.v - min.v) / max.v %}
{% if weight < 0.5 %}
{% set weight = 0.5 %}
{% endif %}
{% set data = namespace(results=[],weights=[]) %}
{% for measurement in measurements %}
{% set data.results = data.results + [measurement.v * (weight ** measurement.h)] %}
{% set data.weights = data.weights + [weight ** measurement.h] %}
{% endfor %}
{% set v = ((data.results | sum) / (data.weights | sum)) | round(1) %}
{# Convert value to AQI #}
{% if v <= 12.0 %}
{{aqi(v, 0, 12.0, 0, 50)}}
{% elif 12.0 < v <= 35.4 %}
{{aqi(v, 12.1, 35.4, 51, 100)}}
{% elif 35.4 < v <= 55.4 %}
{{aqi(v, 35.5, 55.4, 101, 150)}}
{% elif 55.4 < v <= 150.4 %}
{{aqi(v, 55.5, 150.4, 151, 200)}}
{% elif 150.4 < v <= 250.4 %}
{{aqi(v, 150.5, 250.4, 201, 300)}}
{% elif 250.4 < v <= 500.4 %}
{{aqi(v, 250.5, 500.4, 301, 500)}}
{% else %}
500
{% endif %}
state_class: measurement
unique_id: nowcast_pm25
Note: I guess this could be done as a custom component as well, but I kind of wanted to see what I could do with simple Jinja2 and some SQL sensors.