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