from abc import abstractmethod
import numpy as np
import logging
logger = logging.getLogger(__name__)
[docs]
class Reader:
[docs]
@abstractmethod
def variables(self):
"""
Get list of variables
"""
return []
[docs]
@abstractmethod
def read_grid(self, grid, var, t):
"""
Read variable for grid.
Args:
grid: grid to read variable for
var: list of strs, name of variables to read
t: datetime (UTC)
Returns:
v0, v1, ...
tuple with each variables specified in var with same shape as `grid`.
"""
pass
[docs]
class ConstantReader(Reader):
def __init__(self, consts):
"""
Args:
consts: dict, name of var: constant
"""
logger.info("constant reader for: %s" % ", ".join(consts.keys()))
self.consts = consts
super().__init__()
[docs]
def read_grid(self, grid, var, _):
for v in var:
assert v in self.consts, "missing variable %s" % v
return tuple(self.consts[v] * np.ones(grid.grid.shape) for v in var)
[docs]
def variables(self):
return list(self.consts.keys())
[docs]
@staticmethod
def new_xy(x=.5, y=.5):
return ConstantReader({
'x_sea_water_velocity': x,
'y_sea_water_velocity': y
})
[docs]
class OpendriftReader(Reader):
"""
Wrapper around an Opendrift reader.
"""
def __init__(self, reader):
"""
Args:
reader: opendrift reader
"""
logger.info("opendrift reader: %s (%s)" %
(reader.name, ", ".join(reader.variables)))
self.r = reader
super().__init__()
[docs]
def variables(self):
return self.r.variables
[docs]
def read_grid(self, grid, var, t):
logger.debug('reading %s for %s' % (var, t))
x, y = self.r.lonlat2xy(grid.lons.ravel(), grid.lats.ravel())
env, _ = self.r.get_variables_interpolated_xy(
variables=var,
time=t,
x=x,
y=y,
z=np.zeros(grid.grid.shape).ravel(),
rotate_to_proj=grid.srs)
u = tuple(env[v].reshape(grid.grid.shape).filled(fill_value=np.nan) for v in var)
for uu, vv in zip(u, var):
if np.isnan(uu).any():
logger.warning("nan's in %s" % vv)
return u