/
classes.py
117 lines (100 loc) · 3.72 KB
/
classes.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
from typing import Any, Callable, Dict, List, OrderedDict, Sequence
from cffi import FFI
from refcount.interop import DeletableCffiNativeHandle, CffiData, CffiWrapperFactory
from cinterop.cffi.marshal import ConvertibleToTimestamp
import uchronia.wrap.uchronia_wrap_generated as uwg
class TimeSeriesProvider(DeletableCffiNativeHandle):
def __init__(
self,
handle: CffiData,
release_native: Callable[[CffiData], None],
type_id: str = None,
prior_ref_count: int = 0,
):
super(TimeSeriesProvider, self).__init__(
handle, release_native, type_id, prior_ref_count
)
class TimeSeriesMixin:
def __init__(self):
super(TimeSeriesMixin, self).__init__(
)
def get_item(self, i:int, convert_to_xr=True):
import uchronia.uchronia_time_series as uts
return uts.get_item(self, i, convert_to_xr)
class EnsembleForecastTimeSeries(DeletableCffiNativeHandle, TimeSeriesMixin):
def __init__(
self,
handle: CffiData,
release_native: Callable[[CffiData], None],
type_id: str = None,
prior_ref_count: int = 0,
):
super(EnsembleForecastTimeSeries, self).__init__(
handle, release_native, type_id, prior_ref_count
)
class EnsembleTimeSeries(DeletableCffiNativeHandle, TimeSeriesMixin):
def __init__(
self,
handle: CffiData,
release_native: Callable[[CffiData], None],
type_id: str = None,
prior_ref_count: int = 0,
):
super(EnsembleTimeSeries, self).__init__(
handle, release_native, type_id, prior_ref_count
)
class TimeSeriesLibrary(DeletableCffiNativeHandle):
def __init__(
self,
handle: CffiData,
release_native: Callable[[CffiData], None],
type_id: str = None,
prior_ref_count: int = 0,
):
super(TimeSeriesLibrary, self).__init__(
handle, release_native, type_id, prior_ref_count
)
class TimeSeries(DeletableCffiNativeHandle, TimeSeriesMixin):
def __init__(
self,
handle: CffiData,
release_native: Callable[[CffiData], None],
type_id: str = None,
prior_ref_count: int = 0,
):
super(TimeSeries, self).__init__(
handle, release_native, type_id, prior_ref_count
)
class EnsemblePtrTimeSeries(DeletableCffiNativeHandle):
def __init__(
self,
handle: CffiData,
release_native: Callable[[CffiData], None],
type_id: str = None,
prior_ref_count: int = 0,
):
super(EnsemblePtrTimeSeries, self).__init__(
handle, release_native, type_id, prior_ref_count
)
_api_type_wrapper = {
# "DATATYPES_TIME_SERIES_DOUBLE_PTR": None,
"DATATYPES_ENSEMBLE_TIME_SERIES_DOUBLE_PTR": EnsembleTimeSeries,
"ENSEMBLE_DATA_SET_PTR": TimeSeriesLibrary,
"ENSEMBLE_FORECAST_TIME_SERIES_PTR": EnsembleForecastTimeSeries,
"TIME_SERIES_PTR": TimeSeries,
# "ENSEMBLE_TIME_SERIES_PTR ": None,
"ENSEMBLE_PTR_TIME_SERIES_PTR": EnsemblePtrTimeSeries,
"TIME_SERIES_PROVIDER_PTR": TimeSeriesProvider,
# "TS_GEOMETRY_PTR": None,
# "DATE_TIME_INFO_PTR": None,
# "MULTI_REGULAR_SERIES_STRUCT_PTR": None,
}
_STRICT_WRAPPING = False
_wrapper_factory = CffiWrapperFactory(_api_type_wrapper, _STRICT_WRAPPING)
def _create_wrapper(obj: Any, type_id: str, release_native: Callable):
return _wrapper_factory.create_wrapper(obj, type_id, release_native)
def wrap_cffi_native_handle(obj: Any, type_id: str, release_native: Callable):
if not isinstance(obj, FFI.CData):
return obj
return _create_wrapper(obj, type_id, release_native)
uwg.set_wrap_cffi_native_handle(wrapper_function=wrap_cffi_native_handle)