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 Datatype 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
 37
 38def from_csv(file_name, single_tab=True, dialect='excel', **fmtparams):
 39    ''' return a 'tab' NtvSingle from a csv file
 40
 41    *parameters*
 42
 43    - **file_name** : name of the csv file
 44    - **single_tab** : boolean (default True) - if True return a 'tab' NtvSingle,
 45    else return a NtvSet.
 46    - **dialect, fmtparams** : parameters of csv.DictReader object'''
 47    with open(file_name, newline='') as csvfile:
 48        reader = csv.DictReader(csvfile, dialect=dialect, **fmtparams)
 49        names = reader.fieldnames
 50        list_ntv_value = [[] for nam in names]
 51        for row in reader:
 52            for ind_field, val in enumerate(list(row.values())):
 53                list_ntv_value[ind_field].append(json.loads(val))
 54    list_ntv = []
 55    for ind_field, field in enumerate(names):
 56        list_ntv.append(
 57            NtvList(list_ntv_value[ind_field], *NtvUtil.from_obj_name(field)[:2]))
 58    if single_tab:
 59        return NtvSingle(NtvList(list_ntv, None, None).to_obj(), None, 'tab')
 60    return NtvList(list_ntv, None, None)
 61
 62
 63def to_csv(file_name, ntv, *args, restval='', extrasaction='raise', dialect='excel', **kwds):
 64    ''' convert a 'tab' NtvSingle into csv file and return the file name
 65
 66    *parameters*
 67
 68    - **file_name** : name of the csv file
 69    - **ntv** : 'tab' NtvSingle to convert
 70    - **args, restval, extrasaction, dialect, kwds** : parameters of csv.DictWriter object'''
 71    if isinstance(ntv, NtvSingle):
 72        ntv_set = Ntv.obj(ntv.ntv_value)
 73    else:
 74        ntv_set = ntv
 75    list_ntv = [Ntv.obj(field) for field in ntv_set]
 76    fieldnames = [ntv_field.json_name(string=True) for ntv_field in list_ntv]
 77    with open(file_name, 'w', newline='') as csvfile:
 78        writer = csv.DictWriter(csvfile, fieldnames=fieldnames, restval=restval,
 79                                extrasaction=extrasaction, dialect=dialect, *args, **kwds)
 80        writer.writeheader()
 81        for i in range(len(list_ntv[0])):
 82            writer.writerow({name: field_ntv[i].to_obj(field_ntv.ntv_type, encoded=True)
 83                             for name, field_ntv in zip(fieldnames, list_ntv)})
 84    return file_name
 85
 86
 87class ShapelyConnec(NtvConnector):
 88    '''NTV connector for geographic location'''
 89
 90    clas_obj = 'geometry'
 91    clas_typ = 'geometry'
 92
 93    @staticmethod
 94    def to_obj_ntv(ntv_value, **kwargs):
 95        ''' convert ntv_value into a shapely geometry object defined by 'type_geo'.
 96
 97        *Parameters*
 98
 99        - **type_geo** : type of geometry (point, multipoint,
100        linestring, multilinestring', polygon, multipolygon)
101        - **ntv_value** : array - coordinates'''
102        from shapely import geometry
103        type_geo = ShapelyConnec.type_geo(ntv_value) if not 'type_geo' in kwargs \
104            or kwargs['type_geo'] == 'geometry' else kwargs['type_geo']
105        return geometry.shape({"type": type_geo,
106                               "coordinates": ntv_value})
107
108    @staticmethod
109    def to_json_ntv(value, name=None, typ=None):
110        ''' convert NTV object (value, name, type) into NTV json (json-value, name, type).
111
112        *Parameters*
113
114        - **typ** : string (default None) - NTV type of geometry (point, multipoint,
115        line, multiline', polygon, multipolygon),
116        - **name** : string (default None) - name of the NTV object
117        - **value** : shapely geometry'''
118        return (Ntv._listed(value.__geo_interface__['coordinates']), name, typ)
119
120    @staticmethod
121    def to_coord(geom):
122        ''' convert shapely geometry into geojson coordinates.'''
123        return Ntv._listed(geom.__geo_interface__['coordinates'])
124
125    @staticmethod
126    def to_geojson(geom):
127        ''' convert shapely geometry into geojson string'''
128        return json.dumps(geom.__geo_interface__)
129
130    @staticmethod
131    def from_geojson(geojson):
132        ''' convert geojson string into shapely geometry.'''
133        from shapely import geometry
134        return geometry.shape(json.loads(geojson))
135
136    @staticmethod
137    def to_geometry(value):
138        '''convert geojson coordinates into shapely geometry'''
139        return ShapelyConnec.to_obj_ntv(value, type_geo=NtvConnector.DIC_GEO[ShapelyConnec.type_geo(value)])
140
141    @staticmethod
142    def type_geo(value):
143        '''return geometry type of the value'''
144        if not value or not isinstance(value, list):
145            return 'not a geometry'
146        val = value[0]
147        if not isinstance(val, list):
148            return 'point'
149        val = val[0]
150        if not isinstance(val, list):
151            return 'line'
152        return 'polygon'
153
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 tab_dataset.field 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 tab_dataset.field 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 tab_dataset.dataset 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 tab_dataset.dataset 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
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 [str(ntv.pointer(index=True, item_idx=ind)),
366                    ['rectangle', name[:-1]]]
367        if not name:
368            name = '<b>::</b>\n'
369        name = name.replace('"', "'")
370        return [str(ntv.pointer(index=True, item_idx=ind)),
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                    [str(ntv.pointer(index=True)), 'normalarrow',
385                     str(ntv_val.pointer(index=True, item_idx=ind))])
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