#!/usr/bin/python3 """ Copyright (C) 2023 NIBIO This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see <https://www.gnu.org/licenses/>. """ """ Util methods for data manipulation Author: Tor-Einar Skog <tor-einar.skog@nibio.no> """ import numpy as np import pandas as pd from pandas import DataFrame import json from .entities import * def get_dataframe_from_weather_observations(weather_observations: list, timezone:datetime.tzinfo) -> DataFrame: """ Create a Pandas data frame with timeseries from VIPS weather observations """ # Analyze the input data # Which weather parameters are included # Get first and last time_measured # What's the most specific log interval (hour? day?) params = set() start_date = None end_date = None min_log_interval = None for obs in weather_observations: start_date = obs.timeMeasured if start_date is None else obs.timeMeasured if obs.timeMeasured < start_date else start_date end_date = obs.timeMeasured if end_date is None else obs.timeMeasured if obs.timeMeasured > end_date else end_date params.add(obs.elementMeasurementTypeId) if min_log_interval is None: min_log_interval = obs.logIntervalId elif min_log_interval == WeatherObservation.LOG_INTERVAL_ID_1D and obs.logIntervalId == WeatherObservation.LOG_INTERVAL_ID_1H: min_log_interval = WeatherObservation.LOG_INTERVAL_ID_1H # Start date must be converted to the given timezone tz_start_date = start_date.astimezone(timezone) # What's the frequency? Hourly? Daily? TODO: Support other log intervals freq = "H" if min_log_interval == 1 else "D" # This creates a datetime_range in the given timezone datetime_range = pd.date_range(tz_start_date,periods = len(weather_observations),freq=freq,tz=timezone) # Create the dataframe with the correct columns df = pd.DataFrame(index=datetime_range, columns=list(params)) # Loop through the observations again, adding values to the frame for obs in weather_observations: df.at[obs.timeMeasured, obs.elementMeasurementTypeId] = obs.value return df def get_result_list_from_dataframe(df: DataFrame): """ Create a list of VIPS Result objects from a dataframe Requires timeseries as index and a column named WARNING_STATUS with INT values ranging from 0-4 (see entities.Result class) """ result_list = [] dictframe = df.to_dict(orient="index") for timestamp in dictframe: values = dictframe[timestamp] result_list.append(Result( valid_time_start = timestamp, # valid_time_start valid_time_end = None, warning_status = values.pop("WARNING_STATUS"), all_values = json.dumps(values) )) return result_list def get_weather_observations_from_json(weather_data_raw: str) -> list: """ Convert a raw json string with VIPS of weather data to a list of WeatherObservation objects """ obs_json_list = [] #print(weather_data_raw) weather_data_json = json.loads(weather_data_raw) return get_weather_observations_from_json_list(weather_data_json) def get_weather_observations_from_json_list(weather_data: list) -> list: """ Convert a json list of VIPS of weather data to a list of WeatherObservation objects """ retval = [] for node in weather_data: if isinstance(node, WeatherObservation): retval.append(node) else: retval.append(WeatherObservation(**node)) return retval def get_temp_adjusted_for_base(temp: float, base_temp = 0.0) -> float: adjusted = temp - base_temp return max(0.0, adjusted)