ntv-numpy.ntv_numpy.xdataset

  1# -*- coding: utf-8 -*-
  2"""
  3Created on Thu Mar  7 09:56:11 2024
  4
  5@author: a lab in the Air
  6"""
  7from abc import ABC, abstractmethod
  8import json
  9from json_ntv import Ntv
 10from ntv_numpy.ndarray import Nutil
 11from ntv_numpy.xndarray import Xndarray
 12from ntv_numpy.xconnector import XarrayConnec, ScippConnec, AstropyNDDataConnec
 13
 14
 15class XdatasetCategory(ABC):
 16    ''' category of Xndarray (dynamic tuple of full_name) - see Xdataset docstring'''
 17
 18    xnd: list = NotImplemented
 19    names: list = NotImplemented
 20
 21    @abstractmethod
 22    def dims(self, var, json_name=False):
 23        '''method defined in Xdataset class'''
 24
 25    @property
 26    def global_vars(self):
 27        '''return a tuple of namedarrays or variable Xndarray full_name'''
 28        return tuple(sorted(nda for nda in self.namedarrays + self.variables))
 29
 30    @property
 31    def data_arrays(self):
 32        '''return a tuple of data_arrays Xndarray full_name'''
 33        return tuple(sorted(nda for nda in self.namedarrays if not nda in self.dimensions))
 34
 35    @property
 36    def dimensions(self):
 37        '''return a tuple of dimensions Xndarray full_name'''
 38        dimable = []
 39        for var in self.variables:
 40            dimable += self.dims(var)
 41        return tuple(sorted(set(nda for nda in dimable if nda in self.namedarrays)))
 42
 43    @property
 44    def coordinates(self):
 45        '''return a tuple of coordinates Xndarray full_name'''
 46        dims = set(self.dimensions)
 47        if not dims:
 48            return ()
 49        return tuple(sorted(set(xnda.name for xnda in self.xnd
 50                                if xnda.xtype == 'variable' and set(xnda.links) != dims)))
 51
 52    @property
 53    def data_vars(self):
 54        '''return a tuple of data_vars Xndarray full_name'''
 55        dims = set(self.dimensions)
 56        if not dims:
 57            return self.variables
 58        return tuple(sorted(xnda.name for xnda in self.xnd
 59                            if xnda.xtype == 'variable' and set(xnda.links) == dims))
 60
 61    @property
 62    def namedarrays(self):
 63        '''return a tuple of namedarray Xndarray full_name'''
 64        return tuple(sorted(xnda.name for xnda in self.xnd if xnda.xtype == 'namedarray'))
 65
 66    @property
 67    def variables(self):
 68        '''return a tuple of variables Xndarray full_name'''
 69        return tuple(sorted(xnda.name for xnda in self.xnd if xnda.xtype == 'variable'))
 70
 71    @property
 72    def undef_vars(self):
 73        '''return a tuple of variables Xndarray full_name with inconsistent shape'''
 74        return tuple(sorted([var for var in self.variables if self[var].shape !=
 75                             [len(self[dim]) for dim in self.dims(var)]]))
 76
 77    @property
 78    def undef_links(self):
 79        '''return a tuple of variables Xndarray full_name with inconsistent links'''
 80        return tuple(sorted([link for var in self.variables for link in self[var].links
 81                             if not link in self.names]))
 82
 83    @property
 84    def masks(self):
 85        '''return a tuple of additional Xndarray full_name with boolean ntv_type'''
 86        return tuple(sorted([xnda.full_name for xnda in self.xnd
 87                             if xnda.xtype == 'additional' and xnda.ntv_type == 'boolean']))
 88
 89    @property
 90    def data_add(self):
 91        '''return a tuple of additional Xndarray full_name with not boolean ntv_type'''
 92        return tuple(sorted([xnda.full_name for xnda in self.xnd
 93                             if xnda.xtype == 'additional' and xnda.ntv_type != 'boolean']))
 94
 95    @property
 96    def metadata(self):
 97        '''return a tuple of metadata Xndarray full_name'''
 98        return tuple(sorted(xnda.name for xnda in self.xnd if xnda.xtype == 'metadata'))
 99
100    @property
101    def additionals(self):
102        '''return a tuple of additionals Xndarray full_name'''
103        return tuple(sorted(xnda.full_name for xnda in self.xnd if xnda.xtype == 'additional'))
104
105    def var_group(self, name):
106        '''return a tuple of Xndarray full_name with the same name'''
107        return tuple(sorted(xnda.full_name for xnda in self.xnd if xnda.name == name))
108
109    def add_group(self, add_name):
110        '''return a tuple of Xndarray full_name with the same add_name'''
111        return tuple(sorted(xnda.full_name for xnda in self.xnd if xnda.add_name == add_name))
112
113
114class XdatasetInterface(ABC):
115    ''' Xdataset interface - see Xdataset docstring'''
116
117    name: str = NotImplemented
118    xnd: list = NotImplemented
119
120    @staticmethod
121    def read_json(jsn, **kwargs):
122        ''' convert json data into a Xdataset.
123
124        *Parameters*
125
126        - **convert** : boolean (default True) - If True, convert json data with
127        non Numpy ntv_type into Xndarray with python type
128        '''
129        option = {'convert': True} | kwargs
130        jso = json.loads(jsn) if isinstance(jsn, str) else jsn
131        value, name = Ntv.decode_json(jso)[:2]
132
133        xnd = [Xndarray.read_json({key: val}, **option)
134               for key, val in value.items()]
135        return Xdataset(xnd, name)
136
137    def to_json(self, **kwargs):
138        ''' convert a Xdataset into json-value.
139
140        *Parameters*
141
142        - **encoded** : Boolean (default False) - json value if False else json text
143        - **header** : Boolean (default True) - including 'xdataset' type
144        - **notype** : list of Boolean (default list of None) - including data type if False
145        - **novalue** : Boolean (default False) - including value if False
146        - **noshape** : Boolean (default True) - if True, without shape if dim < 1
147        - **format** : list of string (default list of 'full') - representation
148        format of the ndarray,
149        '''
150        notype = kwargs['notype'] if ('notype' in kwargs and isinstance(kwargs['notype'], list) and
151                                      len(kwargs['notype']) == len(self)) else [False] * len(self)
152        forma = kwargs['format'] if ('format' in kwargs and isinstance(kwargs['format'], list) and
153                                     len(kwargs['format']) == len(self)) else ['full'] * len(self)
154        noshape = kwargs.get('noshape', True)
155        dic_xnd = {}
156        for xna, notyp, forma in zip(self.xnd, notype, forma):
157            # not_shape = True if len(xna.links) == 1 else noshape
158            dic_xnd |= xna.to_json(notype=notyp, novalue=kwargs.get('novalue', False),
159                                   noshape=noshape, format=forma, header=False)
160        return Nutil.json_ntv(self.name, 'xdataset', dic_xnd,
161                              header=kwargs.get('header', True),
162                              encoded=kwargs.get('encoded', False))
163
164    def to_xarray(self, **kwargs):
165        '''return a DataArray or a Dataset from a Xdataset
166
167        *Parameters*
168
169        - **dataset** : Boolean (default True) - if False and a single data_var, return a DataArray
170        '''
171        return XarrayConnec.xexport(self, **kwargs)
172
173    @staticmethod
174    def from_xarray(xar, **kwargs):
175        '''return a Xdataset from a DataArray or a Dataset'''
176        return XarrayConnec.ximport(xar, Xdataset, **kwargs)
177
178    def to_scipp(self, **kwargs):
179        '''return a sc.DataArray or a sc.Dataset from a Xdataset
180
181        *Parameters*
182
183        - **dataset** : Boolean (default True) - if False and a single data_var,
184        return a DataArray
185        - **datagroup** : Boolean (default True) - if True return a DataGroup with
186        metadata and data_arrays
187        - **ntv_type** : Boolean (default True) - if True add ntv-type to the name
188        '''
189        return ScippConnec.xexport(self, **kwargs)
190
191    @staticmethod
192    def from_scipp(sci, **kwargs):
193        '''return a Xdataset from a scipp object DataArray, Dataset or DataGroup'''
194        return ScippConnec.ximport(sci, Xdataset, **kwargs)
195
196    def to_nddata(self, **kwargs):
197        '''return a NDData from a Xdataset'''
198        return AstropyNDDataConnec.xexport(self, **kwargs)
199
200    @staticmethod
201    def from_nddata(ndd, **kwargs):
202        '''return a Xdataset from a NDData'''
203        return AstropyNDDataConnec.ximport(ndd, Xdataset, **kwargs)
204
205
206class Xdataset(XdatasetCategory, XdatasetInterface):
207    ''' Representation of a multidimensional Dataset
208
209    *Attributes :*
210    - **name** :  String - name of the Xdataset
211    - **xnd**:   list of Xndarray
212
213    *dynamic values (@property)*
214    - `xtype`
215    - `validity`
216    - `dic_xnd`
217    - `partition`
218    - `info`
219
220    *methods*
221    - `parent`
222    - `dims`
223    - `shape_dims`
224    - `to_canonical`
225    - `to_ndarray`
226
227    *XdatasetCategory (@property)*
228    - `names`
229    - `global_vars`
230    - `data_arrays`
231    - `dimensions`
232    - `coordinates`
233    - `data_vars`
234    - `namedarrays`
235    - `variables`
236    - `undef_vars`
237    - `undef_links`
238    - `masks`
239    - `data_add`
240    - `metadata`
241    - `additionals`
242    - `var_group`
243    - `add_group`
244
245    *XdatasetInterface methods *
246    - `read_json` (static)
247    - `to_json`
248    - `from_xarray` (static)
249    - `to_xarray`
250    - `from_scipp` (static)
251    - `to_scipp`
252    - `from_nddata` (static)
253    - `to_nddata`
254    '''
255
256    def __init__(self, xnd=None, name=None):
257        '''Xdataset constructor
258
259            *Parameters*
260
261            - **xnd** : Xdataset/Xndarray/list of Xndarray (default None),
262            - **name** : String (default None) - name of the Xdataset
263        '''
264        self.name = name
265        match xnd:
266            case list():
267                self.xnd = xnd
268            case xdat if isinstance(xdat, Xdataset):
269                self.name = xdat.name
270                self.xnd = xdat.xnd
271            case xnda if isinstance(xnda, Xndarray):
272                self.xnd = [xnda]
273            case _:
274                self.xnd = []
275
276    def __repr__(self):
277        '''return classname and number of value'''
278        return self.__class__.__name__ + '[' + str(len(self)) + ']'
279
280    def __str__(self):
281        '''return json string format'''
282        return json.dumps(self.to_json())
283
284    def __eq__(self, other):
285        '''equal if xnd are equal'''
286        for xnda in self.xnd:
287            if not xnda in other:
288                return False
289        for xnda in other.xnd:
290            if not xnda in self:
291                return False
292        return True
293
294    def __len__(self):
295        '''number of Xndarray'''
296        return len(self.xnd)
297
298    def __contains__(self, item):
299        ''' item of xnd'''
300        return item in self.xnd
301
302    def __getitem__(self, selec):
303        ''' return Xndarray or tuple of Xndarray with selec:
304            - string : name of a xndarray,
305            - integer : index of a xndarray,
306            - index selector : index interval
307            - tuple : names or index '''
308        if selec is None or selec == '' or selec in ([], ()):
309            return self
310        if isinstance(selec, (list, tuple)) and len(selec) == 1:
311            selec = selec[0]
312        if isinstance(selec, tuple):
313            return [self[i] for i in selec]
314        if isinstance(selec, str):
315            return self.dic_xnd[selec]
316        if isinstance(selec, list):
317            return self[selec[0]][selec[1:]]
318        return self.xnd[selec]
319
320    def __delitem__(self, ind):
321        '''remove a Xndarray (ind is index, name or tuple of names).'''
322        if isinstance(ind, int):
323            del self.xnd[ind]
324        elif isinstance(ind, str):
325            del self.xnd[self.names.index(ind)]
326        elif isinstance(ind, tuple):
327            ind_n = [self.names[i] if isinstance(i, int) else i for i in ind]
328            for i in ind_n:
329                del self[i]
330
331    def __copy__(self):
332        ''' Copy all the data '''
333        return self.__class__(self)
334
335    def parent(self, var):
336        '''return the Xndarray parent (where the full_name is equal to the name)'''
337        if var.name in self.names:
338            return self[var.name]
339        return var
340
341    def dims(self, var, json_name=False):
342        '''return the list of parent namedarrays of the links of a Xndarray
343
344        *parameters*
345
346        - **var**: string - full_name of the Xndarray
347        - **json_name**: boolean (defaut False) - if True return json_name else full_name
348        '''
349        if not var in self.names:
350            return None
351        if self[var].add_name and not self[var].links:
352            return self.dims(self[var].name, json_name)
353        if var in self.namedarrays:
354            return [self[var].json_name if json_name else var]
355        if not var in self.variables + self.additionals:
356            return None
357        list_dims = []
358        for link in self[var].links:
359            list_dims += self.dims(link, json_name) if self.dims(link,
360                                                                 json_name) else [link]
361        return list_dims
362
363    def shape_dims(self, var):
364        '''return a shape with the dimensions associated to the var full_name'''
365        return [len(self[dim]) for dim in self.dims(var)
366                ] if set(self.dims(var)) <= set(self.names) else None
367
368    @property
369    def validity(self):
370        '''return the validity state: 'inconsistent', 'undifined' or 'valid' '''
371        for xnda in self:
372            if xnda.mode in ['relative', 'inconsistent']:
373                return 'undefined'
374        if self.undef_links or self.undef_vars:
375            return 'inconsistent'
376        return 'valid'
377
378    @property
379    def xtype(self):
380        '''return the Xdataset type: 'meta', 'group', 'mono', 'multi' '''
381        if self.metadata and not (self.additionals or self.variables or
382                                  self.namedarrays):
383            return 'meta'
384        if self.validity != 'valid':
385            return 'group'
386        match len(self.data_vars):
387            case 0:
388                return 'group'
389            case 1:
390                return 'mono'
391            case _:
392                return 'multi'
393
394    @property
395    def dic_xnd(self):
396        '''return a dict of Xndarray where key is the full_name'''
397        return {xnda.full_name: xnda for xnda in self.xnd}
398
399    @property
400    def names(self):
401        '''return a tuple with the Xndarray full_name'''
402        return tuple(xnda.full_name for xnda in self.xnd)
403
404    @property
405    def partition(self):
406        '''return a dict of Xndarray grouped with category'''
407        dic = {}
408        dic |= {'data_vars': list(self.data_vars)} if self.data_vars else {}
409        dic |= {'data_arrays': list(self.data_arrays)
410                } if self.data_arrays else {}
411        dic |= {'dimensions': list(self.dimensions)} if self.dimensions else {}
412        dic |= {'coordinates': list(self.coordinates)
413                } if self.coordinates else {}
414        dic |= {'additionals': list(self.additionals)
415                } if self.additionals else {}
416        dic |= {'metadata': list(self.metadata)} if self.metadata else {}
417        return dic
418
419    @property
420    def info(self):
421        '''return a dict with Xdataset information '''
422        inf = {'name': self.name, 'xtype': self.xtype} | self.partition
423        inf['validity'] = self.validity
424        inf['length'] = len(self[self.data_vars[0]]) if self.data_vars else 0
425        inf['width'] = len(self)
426        return {key: val for key, val in inf.items() if val}
427
428    def to_canonical(self):
429        '''remove optional links of the included Xndarray'''
430        for name in self.names:
431            if self[name].links in ([self[name].name], [name]):
432                self[name].links = None
433        for add in self.additionals:
434            if self[add].links in [self[self[add].name].links,
435                                   [self[add].name]]:
436                self[add].links = None
437        return self
438
439    def to_ndarray(self, full_name):
440        '''convert a Xndarray from a Xdataset in a np.ndarray'''
441        if self.shape_dims(full_name) is None:
442            data = self[full_name].ndarray
443        else:
444            data = self[full_name].darray.reshape(self.shape_dims(full_name))
445        if data.dtype.name[:8] == 'datetime':
446            data = data.astype('datetime64[ns]')
447        return data