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