python.observation.field_interface

  1# -*- coding: utf-8 -*-
  2"""
  3Created on Sun Oct  2 22:24:59 2022
  4
  5@author: philippe@loco-labs.io
  6
  7The `python.observation.field_interface` module contains the `FieldInterface` class
  8(`python.observation.field.Field` methods).
  9"""
 10# %% declarations
 11import json
 12import datetime
 13import numpy as np
 14import pandas as pd
 15import cbor2
 16
 17from observation.esconstante import ES
 18from observation.util import util, identity
 19from json_ntv.ntv import Ntv, NtvSingle, NtvList
 20
 21
 22class CborDecoder(json.JSONDecoder):
 23    ''' Cbor extension for integer keys (codification keys)'''
 24
 25    def __init__(self):
 26        json.JSONDecoder.__init__(self, object_hook=self.codecbor)
 27
 28    def codecbor(self, dic):
 29        dic2 = {}
 30        for k, v in dic.items():
 31            try:
 32                k2 = int(k)
 33            except:
 34                k2 = k
 35            dic2[k2] = v
 36        return dic2
 37
 38
 39class FieldError(Exception):
 40    ''' Field Exception'''
 41    # pass
 42
 43
 44class FieldEncoder(json.JSONEncoder):
 45    """new json encoder for Field and Dataset"""
 46
 47    def default(self, o):
 48        if isinstance(o, datetime.datetime):
 49            return o.isoformat()
 50        option = {'encoded': False, 'format': 'json'}
 51        if o.__class__.__name__ in ('Dataset', 'TimeSlot', 'Ndataset', 'Sdataset'):
 52            return o.json(**option)
 53        #if issubclass(o.__class__, ESValue):
 54        #    return o.json(**option)
 55        try:
 56            return o.to_json(**option)
 57        except:
 58            try:
 59                return o.__to_json__()
 60            except:
 61                return json.JSONEncoder.default(self, o)
 62
 63
 64class FieldInterface:
 65    '''this class includes Field methods :
 66
 67    - `FieldInterface.json`
 68    - `FieldInterface.to_obj`
 69    - `FieldInterface.to_dict_obj`
 70    - `FieldInterface.to_numpy`
 71    - `FieldInterface.to_pandas`
 72    - `FieldInterface.vlist`
 73    - `FieldInterface.vName`
 74    - `FieldInterface.vSimple`
 75    '''
 76
 77
 78    @classmethod 
 79    def decode_ntv(cls, field, fast=False):
 80        '''Generate a tuple data from a Ntv value(bytes, string, json, Ntv object)
 81
 82        *Parameters*
 83
 84        - **field** : bytes, string json or Ntv object to convert
 85        - **format** : string (default 'json') - format to convert ntv_value
 86        - **fast**: boolean (default False) - if True, codec is created without 
 87        conversion, else codec is created with json structure
 88
 89        *Returns* 
 90
 91        - **tuple** : name, dtype, codec, parent, keys, coef, leng
 92            name (None or string): name of the Field
 93            dtype (None or string): type of data
 94            codec (list): list of Field codec values
 95            parent (None or int): Field parent or None
 96            keys (None or list): Field keys
 97            coef (None or int): coef if primary Field else None
 98            leng (int): length of the Field
 99        '''
100        ntv = Ntv.obj(field)
101        typ = ntv.type_str if ntv.ntv_type else None
102        nam = ntv.name
103        val = ntv.val
104        if isinstance(ntv, NtvSingle):
105            return (nam, typ, [cls.s_to_i(val, fast)], None, None, None, 1)
106        if len(ntv) == 0:
107            return (nam, typ, cls.n_to_i(val, fast), None, None, None, 0)
108        if len(ntv) > 3 or isinstance(ntv[0], NtvSingle):
109            return (nam, typ, cls.n_to_i(val, fast), None, None, None, len(ntv))
110        if len(ntv) == 1:
111            return (nam, typ, [cls.s_to_i(val, fast)[0]], None, None, None, 1)
112
113        ntvc = ntv[0]
114        leng = max(len(ind) for ind in ntv)
115        typc = ntvc.type_str if ntvc.ntv_type else None
116        valc = ntvc.val
117        if len(ntv) == 3 and isinstance(ntv[1], NtvSingle) and \
118            isinstance(ntv[1].val, (int, str)) and not isinstance(ntv[2], NtvSingle) and \
119            isinstance(ntv[2][0].val, int):
120            return (nam, typc, cls.n_to_i(valc, fast), ntv[1].val, ntv[2].to_obj(), None, leng)
121        if len(ntv) == 2 and len(ntv[1]) == 1 and isinstance(ntv[1].val, (int, str)):
122            return (nam, typc, cls.n_to_i(valc, fast), ntv[1].val, None, None, leng) 
123        if len(ntv) == 2 and len(ntv[1]) == 1 and isinstance(ntv[1].val, list):
124            leng = leng * ntv[1][0].val
125            return (nam, typc, cls.n_to_i(valc, fast), None, None, ntv[1][0].val, leng) 
126        if len(ntv) == 2 and len(ntv[1]) > 1  and isinstance(ntv[1][0].val, int):
127            return (nam, typc, cls.n_to_i(valc, fast), None, ntv[1].to_obj(), None, leng)
128        return (nam, typ, cls.n_to_i(val, fast), None, None, None, len(ntv))
129
130    @staticmethod 
131    def encode_coef(lis):
132        '''Generate a repetition coefficient for periodic list'''
133        if len(lis) < 2:
134            return 0
135        coef = 1
136        while coef != len(lis):
137            if lis[coef-1] != lis[coef]:
138                break
139            coef += 1
140        #print('coef : ', coef)
141        '''coef = 0
142        for i in range(1,len(lis)):
143            coef = i
144            if lis[i-1] != lis[i]:
145                break'''
146        if (not len(lis) % (coef * (max(lis) + 1)) and 
147            lis == FieldInterface.keysfromcoef(coef, max(lis) + 1, len(lis))):
148            return coef
149        return 0
150
151    @staticmethod 
152    def keysfromcoef(coef, period, leng=None):
153        ''' return a list of keys with periodic structure'''
154        if not leng:
155            leng = coef * period
156        return None if not coef or not period else [ (ikey % (coef * period)) // coef 
157                                                    for ikey in range(leng)]
158    
159    def to_dict_obj(self, typevalue=None, simpleval=False, modecodec='optimize', **kwargs):
160        '''deprecated method'''
161        option = {'encoded': False, 'format': 'json', 'untyped': False,
162                  'codif': {}, 'geojson': False} | kwargs
163        dic = {}
164        if self.typevalue:
165            dic['type'] = self.typevalue
166        ds = pd.Series(range(len(self.keys)), index=self.keys, dtype='int64')
167        dic['value'] = [{'record': ds[i].tolist(),
168                         'codec': util.json(cod, encoded=False, typevalue=None,
169                                            simpleval=simpleval, modecodec=modecodec,
170                                            untyped=option['untyped'], datetime=False,
171                                            geojson=option['geojson'])}
172                        for i, cod in enumerate(self.codec)]
173        return {self.name: dic}
174
175    def to_numpy(self, func=None, codec=False, npdtype=None, **kwargs):
176        '''
177        Transform Field in a Numpy array.
178
179        *Parameters*
180
181        - **func** : function (default None) - function to apply for each value of the Field.
182        If func is the 'index' string, values are replaced by raw values.
183        - **npdtype** : string (default None) - numpy dtype for the Array ('object' if None)
184        - **kwargs** : parameters to apply to the func function
185
186        *Returns* : Numpy Array'''
187        return self.to_pandas(func=func, codec=codec, npdtype=npdtype, numpy=True, **kwargs)
188
189    def to_ntv(self, modecodec='optimize', codecval=False, def_type=None, 
190               keys=None, parent=None, name=True, coef=None):
191        '''Return a Ntv field value
192
193        *Parameters (kwargs)*
194
195        - **modecodec** : string (default 'optimize') - if 'full', index is with a full codec
196        if 'default' index has keys, if 'optimize' keys are optimized, 
197        - **codecval** : boolean (default False) - if True, only list of codec values is included
198        - **def_type** : string (default 'json') - default ntv_type for NtvList or NtvSet
199        - **name** : boolean (default False) - if False, default index name are not included
200        - **keys** : list (default None) - used only with 'optimize' mode
201        - **parent** : int or str (default None) - used only with 'optimize' mode
202
203        *Returns* : Ntv object'''
204        leng = len(self)
205        codec = self.i_to_n(self.codec)
206        def_type = codec[0].ntv_type if not def_type and codec else def_type
207        idxname = None if self.name == '$default' or not name else self.name       
208        '''if len(self.codec) == 1:
209            return NtvSingle(self.codec[0].ntv_value, idxname, self.codec[0].ntv_type)
210        if codecval:
211            return NtvList(self.codec, idxname, ntv_type=def_type)
212        if len(self.codec) == leng or modecodec == 'full':
213            return NtvList(self.values, idxname, ntv_type=def_type)
214        if modecodec == 'default':
215            return NtvList([NtvList(self.codec, ntv_type=def_type), 
216                            NtvList(self.keys, ntv_type='json')], idxname, ntv_type='json')
217        if modecodec == 'optimize':
218            ntv_value = [NtvList(self.codec, ntv_type=def_type)]'''
219        if leng == 1 or len(codec) == 1 and modecodec != 'full':
220            return NtvSingle(codec[0].ntv_value, idxname, codec[0].ntv_type)
221        if codecval or modecodec == 'nokeys':
222            return NtvList(codec, idxname, ntv_type=def_type)
223        if len(codec) == leng or modecodec == 'full':
224            #if (len(codec) == leng and not self.keys) or modecodec == 'full':
225            #return NtvList(self.l_to_e(self.values), idxname, ntv_type=def_type)
226            return NtvList(self.values, idxname, ntv_type=def_type)
227        if modecodec == 'default':
228            return NtvList([NtvList(codec, ntv_type=def_type), 
229                            NtvList(self.keys, ntv_type='json')], idxname, ntv_type='json')
230        if coef:
231            return NtvList([NtvList(codec, ntv_type=def_type),
232                         NtvList([coef], ntv_type='json')], idxname, ntv_type='json')                        
233        if modecodec == 'optimize':
234            ntv_value = [NtvList(codec, ntv_type=def_type)]
235            if not parent is None:
236                ntv_value.append(NtvSingle(parent, ntv_type='json'))
237            if keys:
238                ntv_value.append(NtvList(keys, ntv_type='json'))    
239            elif parent is None:
240                ntv_value.append(NtvList(self.keys, ntv_type='json'))
241            return NtvList(ntv_value, idxname, ntv_type='json')                
242
243    def to_pandas(self, func=None, codec=False, npdtype=None,
244                  series=True, index=True, numpy=False, **kwargs):
245        '''
246        Transform Field in a Pandas Series, Pandas DataFrame or Numpy array.
247
248        *Parameters*
249
250        - **func** : function (default None) - function to apply for each value of the Field.
251        If func is the 'index' string, values are replaced by raw values.
252        - **npdtype** : string (default None) - numpy dtype for the Array ('object' if None)
253        - **series** : boolean (default True) - if True, return a Series. 
254        If False return a DataFrame
255        - **index** : boolean (default True) - if True, index is keys.
256        - **numpy** : boolean (default False) - if True, return a Numpy array.
257        - **kwargs** : parameters to apply to the func function
258
259        *Returns* : Pandas Series, Pandas DataFrame, Numpy Array'''
260        if len(self) == 0:
261            raise FieldError("Dataset is empty")
262        if npdtype:
263            npdtype = np.dtype(npdtype)
264        else:
265            npdtype = 'object'
266        if func is None:
267            func = identity
268        if func == 'index':
269            return np.array(list(range(len(self))))
270        if not codec:
271            values = util.funclist(self.values, func, **kwargs)
272        else:
273            values = util.funclist(self._codec, func, **kwargs)
274        npdtype1 = npdtype
275        if isinstance(values[0], (datetime.datetime)):
276            npdtype1 = np.datetime64
277        # else:
278        #    npdtype=None
279        pdindex = None
280        if index:
281            pdindex = self._keys
282        try:
283            if numpy:
284                return np.array(values, dtype=npdtype1)
285            if series:
286                return pd.Series(values, dtype=npdtype1,
287                                 index=pdindex, name=self.name)
288            return pd.DataFrame(pd.Series(values, dtype=npdtype1,
289                                          index=pdindex, name=self.name))
290        except:
291            if numpy:
292                return np.array(values, dtype=npdtype)
293            if series:
294                return pd.Series(values, dtype=npdtype,
295                                 index=pdindex, name=self.name)
296            return pd.DataFrame(pd.Series(values, dtype=npdtype,
297                                          index=pdindex, name=self.name))
298
299    def vlist(self, func, *args, extern=True, **kwargs):
300        '''
301        Apply a function to values and return the result.
302
303        *Parameters*
304
305        - **func** : function - function to apply to values
306        - **args, kwargs** : parameters for the function
307        - **extern** : if True, the function is apply to external values, else internal
308
309        *Returns* : list of func result'''
310        if extern:
311            return util.funclist(self.val, func, *args, **kwargs)
312        return util.funclist(self.values, func, *args, **kwargs)
313
314    def vName(self, default=ES.nullName, maxlen=None):
315        '''
316        Return the list of name for ESValue data .
317
318        *Parameters*
319
320        - **default** : value return if no name is available
321        - **maxlen** : integer (default None) - max length of name
322
323        *Returns* : list of name founded'''
324        #return [util.cast(val, dtype='name', default=default, maxlen=maxlen) for val in self.values]
325        return [self.i_to_name(val) for val in self.values]
326
327    """def vSimple(self, string=False):
328        '''
329        Apply a vSimple function to values and return the result.
330
331        *Parameters*
332
333        - **string** : boolean(default False) - if True the values returned are string
334
335        *Returns* : list of vSimple values (string or not)'''
336        if string:
337            return json.dumps([util.cast(val, 'simple', string=string) for val in self.values],
338                              cls=ESValueEncoder)
339        return [util.cast(val, 'simple', string=string) for val in self.values]"""