vAirify/air-quality-backend/api/src/mappers/measurements_mapper.py at 6b3bc4a8d993bead16a90df8cb5d64aa7185a5be · ECMWFCode4Earth/vAirify · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
from datetime import UTC
from typing import List

from src.types import MeasurementDto, MeasurementSummaryDto
from shared.src.aqi.calculator import get_pollutant_index_level, get_overall_aqi_level
from shared.src.aqi.pollutant_type import PollutantType
from shared.src.database.in_situ import InSituMeasurement, InSituAveragedMeasurement


def map_measurement(measurement: InSituMeasurement) -> MeasurementDto:
    return {
        "measurement_date": measurement["measurement_date"].astimezone(UTC),
        "location_type": measurement["location_type"],
        "location_name": measurement["name"],
        "location": {
            "longitude": measurement["location"]["coordinates"][0],
            "latitude": measurement["location"]["coordinates"][1],
        },
        **{
            pollutant_type.value: measurement[pollutant_type.literal()]["value"]
            for pollutant_type in PollutantType
            if pollutant_type.value in measurement
        },
        "api_source": measurement["api_source"],
        "entity": measurement["metadata"]["entity"],
        "sensor_type": measurement["metadata"]["sensor_type"],
        "site_name": measurement["location_name"],
    }


def map_measurements(measurements: list[InSituMeasurement]) -> list[MeasurementDto]:
    return list(map(map_measurement, measurements))


def map_summarized_measurement(
    measurement: InSituAveragedMeasurement,
) -> MeasurementSummaryDto:
    pollutant_data = {}
    mean_aqi_values = []
    for pollutant_type in PollutantType:
        pollutant_value = pollutant_type.literal()
        avg_value = measurement[pollutant_value]["mean"]
        if avg_value is not None:
            aqi = get_pollutant_index_level(
                avg_value,
                pollutant_type,
            )
            if pollutant_value not in pollutant_data:
                pollutant_data[pollutant_value] = {}
            pollutant_data[pollutant_value]["mean"] = {
                "aqi_level": aqi,
                "value": avg_value,
            }
            mean_aqi_values.append(aqi)

    return {
        "measurement_base_time": measurement["measurement_base_time"],
        "location_type": measurement["location_type"],
        "location_name": measurement["name"],
        "overall_aqi_level": {"mean": get_overall_aqi_level(mean_aqi_values)},
        **pollutant_data,
    }


def map_summarized_measurements(
    averages: List[InSituAveragedMeasurement],
) -> List[MeasurementSummaryDto]:
    return list(map(map_summarized_measurement, averages))


def map_measurement_counts(measurements):
    """Map measurements to counts by city and pollutant, including unique location counts per pollutant"""
    counts = {}
    location_sets = {}  # Track unique locations per city and pollutant

    for measurement in measurements:
        city = measurement["name"]
        if city not in counts:
            counts[city] = {
                "so2": 0,
                "no2": 0,
                "o3": 0,
                "pm10": 0,
                "pm2_5": 0,
                "so2_locations": 0,
                "no2_locations": 0,
                "o3_locations": 0,
                "pm10_locations": 0,
                "pm2_5_locations": 0,
            }
            location_sets[city] = {
                "so2": set(),
                "no2": set(),
                "o3": set(),
                "pm10": set(),
                "pm2_5": set(),
            }

        location_name = measurement["location_name"]

        # Count measurements and track locations per pollutant
        if "so2" in measurement and measurement["so2"] is not None:
            counts[city]["so2"] += 1
            location_sets[city]["so2"].add(location_name)

        if "no2" in measurement and measurement["no2"] is not None:
            counts[city]["no2"] += 1
            location_sets[city]["no2"].add(location_name)

        if "o3" in measurement and measurement["o3"] is not None:
            counts[city]["o3"] += 1
            location_sets[city]["o3"].add(location_name)

        if "pm10" in measurement and measurement["pm10"] is not None:
            counts[city]["pm10"] += 1
            location_sets[city]["pm10"].add(location_name)

        if "pm2_5" in measurement and measurement["pm2_5"] is not None:
            counts[city]["pm2_5"] += 1
            location_sets[city]["pm2_5"].add(location_name)

    # Add location counts per pollutant to the final output
    for city in counts:
        for pollutant in ["so2", "no2", "o3", "pm10", "pm2_5"]:
            counts[city][f"{pollutant}_locations"] = len(location_sets[city][pollutant])

    return counts