Under the hood: Get Timeseries at Points

🚧

This is an advanced topic.

Understanding this content is not required for regular use of Eratos.

Below is the complete break-down of Get Timeseries at Points (Point over time)

Where self is a gridded data object initialized through:

from eratos.creds import AccessTokenCreds
from eratos.adapter import Adapter

ecreds = AccessTokenCreds(
  "YOUR_KEY_ID_GOES_HERE",
  "YOUR_KEY_SECRET_GOES_HERE"
)
eadapter = Adapter(ecreds)

# Request acccess to a dataset resource in Eratos
max_temperature_data = eadapter.Resource(ern='ern:e-pn.io:resource:eratos.blocks.silo.maxtemperature')

# Access the gridded dataset via the Gridded API:
gridded_max_temperature_data = max_temperature_data.data().gapi()

Function break-down

from shapely import wkt, geometry
from datetime import timezone
import datetime
import numpy as np

 def get_timeseries_at_points(self,var, point_list, startDate, endDate, time_stride = 1):
    
      # Check input variables are correct type 
      if type(point_list) is not list:
          raise TypeError('invalid type for point_list: %s' % type(point_list))
      if len(point_list) < 1:
          raise Exception('point_list is empty, please ensure there is at least one WKT point inside list')
      if type(var) is not str:
          raise TypeError('invalid type for var: %s' % type(var))
      if type(startDate) is not str:
          raise TypeError('invalid type for endDate: %s' % type(startDate))
      if type(endDate) is not str:
          raise TypeError('invalid type for endDate: %s' % type(endDate))  
    
     # Load Point strings as Shapely Points
      wkt_list = []
      for point in (point_list):
          loc_shape = wkt.loads(point)
          if type(loc_shape) is not geometry.Point:
              raise ValueError('value inside point_list should be a WKT point')
          loc= [loc_shape.y, loc_shape.x]
          wkt_list.append(loc)
        
      # date: Create a Unix timestamp in UTC timezone from the DD-MM-YYYY-MM formatted date string - e.g. "01-01-2022"
      unix_ts_utc_start = datetime.datetime.strptime(startDate, '%Y-%m-%d').replace(tzinfo=timezone.utc).timestamp()
      unix_ts_utc_end = datetime.datetime.strptime(endDate, '%Y-%m-%d').replace(tzinfo=timezone.utc).timestamp()
      times = self.get_subset_as_array('time')
      start_idx = np.where(times == unix_ts_utc_start)[0][0]
      end_idx = (np.where(times == unix_ts_utc_end)[0][0])+1
   
      data_query_array = self.get_point_slices(var, 'SPP', pts=wkt_list, starts=[start_idx], ends=[end_idx],strides =  [time_stride])
    
      return data_query_array