NTV.json_ntv.ntv_connector

  1# -*- coding: utf-8 -*-
  2"""
  3Created on Feb 27 2023
  4
  5@author: Philippe@loco-labs.io
  6
  7The `NTV.ntv_connector` 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
 10A NtvConnector is defined by:
 11- clas_obj: str - define the class name of the object to convert
 12- clas_typ: str - define the NTVtype of the converted object
 13- to_obj_ntv: method - converter from JsonNTV to the object
 14- to_json_ntv: method - converter from the object to JsonNTV
 15    
 16It contains :
 17
 18- methods `from_csv` and `to_csv` to convert CSV files and 'tab' NTV entity
 19- the child classes of `NTV.json_ntv.ntv.NtvConnector` abstract class:
 20    - `SfieldConnec`:    'field' connector
 21    - `SdatasetConnec`:  'tab' connector
 22    - `NfieldConnec`:    'field' connector
 23    - `NdatasetConnec`:  'tab' connector
 24    - `MermaidConnec`:   '$mermaid' connector
 25    - `ShapelyConnec`:   'geometry' connector
 26    - `CborConnec`:      '$cbor' connector
 27
 28
 29"""
 30import datetime
 31import csv
 32import json
 33import pandas as pd
 34import numpy as np
 35
 36from json_ntv.ntv import Ntv, NtvConnector, NtvList, NtvSingle, NtvTree
 37#from observation import Sfield
 38
 39
 40def from_csv(file_name, single_tab=True, dialect='excel', **fmtparams):
 41    ''' return a 'tab' NtvSingle from a csv file
 42
 43    *parameters*
 44
 45    - **file_name** : name of the csv file
 46    - **single_tab** : boolean (default True) - if True return a 'tab' NtvSingle,
 47    else return a NtvSet.
 48    - **dialect, fmtparams** : parameters of csv.DictReader object'''
 49    with open(file_name, newline='') as csvfile:
 50        reader = csv.DictReader(csvfile, dialect=dialect, **fmtparams)
 51        names = reader.fieldnames
 52        list_ntv_value = [[] for nam in names]
 53        for row in reader:
 54            for ind_field, val in enumerate(list(row.values())):
 55                list_ntv_value[ind_field].append(json.loads(val))
 56    list_ntv = []
 57    for ind_field, field in enumerate(names):
 58        list_ntv.append(
 59            NtvList(list_ntv_value[ind_field], *Ntv.from_obj_name(field)[:2]))
 60    if single_tab:
 61        return NtvSingle(NtvList(list_ntv, None, None).to_obj(), None, 'tab')
 62    return NtvList(list_ntv, None, None)
 63
 64
 65def to_csv(file_name, ntv, *args, restval='', extrasaction='raise', dialect='excel', **kwds):
 66    ''' convert a 'tab' NtvSingle into csv file and return the file name
 67
 68    *parameters*
 69
 70    - **file_name** : name of the csv file
 71    - **ntv** : 'tab' NtvSingle to convert
 72    - **args, restval, extrasaction, dialect, kwds** : parameters of csv.DictWriter object'''
 73    if isinstance(ntv, NtvSingle):
 74        ntv_set = Ntv.obj(ntv.ntv_value)
 75    else:
 76        ntv_set = ntv
 77    list_ntv = [Ntv.obj(field) for field in ntv_set]
 78    fieldnames = [ntv_field.json_name(string=True) for ntv_field in list_ntv]
 79    with open(file_name, 'w', newline='') as csvfile:
 80        writer = csv.DictWriter(csvfile, fieldnames=fieldnames, restval=restval,
 81                                extrasaction=extrasaction, dialect=dialect, *args, **kwds)
 82        writer.writeheader()
 83        for i in range(len(list_ntv[0])):
 84            writer.writerow({name: field_ntv[i].to_obj(field_ntv.ntv_type, encoded=True)
 85                             for name, field_ntv in zip(fieldnames, list_ntv)})
 86    return file_name
 87
 88
 89class ShapelyConnec(NtvConnector):
 90    '''NTV connector for geographic location'''
 91
 92    clas_obj = 'geometry'
 93    clas_typ = 'geometry'
 94
 95    @staticmethod
 96    def to_obj_ntv(ntv_value, **kwargs):
 97        ''' convert ntv_value into a shapely geometry object defined by 'type_geo'.
 98
 99        *Parameters*
100
101        - **type_geo** : type of geometry (point, multipoint, line, multiline',
102        polygon, multipolygon)
103        - **ntv_value** : array - coordinates'''
104        from shapely import geometry
105        type_geo = ShapelyConnec.type_geo(ntv_value) if not 'type_geo' in kwargs \
106            or kwargs['type_geo'] == 'geometry' else kwargs['type_geo']
107        return geometry.shape({"type": type_geo, "coordinates": ntv_value})
108
109    @staticmethod
110    def to_json_ntv(value, name=None, typ=None):
111        ''' convert NTV object (value, name, type) into NTV json (json-value, name, type).
112
113        *Parameters*
114
115        - **typ** : string (default None) - NTV type of geometry (point, multipoint,
116        line, multiline', polygon, multipolygon),
117        - **name** : string (default None) - name of the NTV object
118        - **value** : shapely geometry'''
119        return (Ntv._listed(value.__geo_interface__['coordinates']), name, typ)
120
121    @staticmethod
122    def to_coord(geom):
123        ''' convert shapely geometry into geojson coordinates.'''
124        return Ntv._listed(geom.__geo_interface__['coordinates'])
125
126    @staticmethod
127    def to_geojson(geom):
128        ''' convert shapely geometry into geojson string'''
129        return json.dumps(geom.__geo_interface__)
130
131    @staticmethod
132    def from_geojson(geojson):
133        ''' convert geojson string into shapely geometry.'''
134        from shapely import geometry
135        return geometry.shape(json.loads(geojson))
136    
137    @staticmethod 
138    def to_geometry(value):
139        '''convert geojson coordinates into shapely geometry'''
140        return ShapelyConnec.to_obj_ntv(value, type_geo=ShapelyConnec.type_geo(value))
141    
142    @staticmethod 
143    def type_geo(value):
144        '''return geometry type of the value'''
145        if not value or not isinstance(value, list):
146            return 'not a geometry'
147        val = value[0]
148        if not isinstance(val, list):
149            return 'point'
150        val = val[0]
151        if not isinstance(val, list):
152            return 'line'        
153        return 'polygon'
154    
155class CborConnec(NtvConnector):
156    '''NTV connector for binary data'''
157
158    clas_obj = 'bytes'
159    clas_typ = '$cbor'
160
161    @staticmethod
162    def to_obj_ntv(ntv_value, **kwargs):
163        ''' convert json ntv_value into a binary CBOR object (no parameters).'''
164        import cbor2
165        return cbor2.dumps(ntv_value, datetime_as_timestamp=True,
166                           timezone=datetime.timezone.utc, canonical=False,
167                           date_as_datetime=True)
168
169    @staticmethod
170    def to_json_ntv(value, name=None, typ=None):
171        ''' convert NTV binary object (value, name, type) into NTV json (json-value, name, type).
172
173        *Parameters*
174
175        - **typ** : string (default None) - type of the NTV object,
176        - **name** : string (default None) - name of the NTV object
177        - **value** : binary data'''
178        import cbor2
179        return (cbor2.loads(value), name, typ)
180
181
182class NfieldConnec(NtvConnector):
183    '''NTV connector for NTV Field data'''
184
185    clas_obj = 'Nfield'
186    clas_typ = 'field'
187
188    @staticmethod
189    def to_obj_ntv(ntv_value, **kwargs):
190        ''' convert json ntv_value into a NTV Field object (no parameters).'''
191        from observation.fields import Nfield
192        ntv = Ntv.obj(ntv_value)
193        return Nfield.from_ntv(ntv)
194
195    @staticmethod
196    def to_json_ntv(value, name=None, typ=None):
197        ''' convert NTV Field object (value, name, type) into NTV json (json-value, name, type).
198
199        *Parameters*
200
201        - **typ** : string (default None) - type of the NTV object,
202        - **name** : string (default None) - name of the NTV object
203        - **value** : NTV Field values (default format)'''
204        return (value.to_ntv(name=True).to_obj(), name,
205                NfieldConnec.clas_typ if not typ else typ)
206
207
208class SfieldConnec(NtvConnector):
209    '''NTV connector for simple Field data'''
210
211    clas_obj = 'Sfield'
212    clas_typ = 'field'
213
214    @staticmethod
215    def to_obj_ntv(ntv_value, **kwargs):
216        ''' convert json ntv_value into a simple Field object (no parameters).'''
217        from observation.fields import Sfield
218        ntv = Ntv.obj(ntv_value)
219        return Sfield.from_ntv(ntv)
220
221    @staticmethod
222    def to_json_ntv(value, name=None, typ=None):
223        ''' convert simple Field object (value, name, type) into NTV json (json-value, name, type).
224
225        *Parameters*
226
227        - **typ** : string (default None) - type of the NTV object,
228        - **name** : string (default None) - name of the NTV object
229        - **value** : simple Field values (default format)'''
230        return (value.to_ntv(name=True).to_obj(), name,
231                NfieldConnec.clas_typ if not typ else typ)
232
233
234class NdatasetConnec(NtvConnector):
235    '''NTV connector for NTV Dataset data'''
236
237    clas_obj = 'Ndataset'
238    clas_typ = 'tab'
239
240    @staticmethod
241    def to_obj_ntv(ntv_value, **kwargs):
242        ''' convert json ntv_value into a NTV Dataset object (no parameters).'''
243        from observation.datasets import Ndataset
244
245        ntv = Ntv.obj(ntv_value)
246        return Ndataset.from_ntv(ntv)
247
248    @staticmethod
249    def to_json_ntv(value, name=None, typ=None):
250        ''' convert NTV Dataset object (value, name, type) into NTV json (json-value, name, type).
251
252        *Parameters*
253
254        - **typ** : string (default None) - type of the NTV object,
255        - **name** : string (default None) - name of the NTV object
256        - **value** : NTV Dataset values'''
257        return (value.to_ntv().to_obj(), name,
258                NdatasetConnec.clas_typ if not typ else typ)
259
260
261class SdatasetConnec(NtvConnector):
262    '''NTV connector for simple Dataset data'''
263
264    clas_obj = 'Sdataset'
265    clas_typ = 'tab'
266
267    @staticmethod
268    def to_obj_ntv(ntv_value, **kwargs):
269        ''' convert json ntv_value into a simple Dataset object (no parameters).'''
270        from observation.datasets import Sdataset
271
272        ntv = Ntv.obj(ntv_value)
273        return Sdataset.from_ntv(ntv)
274
275    @staticmethod
276    def to_json_ntv(value, name=None, typ=None):
277        ''' convert simple Dataset object (value, name, type) into NTV json
278        (json-value, name, type).
279
280        *Parameters*
281
282        - **typ** : string (default None) - type of the NTV object,
283        - **name** : string (default None) - name of the NTV object
284        - **value** : simple Dataset values'''
285        return (value.to_ntv().to_obj(), name,
286                SdatasetConnec.clas_typ if not typ else typ)
287
288class MermaidConnec(NtvConnector):
289    '''NTV connector for Mermaid diagram'''
290
291    clas_obj = 'Mermaid'
292    clas_typ = '$mermaid'
293
294    @staticmethod
295    def to_obj_ntv(ntv_value, **kwargs):
296        ''' convert ntv_value into a mermaid flowchart
297
298        *Parameters*
299
300        - **title**: String (default '') - title of the flowchart
301        - **disp**: Boolean (default False) - if true, return a display else return
302        a mermaid text diagram
303        - **row**: Boolean (default False) - if True, add the node row
304        - **leaves**: Boolean (default False) - if True, add the leaf row
305        '''
306        from base64 import b64encode
307        from IPython.display import Image, display
308
309        option = {'title': '', 'disp': False, 'row': False,
310                  'leaves': False} | kwargs
311        diagram = MermaidConnec.diagram
312        link = MermaidConnec._mermaid_link
313        ntv = Ntv.obj(ntv_value)
314        node_link = {'nodes': [], 'links': []}
315        dic_node = {}
316        if option['leaves']:
317            nodes = [node for node in NtvTree(
318                ntv) if not isinstance(node.val, list)]
319            dic_node = {node: row for row, node in enumerate(nodes)}
320        link(ntv, None, node_link, option['row'], dic_node, None)
321        mermaid_json = {option['title'] + ':$flowchart': {
322            'orientation': 'top-down',
323            'node::': {node[0]: node[1] for node in node_link['nodes']},
324            'link::': node_link['links']}}
325        if option['disp']:
326            return display(Image(url="https://mermaid.ink/img/" +
327                                 b64encode(diagram(mermaid_json).encode("ascii")).decode("ascii")))
328        return diagram(mermaid_json)
329
330    @staticmethod
331    def diagram(json_diag):
332        '''create a mermaid code from a mermaid json'''
333        ntv = Ntv.obj(json_diag)
334        erdiagram = MermaidConnec._er_diagram
335        flowchart = MermaidConnec._flowchart
336        diag_type = ntv.type_str[1:]
337        diag_txt = '---\ntitle: ' + ntv.name + '\n---\n' if ntv.name else ''
338        diag_txt += diag_type
339        match diag_type:
340            case 'erDiagram':
341                diag_txt += erdiagram(ntv)
342            case 'flowchart':
343                diag_txt += flowchart(ntv)
344        return diag_txt
345
346    @staticmethod
347    def _mermaid_node(ntv, def_typ_str, num, dic_node, ind):
348        '''create and return a node'''
349        j_name, j_sep, j_type = ntv.json_name(def_typ_str)
350        name = ''
351        if j_name:
352            name += '<b>' + j_name + '</b>\n'
353        if j_type:
354            name += j_type + '\n'
355        if ntv in dic_node:
356            num += ' ' + str(dic_node[ntv])
357        if num:
358            name += '<i>' + num + '</i>\n'
359        elif isinstance(ntv, NtvSingle):
360            if isinstance(ntv.val, str):
361                name += '<i>' + ntv.val + '</i>\n'
362            else:
363                name += '<i>' + json.dumps(ntv.val) + '</i>\n'
364            return [ntv.pointer(index=True, item_idx=ind).json(default='/'), 
365                    ['rectangle', name[:-1]]]
366        if not name:
367            name = '<b>::</b>\n'
368        name = name.replace('"', "'")
369        return [ntv.pointer(index=True, item_idx=ind).json(default='/'), 
370                ['roundedge', name[:-1]]]
371
372    @staticmethod
373    def _mermaid_link(ntv, def_typ_str, node_link, row, dic_node, ind):
374        '''add nodes and links from ntv in node_link '''
375        num = str(len(node_link['nodes'])) if row else ''
376        node_link['nodes'].append(MermaidConnec._mermaid_node(
377            ntv, def_typ_str, num, dic_node, ind))
378        if isinstance(ntv, NtvList):
379            for ind, ntv_val in enumerate(ntv):
380                MermaidConnec._mermaid_link(ntv_val, ntv.type_str,
381                                            node_link, row, dic_node, ind)
382                node_link['links'].append(
383                    [ntv.pointer(index=True).json(default='/'), 'normalarrow',
384                     ntv_val.pointer(index=True, item_idx=ind).json(default='/')])
385
386    @staticmethod
387    def _flowchart(ntv):
388        orientation = {'top-down': 'TD', 'top-bottom': 'TB',
389                       'bottom-top': 'BT', 'right-left': 'RL', 'left-right': 'LR'}
390        fcnode = MermaidConnec._fc_node
391        fclink = MermaidConnec._fc_link
392        flc = Ntv.obj(ntv.val)
393        diag_txt = ' ' + orientation[flc['orientation'].val]
394        for node in flc['node']:
395            diag_txt += fcnode(node)
396        for link in flc['link']:
397            diag_txt += fclink(link)
398        return diag_txt + '\n'
399
400    @staticmethod
401    def _fc_link(link):
402        link_t = {'normal': ' ---', 'normalarrow': ' -->',
403                  'dotted': ' -.-', 'dottedarrow': ' -.->'}
404        link_txt = '\n    ' + str(link[0].val) + link_t[link[1].val]
405        if len(link) == 4:
406            link_txt += '|' + link[3].val + '|'
407        return link_txt + ' ' + str(link[2].val)
408
409    @staticmethod
410    def _fc_node(node):
411        shape_l = {'rectangle': '[', 'roundedge': '(', 'stadium': '(['}
412        shape_r = {'rectangle': ']', 'roundedge': ')', 'stadium': '])'}
413        return '\n    ' + node.name + shape_l[node[0].val] + '"' + \
414               node[1].val.replace('"', "'") + '"' + shape_r[node[0].val]
415
416    @staticmethod
417    def _er_diagram(ntv):
418        erentity = MermaidConnec._er_entity
419        errelation = MermaidConnec._er_relation
420        diag_txt = ''
421        erd = Ntv.obj(ntv.val)
422        for entity in erd['entity']:
423            diag_txt += erentity(entity)
424        for relation in erd['relationship']:
425            diag_txt += errelation(relation)
426        return diag_txt
427
428    @staticmethod
429    def _er_entity(entity):
430        ent_txt = '\n    ' + entity.name + ' {'
431        for att in entity:
432            ent_txt += '\n        ' + att[0].val + ' ' + att[1].val
433            if len(att) > 2:
434                if att[2].val in ('PK', 'FK', 'UK'):
435                    ent_txt += ' ' + att[2].val
436                else:
437                    ent_txt += ' "' + att[2].val + '"'
438            if len(att) > 3:
439                ent_txt += ' "' + att[3].val + '"'
440        return ent_txt + '\n    }'
441
442    @staticmethod
443    def _er_relation(rel):
444        rel_left = {'exactly one': ' ||', 'zero or one': ' |o',
445                    'zero or more': ' }o', 'one or more': ' }|'}
446        rel_right = {'exactly one': '|| ', 'zero or one': 'o| ',
447                     'zero or more': 'o{ ', 'one or more': '|{ '}
448        identif = {'identifying': '--', 'non-identifying': '..'}
449        rel_txt = '\n    ' + rel[0].val + rel_left[rel[1].val] + \
450            identif[rel[2].val] + rel_right[rel[3].val] + rel[4].val
451        if len(rel) > 5:
452            rel_txt += ' : ' + rel[5].val
453        return rel_txt