Source code for pi_weather.api.agg_models

"""Logic to merge 2 sensors data into one aggregate reading"""

from typing import Optional

from pi_weather.api.models import (
    LatestAggregateReading,
    LatestSensorReading,
    LatestSTCC4SensorReading,
)
from pi_weather.utils.logs import get_logger

logger = get_logger(__name__)




def get_latest_aggregate_reading(
    data_bmp390: LatestSensorReading | None = None,
    data_stcc4: LatestSTCC4SensorReading | None = None,
) -> Optional[LatestAggregateReading]:
    """
    Function to merge 2 inputs readings into an aggregate reading

    :param data_bmp390: Latest reading data for BMP390
    :type data_bmp390: LatestSensorReading
    :param data_stcc4: Latest reading data for STCC4
    :type data_stcc4: LatestSTCC4SensorReading
    :return: Agg readings
    :rtype: Optional[LatestAggregateReading]
    """
    if data_bmp390 is None:
        logger.error("Invalid input data for sensor BMP390")
        return None

    if data_stcc4 is None:
        logger.error("Invalid input data for sensor STCC4")
        return None

    logger.info("Merging aggregate reading data")
    data_agg: LatestAggregateReading = LatestAggregateReading(
        bmp390_event_ts=data_bmp390.event_ts,
        bmp390_sensor_ts=data_bmp390.sensor_ts,
        bmp390_pressure=data_bmp390.pressure,
        bmp390_temperature=data_bmp390.temperature,
        bmp390_altitude=data_bmp390.altitude,
        stcc4_event_ts=data_stcc4.event_ts,
        stcc4_sensor_status=data_stcc4.sensor_status,
        stcc4_temperature=data_stcc4.temperature,
        stcc4_humidity=data_stcc4.humidity,
        stcc4_co2_concentration=data_stcc4.co2_concentration,
        stcc4_air_quality=data_stcc4.air_quality,
    )
    return data_agg