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
Updated 8 days ago