Source code for opendrift.readers.operators.readerops

from types import LambdaType
from ..basereader import BaseReader

[docs] def none_or_cmp(a, b, cmp): if a is None: return b if b is None: return a return cmp(a, b)
[docs] class Combined(BaseReader): """ A combination of two readers. """ a: BaseReader b: BaseReader op: LambdaType def __init__(self, a, b, op): self.a = a self.b = b self.op = op self.variables = list(set(self.a.variables).intersection(self.b.variables)) self.start_time = none_or_cmp(self.a.start_time, self.b.start_time, max) self.end_time = none_or_cmp(self.a.end_time, self.b.end_time, min) self.xmin = -180 self.xmax = 180 self.ymin = -90 self.ymax = 90 = f'Combined({} | {})' self.proj4 = '+proj=latlong' super().__init__()
[docs] def covers_positions(self, lon, lat): return np.intersect1d(self.a.covers_positions(lon, lat), self.b.covers_positions(lon, lat))
[docs] def covers_time(self, time): return self.a.covers_time(time) and self.b.covers_time(time)
[docs] def get_variables_interpolated(self, variables, *args, **kwargs): assert set(variables).issubset(self.variables), f"{variables} is not subset of {self.variables}" env_a, env_profiles_a = self.a.get_variables_interpolated(variables, *args, **kwargs) env_b, env_profiles_b = self.b.get_variables_interpolated(variables, *args, **kwargs) variables = [ var for var in env_a.keys() if var not in ['x', 'y', 'z', 'time'] ] for var in variables: env_a[var] = self.op(env_a[var], env_b[var]) if env_profiles_a is not None: variables = [ var for var in env_profiles_a.keys() if var not in ['x', 'y', 'z', 'time'] ] for var in variables: env_profiles_a[var] = self.op(env_profiles_a[var], env_profiles_b[var]) return env_a, env_profiles_a