Built-in AQI entity

[ryan-lang]

I wanted to share some experience I have doing AQI calculations, since it sounds from @siku2 like there's some interest in adding this as a feature. It looks like there's been some discussion (#36) on how to do this with templates, but this would be a new approach as part of the sensor itself.

For US EPA, it's a basic step function. You're essentially mapping values from one predefined range (PM values) to another predefined range (index). I used the values published here: US EPA AQI Breakpoints
Note that the breakpoints are not the same for PM2.5 and PM10. Overall AQI is found by taking the maximum of PM2.5 and PM10 AQI values.

A brief glance makes it seem like other health agencies use the same basic function, but with different breakpoints, but I could be wrong about this.

One gotcha in the implementation is that there's technically a discontinuity between the high breakpoint in one step and the low breakpoint in the next. eg. looking at the first two PM2.5 breakpoints, what happens if your PM2.5 value is 12.05? A silly thing, but it bit me on the first try... Solution is to round first.

[ryan-lang]

My original implementation, for posterity:

class AQICalculator:
    def calculate(self, pm25, pm10):
        raise NotImplementedError
    
class EPA_USA(AQICalculator):
    EPA_USA_BREAKPOINTS_PM25 = [
        (0.0, 12.0, 0, 50),
        (12.1, 35.4, 51, 100),
        (35.5, 55.4, 101, 150),
        (55.5, 150.4, 151, 200),
        (150.5, 250.4, 201, 300),
        (250.5, 350.4, 301, 400),
        (350.5, 500.4, 401, 500)
    ]
    EPA_USA_BREAKPOINTS_PM10 = [
        (0, 54, 0, 50),
        (55, 154, 51, 100),
        (155, 254, 101, 150),
        (255, 354, 151, 200),
        (355, 424, 201, 300),
        (425, 504, 301, 400),
        (505, 604, 401, 500)
    ]

    def calculate_aqi(self, concentration, breakpoints):
        # Round the concentration to the nearest 1 decimal place
        concentration = round(concentration, 1)

        for low_concentration, high_concentration, low_index, high_index in breakpoints:
            if low_concentration <= concentration < high_concentration:
                return int((high_index - low_index) / (high_concentration - low_concentration) * (concentration - low_concentration) + low_index)
        return None  # For concentrations not covered by any breakpoint

    def calculate(self, pm25, pm10):
        aqi_pm25 = self.calculate_aqi(pm25, self.EPA_USA_BREAKPOINTS_PM25)
        aqi_pm10 = self.calculate_aqi(pm10, self.EPA_USA_BREAKPOINTS_PM10)

        if aqi_pm25 is None:
            return aqi_pm10
        elif aqi_pm10 is None:
            return aqi_pm25
        else:
            return max(aqi_pm25, aqi_pm10)

# class Canada_Health(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class EU_EEA_Common(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class EU_European(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class UK_COMEAP_Daily(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class Australia_NEPM(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class India_CPCB_National(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class China_MEP(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class Mexico_IMECA(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class Japan_MOE(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class AIRKOREA_MOE_Common(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class Singapore_PSI(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# class Colombia_MADS(AQICalculator):
#     def calculate(self, pm25, pm10):
#         # Implementation here
#         pass

# Map algorithm names to their respective classes
ALGORITHMS = {
    'EPA_USA': {'class': EPA_USA, 'friendly_name': 'United States EPA'},
    # 'CANADA_HEALTH': {'class': Canada_Health, 'friendly_name': 'Canada Health'},
    # 'EU_EEA_COMMON': {'class': EU_EEA_Common, 'friendly_name': 'EU EEA Common'},
    # 'EU_EUROPEAN': {'class': EU_European, 'friendly_name': 'EU European'},
    # 'UK_COMEAP_DAILY': {'class': UK_COMEAP_Daily, 'friendly_name': 'UK COMEAP Daily'},
    # 'AUSTRALIA_NEPM': {'class': Australia_NEPM, 'friendly_name': 'Australia NEPM'},
    # 'INDIA_CPCB_NATIONAL': {'class': India_CPCB_National, 'friendly_name': 'India CPCB National'},
    # 'CHINA_MEP': {'class': China_MEP, 'friendly_name': 'China MEP'},
    # 'MEXICO_IMECA': {'class': Mexico_IMECA, 'friendly_name': 'Mexico IMECA'},
    # 'JAPAN_MOE': {'class': Japan_MOE, 'friendly_name': 'Japan MOE'},
    # 'AIRKOREA_MOE_COMMON': {'class': AIRKOREA_MOE_Common, 'friendly_name': 'AIRKOREA MOE Common'},
    # 'SINGAPORE_PSI': {'class': Singapore_PSI, 'friendly_name': 'Singapore PSI'},
    # 'COLOMBIA_MADS': {'class': Colombia_MADS, 'friendly_name': 'Colombia MADS'},
}

[ryan-lang]

One other thing... the Nowcast values will provide the best AQI, because that is a recency-weighted average, and better takes into account current conditions.

[ByteTheHand]

You are right on NowCast, but I really prefer real time data instead. I realize that AQI is actually a 24-hour exposure value, “real time AQI” isn’t a scientific reality, but for automations, NowCast is pretty useless.

I have automations that turn on and off filter fans and window fans based on inside and outside AQI. NowCast is designed to move slowly and is generally a couple of hours behind when things change rapidly. The last smoke incursion event here sent PM2.5 from near zero to over 50 ug/m3 in an hour or two. Then on the clear out day it went from real time AQI 145 to 11 in two hours. It took AirNow using NowCast about 6 hours to show “good” air quality. So NowCast isn’t something that I can use. I modified the code in my instance to report the 1-minute PM2.5 and then use that to know when I can turn window fans on or off and when inside filter fans need to be turned on and off.

I also like that you are pulling in the PM10 for a more accurate AQI, but in the 4 years I’ve been monitoring with AirLink, the dominant pollutant has always been PM2.5 with PM10 being far, far lower and from the EPA ozone is never the critical component either, so less critical to me at least.

I would love for the data to expose all of the different PM2.5 data elements. I also love that you have the code to make the actual AQI number available. That would be awesome.