Merge branch 'dClimate:master' into master

Author: dmp267

dweather_client/aliases_and_units.py

@@ -26,6 +26,8 @@
     "fraction": u.dimensionless_unscaled,
     "percentage": u.pct,
     "mbar": u.def_unit('mbar', u.bar / 1000),
+    "m**3/s": u.m**3 / u.s,
+    "m^3/s": u.m**3 / u.s,
 }
 
 METRIC_TO_IMPERIAL = {
@@ -34,14 +36,17 @@
     u.deg_C: lambda q: q.to(imperial.deg_F, equivalencies=u.temperature()),
     u.K: lambda q: q.to(imperial.deg_F, equivalencies=u.temperature()),
     u.kg / u.m**2: lambda q: q.to(imperial.pound / imperial.ft ** 2),
-    u.m / u.s: lambda q: q.to(imperial.mile / u.hour)
+    u.m / u.s: lambda q: q.to(imperial.mile / u.hour),
+    u.m**3 / u.s: lambda q: q.to(imperial.yard**3 / u.s)
+
 }
 
 IMPERIAL_TO_METRIC = {
     imperial.inch: lambda q: q.to(u.mm),
     imperial.deg_F: lambda q: q.to(u.deg_C, equivalencies=u.temperature()),
     imperial.pound / imperial.ft ** 2: lambda q: q.to(u.kg / u.m**2),
-    imperial.mile / u.hour: lambda q: q.to(u.m / u.s)
+    imperial.mile / u.hour: lambda q: q.to(u.m / u.s),
+    imperial.yard**3 / u.s: lambda q: q.to(u.m**3 / u.s)
 }
 
 STATION_ALIASES_TO_COLUMNS = {
@@ -214,6 +219,8 @@ def lookup_station_alias(alias):
              'imperialize': lambda m: m.to(imperial.deg_F, equivalencies=u.temperature())},
     'WSF5': {'vectorize': lambda q: (q/10.0) * u.m/u.s,
              'imperialize': lambda q: q.to(imperial.mile/u.hour)},
+    'FLOWRATE': {'vectorize': lambda q: u.m**3/u.s,
+             'imperialize': lambda q: q.to(imperial.yard**3/u.s)},
 }
 
 parent_dir = os.path.dirname(os.path.abspath(__file__))

dweather_client/client.py

@@ -9,15 +9,16 @@
 from dweather_client.struct_utils import tupleify, convert_nans_to_none
 import datetime
 import pytz
-import csv, json
+import csv
+import json
 import inspect
 import numpy as np
 import pandas as pd
 from astropy import units as u
 from timezonefinder import TimezoneFinder
 from dweather_client import gridded_datasets
 from dweather_client.storms_datasets import IbtracsDataset, AtcfDataset, SimulatedStormsDataset
-from dweather_client.ipfs_queries import AustraliaBomStations, CedaBiomass, CmeStationsDataset, DutchStationsDataset, DwdStationsDataset, DwdHourlyStationsDataset, GlobalHourlyStationsDataset, JapanStations, StationDataset,\
+from dweather_client.ipfs_queries import AustraliaBomStations, CedaBiomass, CmeStationsDataset, DutchStationsDataset, DwdStationsDataset, DwdHourlyStationsDataset, GlobalHourlyStationsDataset, JapanStations, StationDataset, EauFranceDataset,\
     YieldDatasets, FsaIrrigationDataset, AemoPowerDataset, AemoGasDataset, AesoPowerDataset, ForecastDataset, AfrDataset, DroughtMonitor, CwvStations, SpeedwellStations, TeleconnectionsDataset, CsvStationDataset, StationForecastDataset
 from dweather_client.slice_utils import DateRangeRetriever, has_changed
 from dweather_client.ipfs_errors import *
@@ -246,6 +247,7 @@ def get_tropical_storms(
         min_lon=None,
         max_lat=None,
         max_lon=None,
+        as_of=None,
         ipfs_timeout=None):
     """
     return:
@@ -288,11 +290,11 @@ def get_tropical_storms(
 
     with cm as storm_getter:
         if radius:
-            return storm_getter.get_data(basin, radius=radius, lat=lat, lon=lon)
+            return storm_getter.get_data(basin, radius=radius, lat=lat, lon=lon, as_of=as_of)
         elif min_lat:
-            return storm_getter.get_data(basin, min_lat=min_lat, min_lon=min_lon, max_lat=max_lat, max_lon=max_lon)
+            return storm_getter.get_data(basin, min_lat=min_lat, min_lon=min_lon, max_lat=max_lat, max_lon=max_lon, as_of=as_of)
         else:
-            return storm_getter.get_data(basin)
+            return storm_getter.get_data(basin, as_of=as_of)
 
 
 def get_station_history(
@@ -485,6 +487,7 @@ def get_hourly_station_history(dataset, station_id, weather_variable, use_imperi
         v) for k, v in final_resp_series.to_dict().items()}
     return result
 
+
 def get_csv_station_history(dataset, station_id, weather_variable, use_imperial_units=True, desired_units=None, ipfs_timeout=None):
     """
     This is almost an exact copy of get_hourly_station_history
@@ -543,14 +546,26 @@ def get_csv_station_history(dataset, station_id, weather_variable, use_imperial_
             "Invalid weather variable for this station")
 
     try:
-        if dataset == "inmet_brazil-hourly":
+        # RawSet style where we only want the most recent file
+        if dataset in ["inmet_brazil-hourly"]:
             with CsvStationDataset(dataset=dataset, ipfs_timeout=ipfs_timeout) as dataset_obj:
-                csv_text = dataset_obj.get_data(station_id, weather_variable)
+                csv_text_list = [dataset_obj.get_data(
+                    station_id, weather_variable)]
+        # ClimateSet style where we need the entire linked list history
+        elif dataset in ["ne_iso-hourly"]:
+            with CsvStationDataset(dataset=dataset, ipfs_timeout=ipfs_timeout) as dataset_obj:
+                csv_text_list = dataset_obj.get_data_recursive(
+                    station_id, weather_variable)
         else:
             raise DatasetError("No such dataset in dClimate")
     except ipfshttpclient.exceptions.ErrorResponse:
         raise StationNotFoundError("Invalid station ID for dataset")
-    df = pd.read_csv(StringIO(csv_text))
+
+    # concat together all retrieved station csv texts
+    dfs = []
+    for csv_text in csv_text_list:
+        dfs.append(pd.read_csv(StringIO(csv_text)))
+    df = pd.concat(dfs, ignore_index=True)
     str_resp_series = df[column_name].astype(str)
     df = df.set_index("dt")
     if desired_units:
@@ -579,18 +594,21 @@ def get_csv_station_history(dataset, station_id, weather_variable, use_imperial_
         v) for k, v in final_resp_series.to_dict().items()}
     return result
 
+
 def get_station_forecast_history(dataset, station_id, forecast_date, desired_units=None, ipfs_timeout=None):
-    try: 
+    try:
         with StationForecastDataset(dataset, ipfs_timeout=ipfs_timeout) as dataset_obj:
             csv_text = dataset_obj.get_data(station_id, forecast_date)
             history = {}
             reader = csv.reader(csv_text.split('\n'))
             headers = next(reader)
             date_col = headers.index('DATE')
             try:  # Make sure weather variable is correct.
-                data_col = headers.index("SETT") #at the moment the only variable is "SETT"
+                # at the moment the only variable is "SETT"
+                data_col = headers.index("SETT")
             except ValueError:
-                raise WeatherVariableNotFoundError("Invalid weather variable for this station")
+                raise WeatherVariableNotFoundError(
+                    "Invalid weather variable for this station")
             for row in reader:
                 try:
                     if not row:
@@ -599,11 +617,12 @@ def get_station_forecast_history(dataset, station_id, forecast_date, desired_uni
                         row[date_col], "%Y-%m-%d").date()] = float(row[data_col])
                 except ValueError:
                     history[datetime.datetime.strptime(
-                        row[date_col], "%Y-%m-%d").date()] = row[data_col]        
+                        row[date_col], "%Y-%m-%d").date()] = row[data_col]
             return history
     except ipfshttpclient.exceptions.ErrorResponse:
         raise StationNotFoundError("Invalid station ID for dataset")
 
+
 def get_station_forecast_stations(dataset, forecast_date, desired_units=None, ipfs_timeout=None):
     with StationForecastDataset(dataset, ipfs_timeout=ipfs_timeout) as dataset_obj:
         csv_text = dataset_obj.get_stations(forecast_date)
@@ -850,13 +869,13 @@ def has_dataset_updated(dataset, slices, as_of, ipfs_timeout=None):
 
 def get_teleconnections_history(weather_variable, ipfs_timeout=None):
     with TeleconnectionsDataset(ipfs_timeout=ipfs_timeout) as dataset_obj:
-        csv_text = dataset_obj.get_data()
+        csv_text = dataset_obj.get_data(weather_variable)
         history = {}
         reader = csv.reader(csv_text.split('\n'))
         headers = next(reader)
         date_col = headers.index('DATE')
-        try:  # Make sure weather variable is correct.
-            data_col = headers.index(weather_variable)
+        try:
+            data_col=headers.index("value")
         except ValueError:
             raise WeatherVariableNotFoundError(
                 "Invalid weather variable for this station")
@@ -874,3 +893,40 @@ def get_teleconnections_history(weather_variable, ipfs_timeout=None):
                 history[datetime.datetime.strptime(
                     row[date_col], "%Y-%m-%d").date()] = row[data_col]
         return history
+
+
+def get_eaufrance_history(station, weather_variable, use_imperial_units=False, desired_units=None, ipfs_timeout=None):
+    try:
+        with EauFranceDataset(ipfs_timeout=ipfs_timeout) as dataset_obj:
+            csv_text = dataset_obj.get_data(station)
+            df = pd.read_csv(StringIO(csv_text))
+            df = df.set_index("DATE")
+            original_units = "m^3/s"
+            if desired_units:
+                converter, dweather_unit = get_unit_converter_no_aliases(
+                    original_units, desired_units)
+            else:
+                converter, dweather_unit = get_unit_converter(
+                    original_units, use_imperial_units)
+            if converter:
+                try:
+                    converted_resp_series = pd.Series(
+                        converter(df[weather_variable].values*dweather_unit), index=df.index)
+                except ValueError:
+                    raise UnitError(
+                        f"Specified unit is incompatible with original, original units are {original_units} and requested units are {desired_units}")
+                if desired_units is not None:
+                    rounded_resp_array = np.vectorize(rounding_formula_temperature)(
+                        str_resp_series, converted_resp_series)
+                    final_resp_series = pd.Series(
+                        rounded_resp_array * converted_resp_series.values.unit, index=df.index)
+                else:
+                    final_resp_series = converted_resp_series
+            else:
+                final_resp_series = pd.Series(
+                    df[weather_variable].values*dweather_unit, index=df.index)
+            result = {datetime.date.fromisoformat(k): convert_nans_to_none(
+                v) for k, v in final_resp_series.to_dict().items()}
+        return result
+    except ipfshttpclient.exceptions.ErrorResponse:
+        raise StationNotFoundError("Invalid station ID for dataset")

dweather_client/ipfs_queries.py

@@ -660,13 +660,34 @@ def __init__(self, dataset, ipfs_timeout=None):
         super().__init__(ipfs_timeout=ipfs_timeout)
         self._dataset = dataset
 
+    def get_hashes(self):
+        """
+        return: list of all hashes in dataset
+        """
+        hashes = self.traverse_ll(self.head, self.as_of)
+        return list(hashes)
+
     def get_data(self, station, weather_variable=None):
         # only some stations need weather variable
         # so this is an optional arg
         super().get_data()
         file_name = f"{self.head}/{station}.csv"
         return self.get_file_object(file_name).read().decode("utf-8")
 
+    def get_data_recursive(self, station, weather_variable=None):
+        # only some stations need weather variable
+        # so this is an optional arg
+        super().get_data()
+        # get all hashes and then effectively just use get_data
+        # recursively to get a full list of csvs
+        hashes = self.get_hashes()
+        csv_text_list = []
+        for hash_ in hashes:
+            file_name = f"{hash_}/{station}.csv"
+            csv_text_list.append(self.get_file_object(
+                file_name).read().decode("utf-8"))
+        return csv_text_list
+
 
 class YieldDatasets(IpfsDataset):
     """
@@ -1114,7 +1135,8 @@ def get_full_date_range_from_metadata(self, h):
         return: list of [start_time, end_time]
         """
         metadata = self.get_metadata(h)
-        str_dates = (metadata["api documentation"]["full date range"][0], metadata["api documentation"]["full date range"][1])
+        str_dates = (metadata["api documentation"]["full date range"]
+                     [0], metadata["api documentation"]["full date range"][1])
         return [datetime.datetime.fromisoformat(dt).date() for dt in str_dates]
 
     def get_relevant_hash(self, forecast_date):
@@ -1144,12 +1166,14 @@ def get_relevant_hash(self, forecast_date):
             prev_date_range = [datetime.date.fromisoformat(
                 d) for d in prev_metadata["date range"]]
             if prev_date_range[0] <= forecast_date <= prev_date_range[1]:
-                print(f"User requested {forecast_date}, returning data for date range {prev_date_range} from hash {prev_hash}") # NOTE for testing, TODO remove afterwards
                 return prev_hash
-            prev_hash = prev_metadata['previous hash'] # iterate backwards in the link list one step
+            # iterate backwards in the link list one step
+            prev_hash = prev_metadata['previous hash']
 
         # If this script runs to the end without returning anything or an error, the forecast date must fall in a hole in the data
-        raise DateOutOfRangeError("forecast date unavailable due to holes in data") # NOTE only returns if there are holes in the data
+        # NOTE only returns if there are holes in the data
+        raise DateOutOfRangeError(
+            "forecast date unavailable due to holes in data")
 
     def get_weather_dict(self, forecast_date, ipfs_hash, lat, lon):
         """
@@ -1194,6 +1218,7 @@ def get_data(self, lat, lon, forecast_date):
 
         return (float(ret_lat), float(ret_lon)), pd.Series(weather_dict)
 
+
 class StationForecastDataset(ForecastDataset):
     """
     Instantiable class for pulling in station data that is also forecast data. 
@@ -1202,11 +1227,11 @@ class StationForecastDataset(ForecastDataset):
     @property
     def dataset(self):
         return self._dataset
-    
+
     def __init__(self, dataset, **kwargs):
         super().__init__(dataset, 1)
         self.head = get_heads()[self.dataset]
-    
+
     def get_data(self, station, forecast_date):
         relevant_hash = self.get_relevant_hash(forecast_date)
         return self.get_file_object(f"{relevant_hash}/{station}.csv").read().decode("utf-8")
@@ -1215,6 +1240,7 @@ def get_stations(self, forecast_date):
         relevant_hash = self.get_relevant_hash(forecast_date)
         return self.get_file_object(f"{relevant_hash}/stations.json").read().decode("utf-8")
 
+
 class TeleconnectionsDataset(IpfsDataset):
     """
     Instantiable class used for pulling in el nino teleconnections data 
@@ -1224,9 +1250,25 @@ class TeleconnectionsDataset(IpfsDataset):
     def __init__(self, ipfs_timeout=None):
         super().__init__(ipfs_timeout=ipfs_timeout)
 
-    def get_data(self):
+    def get_data(self, station):
+        super().get_data()
+        metadata = self.get_metadata(self.head)
+        
+        file_name = f"{self.head}/{station}.csv"
+        return self.get_file_object(file_name).read().decode("utf-8")
+
+
+class EauFranceDataset(IpfsDataset):
+    """
+    Instantiable class used for pulling in el nino teleconnections data 
+    """
+    dataset = "EauFrance-daily"
+
+    def __init__(self, ipfs_timeout=None):
+        super().__init__(ipfs_timeout=ipfs_timeout)
+
+    def get_data(self, station):
         super().get_data()
         metadata = self.get_metadata(self.head)
-        year_month = metadata["time generated"][:7]
-        file_name = f"{self.head}/teleconnections_{year_month}.csv"
+        file_name = f"{self.head}/{station}.csv"
         return self.get_file_object(file_name).read().decode("utf-8")

dweather_client/storms_datasets.py

@@ -1,11 +1,14 @@
 import gzip
 import json
+import datetime
 from abc import abstractmethod
 
 import pandas as pd
 
 from dweather_client.df_utils import boxed_storms, nearby_storms
 from dweather_client.ipfs_queries import IpfsDataset
+from dweather_client.ipfs_errors import *
+
 
 def process_df(input_df, **kwargs):
     if {'radius', 'lat', 'lon'}.issubset(kwargs.keys()):
@@ -23,7 +26,8 @@ def get_data(self, basin, **kwargs):
         if basin not in {'NI', 'SI', 'NA', 'EP', 'WP', 'SP', 'SA'}:
             raise ValueError("Invalid basin ID")
         super().get_data()
-        file_obj = self.get_file_object(f"{self.head}/ibtracs-{basin}.csv.gz")
+        ipfs_hash = self.get_relevant_hash(kwargs["as_of"])
+        file_obj = self.get_file_object(f"{ipfs_hash}/ibtracs-{basin}.csv.gz")
         df = pd.read_csv(
             file_obj, na_values=["", " "], keep_default_na=False, low_memory=False, compression="gzip"
         )
@@ -40,6 +44,34 @@ def get_data(self, basin, **kwargs):
 
         return processed_df
 
+    def get_relevant_hash(self, as_of_date):
+        """
+        return the ipfs hash required to pull in data for a forecast date
+        """
+        cur_hash = self.head
+        if as_of_date == None:
+            return cur_hash
+        cur_metadata = self.get_metadata(cur_hash)
+        # This routine is agnostic to the order of data contained in the hashes (at a cost of inefficiency) -- if the data contains the forecast date, it WILL be found, eventually
+        most_recent_date = datetime.datetime.fromisoformat(cur_metadata["time generated"]).date()
+        if as_of_date >= most_recent_date:
+            return cur_hash
+        prev_hash = cur_metadata['previous hash']
+        while prev_hash is not None:
+            prev_metadata = self.get_metadata(prev_hash)
+            prev_date = datetime.datetime.fromisoformat(prev_metadata["time generated"]).date()
+            if prev_date <= as_of_date <= most_recent_date:
+                return prev_hash
+            # iterate backwards in the link list one step
+            try:
+                prev_hash = prev_metadata['previous hash']
+                most_recent_date = prev_date
+            except KeyError:
+                # Because we added the as_of after a while to this ETL prev_hash won't be 'None' we'll just run into an exception
+                raise DateOutOfRangeError(
+                    "as_of data is earlier than earliest available hash")
+
+
 class AtcfDataset(IpfsDataset):
     dataset = "atcf_btk-seasonal"