NTV.json_ntv.ntv_util

  1# -*- coding: utf-8 -*-
  2"""
  3Created on Feb 27 2023
  4
  5@author: Philippe@loco-labs.io
  6
  7The `ntv_util` module is part of the `NTV.json_ntv` package ([specification document](
  8https://github.com/loco-philippe/NTV/blob/main/documentation/JSON-NTV-standard.pdf)).
  9
 10It contains the classes `NtvConnector`, `NtvTree`, `NtvJsonEncoder` and `NtvError`
 11for NTV entities.
 12"""
 13from abc import ABC, abstractmethod
 14import datetime
 15import json
 16
 17class NtvUtil:
 18    ''' The NtvUtil class includes static methods used by several NTV classes '''
 19
 20    @staticmethod
 21    def from_obj_name(string):
 22        '''return a tuple with name, type_str and separator from string'''
 23        if not isinstance(string, str):
 24            raise NtvError('a json-name have to be str')
 25        if string == '':
 26            return (None, None, None)
 27        sep = None
 28        if '::' in string:
 29            sep = '::'
 30        elif ':' in string:
 31            sep = ':'
 32        if sep is None:
 33            return (string, None, None)
 34        split = string.rsplit(sep, 2)
 35        if len(split) == 1:
 36            return (string, None, sep)
 37        if split[0] == '':
 38            return (None, split[1], sep)
 39        if split[1] == '':
 40            return (split[0], None, sep)
 41        return (split[0], split[1], sep)
 42
 43    @staticmethod
 44    def decode_ntv_tab(ntv, ntv_to_val):
 45        '''Generate a tuple data from a Ntv tab value (bytes, string, json, Ntv object)
 46
 47        *Returns tuple: (name, dtype, codec, parent, keys, coef, leng)*
 48
 49        - name (None or string): name of the Field
 50        - dtype (None or string): type of data
 51        - codec (list): list of Field codec values
 52        - parent (None or int): Field parent or None
 53        - keys (None or list): Field keys
 54        - coef (None or int): coef if primary Field else None
 55        - leng (int): length of the Field
 56        '''
 57        #ntv = Ntv.obj(field)
 58        typ = ntv.type_str if ntv.ntv_type else None
 59        nam = ntv.name
 60        val = ntv_to_val(ntv)
 61        if ntv.__class__.__name__ == 'NtvSingle':
 62            return (nam, typ, [val], None, None, None, 1)
 63        if len(ntv) < 2 or len(ntv) > 3 or ntv[0].__class__.__name__ == 'NtvSingle':
 64            return (nam, typ, val, None, None, None, len(ntv))
 65
 66        ntvc = ntv[0]
 67        leng = max(len(ind) for ind in ntv)
 68        typc = ntvc.type_str if ntvc.ntv_type else None
 69        valc = ntv_to_val(ntvc)
 70        if len(ntv) == 3 and ntv[1].__class__.__name__ == 'NtvSingle' and \
 71                isinstance(ntv[1].val, (int, str)) and \
 72                ntv[2].__class__.__name__ != 'NtvSingle' and \
 73                isinstance(ntv[2][0].val, int):
 74            return (nam, typc, valc, ntv[1].val, ntv[2].to_obj(), None, leng)
 75        if len(ntv) == 2 and len(ntv[1]) == 1 and isinstance(ntv[1].val, (int, str)):
 76            return (nam, typc, valc, ntv[1].val, None, None, leng)
 77        if len(ntv) == 2 and len(ntv[1]) == 1 and isinstance(ntv[1].val, list):
 78            leng = leng * ntv[1][0].val
 79            return (nam, typc, valc, None, None, ntv[1][0].val, leng)
 80        if len(ntv) == 2 and len(ntv[1]) > 1 and isinstance(ntv[1][0].val, int):
 81            return (nam, typc, valc, None, ntv[1].to_obj(), None, leng)
 82        return (nam, typ, val, None, None, None, len(ntv))
 83    
 84class NtvConnector(ABC):
 85    ''' The NtvConnector class is an abstract class used by all NTV connectors
 86    for conversion between NTV-JSON data and NTV-OBJ data.
 87
 88    A NtvConnector child is defined by:
 89    - clas_obj: str - define the class name of the object to convert
 90    - clas_typ: str - define the NTVtype of the converted object
 91    - to_obj_ntv: method - converter from JsonNTV to the object
 92    - to_json_ntv: method - converter from the object to JsonNTV
 93    
 94    *class method :*
 95    - `connector`
 96    - `dic_connec`
 97    - `castable` (@property)
 98    - `dic_obj` (@property)
 99    - `dic_type` (@property)
100
101    *abstract method*
102    - `to_obj_ntv`
103    - `to_json_ntv`
104
105    *static method*
106    - `cast`
107    - `uncast`
108    - `is_json_class`
109    - `is_json`
110    - `init_ntv_keys`
111    '''
112
113    DIC_NTV_CL = {'NtvSingle': 'ntv', 'NtvList': 'ntv'}
114    DIC_GEO_CL = {'Point': 'point', 'MultiPoint': 'multipoint', 'LineString': 'line',
115                  'MultiLineString': 'multiline', 'Polygon': 'polygon',
116                  'MultiPolygon': 'multipolygon'}
117    DIC_DAT_CL = {'date': 'date', 'time': 'time', 'datetime': 'datetime'}
118    DIC_FCT = {'date': datetime.date.fromisoformat, 'time': datetime.time.fromisoformat,
119               'datetime': datetime.datetime.fromisoformat}
120    DIC_GEO = {'point': 'point', 'multipoint': 'multipoint', 'line': 'linestring',
121               'multiline': 'multilinestring', 'polygon': 'polygon',
122               'multipolygon': 'multipolygon'}
123    DIC_CBOR = {'point': False, 'multipoint': False, 'line': False,
124                'multiline': False, 'polygon': False, 'multipolygon': False,
125                'date': True, 'time': False, 'datetime': True}
126    DIC_OBJ = {'tab': 'DataFrameConnec', 'field': 'SeriesConnec',
127               'point': 'ShapelyConnec', 'multipoint': 'ShapelyConnec',
128               'line': 'ShapelyConnec', 'multiline': 'ShapelyConnec',
129               'polygon': 'ShapelyConnec', 'multipolygon': 'ShapelyConnec',
130               'other': None}
131
132    @classmethod
133    @property
134    def castable(cls):
135        '''return a list with class_name allowed for json-obj conversion'''
136        return ['str', 'int', 'bool', 'float', 'dict', 'tuple', 'NoneType',
137                'NtvSingle', 'NtvList'] \
138            + list(NtvConnector.DIC_GEO_CL.keys()) \
139            + list(NtvConnector.DIC_FCT.keys()) \
140            + list(NtvConnector.dic_connec().keys())
141
142    @classmethod
143    @property
144    def dic_obj(cls):
145        '''return a dict with the connectors: { type: class_connec_name }'''
146        return {clas.clas_typ: clas.__name__ for clas in cls.__subclasses__()} |\
147            NtvConnector.DIC_OBJ
148
149    @classmethod
150    @property
151    def dic_type(cls):
152        '''return a dict with the connectors: { class_obj_name: type }'''
153        return {clas.clas_obj: clas.clas_typ for clas in cls.__subclasses__()} |\
154            NtvConnector.DIC_GEO_CL | NtvConnector.DIC_DAT_CL | NtvConnector.DIC_NTV_CL
155
156    @classmethod
157    def connector(cls):
158        '''return a dict with the connectors: { class_connec_name: class_connec }'''
159        return {clas.__name__: clas for clas in cls.__subclasses__()}
160
161    @classmethod
162    def dic_connec(cls):
163        '''return a dict with the clas associated to the connector:
164        { class_obj_name: class_connec_name }'''
165        return {clas.clas_obj: clas.__name__ for clas in cls.__subclasses__()}
166
167    @staticmethod
168    @abstractmethod
169    def to_obj_ntv(ntv_value, **kwargs):
170        ''' abstract - convert ntv_value into the return object'''
171
172    @staticmethod
173    @abstractmethod
174    def to_json_ntv(value, name=None, typ=None):
175        ''' abstract - convert NTV object (value, name, type) into the NTV json
176        (json-value, name, type)'''
177
178    @staticmethod
179    def cast(data, name=None, type_str=None):
180        '''return JSON-NTV conversion (json_value, name, type_str) of the NTV entity
181        defined in parameters.
182
183        *Parameters*
184
185        - **data**: NtvSingle entity or NTVvalue of the NTV entity
186        - **name** : String (default None) - name of the NTV entity
187        - **type_str**: String (default None) - type of the NTV entity
188        '''
189        clas = data.__class__.__name__
190        if clas == 'NtvSingle':
191            name = data.ntv_name
192            type_str = data.type_str
193            data = data.ntv_value
194        dic_geo_cl = NtvConnector.DIC_GEO_CL
195        dic_connec = NtvConnector.dic_connec()
196        match clas:
197            case 'int' | 'float' | 'bool' | 'str':
198                return (data, name, type_str)
199            case 'dict':
200                return ({key: NtvConnector.cast(val, name, type_str)[0]
201                         for key, val in data.items()}, name, type_str)
202            case 'list':
203                return ([NtvConnector.cast(val, name, type_str)[0] for val in data],
204                        name, NtvConnector._typ_obj(data) if not type_str else type_str)
205            case 'tuple':
206                return (list(data), name, 'array' if not type_str else type_str)
207            case 'date' | 'time' | 'datetime':
208                return (data.isoformat(), name, clas if not type_str else type_str)
209            case 'Point' | 'MultiPoint' | 'LineString' | 'MultiLineString' | \
210                    'Polygon' | 'MultiPolygon':
211                return (NtvConnector.connector()[dic_connec['geometry']].to_json_ntv(data)[0],
212                        name, dic_geo_cl[data.__class__.__name__] if not type_str else type_str)
213            case 'NtvSingle' | 'NtvList':
214                return (data.to_obj(), name, 'ntv' if not type_str else type_str)
215            case _:
216                connec = None
217                if clas in dic_connec and dic_connec[clas] in NtvConnector.connector():
218                    connec = NtvConnector.connector()[dic_connec[clas]]
219                if connec:
220                    return connec.to_json_ntv(data, name, type_str)
221                raise NtvError(
222                    'connector is not defined for the class : ', clas)
223        return (data, name, type_str)
224
225    @staticmethod
226    def uncast(value, name=None, type_str=None, **kwargs):
227        '''return OBJ-NTV conversion (obj_value, name, type_str) of a NTV entity
228
229        *Parameters*
230
231        - **data**: NtvSingle entity or NTVvalue of the NTV entity
232        - **name** : String (default None) - name of the NTV entity
233        - **type_str**: String (default None) - type of the NTV entity
234        '''
235        if type_str == 'json':
236            return (value, name, type_str)
237        if value.__class__.__name__ == 'NtvSingle':
238            if not (type_str in set(NtvConnector.dic_type.values()) and 
239                    NtvConnector.is_json(value) or type_str is None):
240                return (value.ntv_value, value.name, value.type_str)
241            type_str = value.type_str if value.ntv_type else None
242            name = value.ntv_name
243            value = value.ntv_value
244        #dic_obj = NtvConnector.dic_obj
245        option = {'dicobj': {}, 'format': 'json', 'type_obj': False} | kwargs
246        value_obj = NtvConnector._uncast_val(value, type_str, **option)
247        return (value_obj, name, type_str if type_str else NtvConnector._typ_obj(value_obj))
248
249    @staticmethod
250    def _typ_obj(value):
251        if isinstance(value, dict):
252            return NtvConnector._typ_obj(list(value.values()))
253        if isinstance(value, (tuple, list)):
254            for val in value:
255                typ = NtvConnector._typ_obj(val)
256                if typ:
257                    return typ
258            return None
259        return NtvConnector.dic_type.get(value.__class__.__name__)
260
261    @staticmethod
262    def _uncast_val(value, type_n, **option):
263        '''return value from ntv value'''
264        dic_fct = NtvConnector.DIC_FCT
265        dic_geo = NtvConnector.DIC_GEO
266        dic_obj = NtvConnector.dic_obj | option['dicobj']
267        dic_cbor = NtvConnector.DIC_CBOR
268        if not type_n or type_n == 'json' or (option['format'] == 'cbor' and 
269                                              not dic_cbor.get(type_n, False)):
270            return value
271        if type_n in dic_fct:
272            if isinstance(value, (tuple, list)):
273                return [NtvConnector._uncast_val(val, type_n, **option) for val in value]
274            if isinstance(value, dict):
275                return {key: NtvConnector._uncast_val(val, type_n, **option)
276                        for key, val in value.items()}
277            return dic_fct[type_n](value)
278        if type_n == 'array':
279            return tuple(value)
280        if type_n == 'ntv':
281            from json_ntv.ntv import Ntv
282            return Ntv.from_obj(value)
283        if type_n in dic_geo:
284            option['type_geo'] = dic_geo[type_n]
285        connec = None
286        if type_n in dic_obj and \
287                dic_obj[type_n] in NtvConnector.connector():
288            connec = NtvConnector.connector()[dic_obj[type_n]]
289        elif dic_obj['other'] in NtvConnector.connector():
290            connec = NtvConnector.connector()['other']
291        if connec:
292            return connec.to_obj_ntv(value, **option)
293        #raise NtvError('type of value not allowed for conversion')
294        return value
295
296    @staticmethod
297    def is_json_class(val):
298        ''' return True if val is a json type'''
299        return val is None or isinstance(val, (list, int, str, float, bool, dict))
300
301    @staticmethod
302    def is_json(obj):
303        ''' check if obj is a json structure and return True if obj is a json-value
304
305        *Parameters*
306
307        - **obj** : object to check
308        - **ntvobj** : boolean (default False) - if True NTV class value are accepted'''
309        if obj is None:
310            return True
311        is_js = NtvConnector.is_json
312        match obj:
313            case str() | int() | float() | bool() as obj:
314                return True
315            case list() | tuple() as obj:
316                if not obj:
317                    return True
318                return min(is_js(obj_in) for obj_in in obj)
319            case dict() as obj:
320                if not obj:
321                    return True
322                if not min(isinstance(key, str) for key in obj.keys()):
323                    raise NtvError('key in dict in not string')
324                return min(is_js(obj_in) for obj_in in obj.values())
325            case _:
326                if not obj.__class__.__name__ in NtvConnector.castable:
327                    raise NtvError(obj.__class__.__name__ +
328                                   'is not valid for NTV')
329                return False
330
331    @staticmethod
332    def keysfromderkeys(parentkeys, derkeys):
333        '''return keys from parent keys and derkeys
334
335        *Parameters*
336
337        - **parentkeys** : list of keys from parent
338        - **derkeys** : list of derived keys
339
340        *Returns* : list of keys'''
341        return [derkeys[pkey] for pkey in parentkeys]
342
343    @staticmethod 
344    def encode_coef(lis):
345        '''Generate a repetition coefficient for periodic list'''
346        if len(lis) < 2:
347            return 0
348        coef = 1
349        while coef != len(lis):
350            if lis[coef-1] != lis[coef]:
351                break
352            coef += 1
353        if (not len(lis) % (coef * (max(lis) + 1)) and 
354            lis == NtvConnector.keysfromcoef(coef, max(lis) + 1, len(lis))):
355            return coef
356        return 0
357    
358    @staticmethod 
359    def keysfromcoef(coef, period, leng=None):
360        ''' return a list of keys with periodic structure'''
361        if not leng:
362            leng = coef * period
363        return None if not (coef and period) else [(ind % (coef * period)) // coef 
364                                                   for ind in range(leng)]    
365    @staticmethod
366    def init_ntv_keys(ind, lidx, leng):
367        ''' initialization of explicit keys data in lidx object of tabular data'''
368        # name: 0, type: 1, codec: 2, parent: 3, keys: 4, coef: 5, leng: 6
369        name, typ, codec, parent, keys, coef, length = lidx[ind]
370        if (keys, parent, coef) == (None, None, None):  # full or unique
371            if len(codec) == 1: # unique
372                lidx[ind][4] = [0] * leng
373            elif len(codec) == leng:    # full
374                lidx[ind][4] = list(range(leng))
375            else:
376                raise NtvError('impossible to generate keys')
377            return
378        if keys and len(keys) > 1 and parent is None:  #complete
379            return
380        if coef:  #primary
381            lidx[ind][4] = [(ikey % (coef * len(codec))) // coef for ikey in range(leng)]
382            lidx[ind][3] = None
383            return  
384        if parent is None:
385            raise NtvError('keys not referenced')          
386        if not lidx[parent][4] or len(lidx[parent][4]) != leng:
387            NtvConnector.init_ntv_keys(parent, lidx, leng)
388        if not keys and len(codec) == len(lidx[parent][2]):    # implicit
389            lidx[ind][4] = lidx[parent][4]
390            lidx[ind][3] = None
391            return
392        lidx[ind][4] = NtvConnector.keysfromderkeys(lidx[parent][4], keys)  # relative
393        #lidx[ind][4] = [keys[pkey] for pkey in lidx[parent][4]]  # relative
394        lidx[ind][3] = None
395        return    
396    
397class NtvTree:
398    ''' The NtvTree class is an iterator class used to traverse a NTV tree structure.
399    Some other methods give tree indicators and data.
400
401    *Attributes :*
402
403    - **ntv** : Ntv entity
404    - **_node**:  Ntv entity - node pointer
405    - **_stack**:  list - stack used to store context in recursive methods 
406
407    *dynamic values (@property)*
408    - `breadth`
409    - `size`
410    - `height`
411    - `adjacency_list`
412    - `nodes`
413    - `dic_nodes`
414    - `leaf_nodes`
415    - `inner_nodes`
416    '''
417
418    def __init__(self, ntv):
419        ''' the parameter of the constructor is the Ntv entity'''
420        self._ntv = ntv
421        self._node = None
422        self._stack = []
423
424    def __iter__(self):
425        ''' iterator without initialization'''
426        return self
427
428    def __next__(self):
429        ''' return next node in the tree'''
430        if self._node is None:
431            self._node = self._ntv
432        elif len(self._node) == 0:
433            raise StopIteration
434        elif self._node.__class__.__name__ == 'NtvList':
435            self._next_down()
436        else:
437            self._next_up()
438        return self._node
439
440    def _next_down(self):
441        ''' find the next subchild node'''
442
443        self._node = self._node[0]
444        self._stack.append(0)
445
446    def _next_up(self):
447        ''' find the next sibling or ancestor node'''
448        parent = self._node.parent
449        if not parent or self._node == self._ntv:
450            raise StopIteration
451        ind = self._stack[-1]
452        if ind < len(parent) - 1:  # if ind is not the last
453            self._node = parent[ind + 1]
454            self._stack[-1] += 1
455        else:
456            if parent == self._ntv:
457                raise StopIteration
458            self._node = parent
459            self._stack.pop()
460            self._next_up()
461            
462    @property
463    def breadth(self):
464        ''' return the number of leaves'''
465        return len(self.leaf_nodes)
466
467    @property
468    def size(self):
469        ''' return the number of nodes'''
470        return len(self.nodes)
471
472    @property
473    def height(self):
474        ''' return the height of the tree'''
475        return max(len(node.pointer()) for node in self.__class__(self._ntv))
476
477    @property
478    def adjacency_list(self):
479        ''' return a dict with the list of child nodes for each parent node'''
480        return {node: node.val for node in self.inner_nodes}
481
482    @property
483    def nodes(self):
484        ''' return the list of nodes according to the DFS preordering algorithm'''
485        return list(self.__class__(self._ntv))
486
487    @property
488    def dic_nodes(self):
489        ''' return a dict of nodes according to the DFS preordering algorithm'''
490        return {node.ntv_name: node for node in self.__class__(self._ntv)
491                if node.ntv_name}
492
493    @property
494    def leaf_nodes(self):
495        ''' return the list of leaf nodes according to the DFS preordering algorithm'''
496        #return [node for node in self.__class__(self._ntv) if not isinstance(node, NtvList)]
497        return [node for node in self.__class__(self._ntv)
498                if node.__class__.__name__ == 'NtvSingle']
499
500    @property
501    def inner_nodes(self):
502        ''' return the list of inner nodes according to the DFS preordering algorithm'''
503        return [node for node in self.__class__(self._ntv)
504                if node.__class__.__name__ == 'NtvList']
505
506
507
508
509class NtvJsonEncoder(json.JSONEncoder):
510    """json encoder for Ntv data"""
511
512    def default(self, o):
513        try:
514            return NtvConnector.cast(o)[0]
515        except NtvError:
516            return json.JSONEncoder.default(self, o)
517        if isinstance(o, (datetime.datetime, datetime.date, datetime.time)):
518            return o.isoformat()
519        return json.JSONEncoder.default(self, o)
520
521
522class NtvError(Exception):
523    ''' NTV Exception'''
524    # pass