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='', encoding="utf-8") 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='', encoding="utf-8") 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 'type_geo' not 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,
140                                        type_geo=NtvConnector.DIC_GEO[
141                                            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
156
157class CborConnec(NtvConnector):
158    '''NTV connector for binary data'''
159
160    clas_obj = 'bytes'
161    clas_typ = '$cbor'
162
163    @staticmethod
164    def to_obj_ntv(ntv_value, **kwargs):
165        ''' convert json ntv_value into a binary CBOR object (no parameters).'''
166        import cbor2
167        return cbor2.dumps(ntv_value, datetime_as_timestamp=True,
168                           timezone=datetime.timezone.utc, canonical=False,
169                           date_as_datetime=True)
170
171    @staticmethod
172    def to_json_ntv(value, name=None, typ=None):
173        ''' convert NTV binary object (value, name, type) into NTV json (json-value, name, type).
174
175        *Parameters*
176
177        - **typ** : string (default None) - type of the NTV object,
178        - **name** : string (default None) - name of the NTV object
179        - **value** : binary data'''
180        import cbor2
181        return (cbor2.loads(value), name, typ)
182
183
184class NfieldConnec(NtvConnector):
185    '''NTV connector for NTV Field data'''
186
187    clas_obj = 'Nfield'
188    clas_typ = 'field'
189
190    @staticmethod
191    def to_obj_ntv(ntv_value, **kwargs):
192        ''' convert json ntv_value into a NTV Field object (no parameters).'''
193        from tab_dataset.field import Nfield
194        ntv = Ntv.obj(ntv_value)
195        return Nfield.from_ntv(ntv)
196
197    @staticmethod
198    def to_json_ntv(value, name=None, typ=None):
199        ''' convert NTV Field object (value, name, type) into NTV json (json-value, name, type).
200
201        *Parameters*
202
203        - **typ** : string (default None) - type of the NTV object,
204        - **name** : string (default None) - name of the NTV object
205        - **value** : NTV Field values (default format)'''
206        return (value.to_ntv(name=True).to_obj(), name,
207                NfieldConnec.clas_typ if not typ else typ)
208
209
210class SfieldConnec(NtvConnector):
211    '''NTV connector for simple Field data'''
212
213    clas_obj = 'Sfield'
214    clas_typ = 'field'
215
216    @staticmethod
217    def to_obj_ntv(ntv_value, **kwargs):
218        ''' convert json ntv_value into a simple Field object (no parameters).'''
219        from tab_dataset.field import Sfield
220        ntv = Ntv.obj(ntv_value)
221        return Sfield.from_ntv(ntv)
222
223    @staticmethod
224    def to_json_ntv(value, name=None, typ=None):
225        ''' convert simple Field object (value, name, type) into NTV json (json-value, name, type).
226
227        *Parameters*
228
229        - **typ** : string (default None) - type of the NTV object,
230        - **name** : string (default None) - name of the NTV object
231        - **value** : simple Field values (default format)'''
232        return (value.to_ntv(name=True).to_obj(), name,
233                NfieldConnec.clas_typ if not typ else typ)
234
235
236class NdatasetConnec(NtvConnector):
237    '''NTV connector for NTV Dataset data'''
238
239    clas_obj = 'Ndataset'
240    clas_typ = 'tab'
241
242    @staticmethod
243    def to_obj_ntv(ntv_value, **kwargs):
244        ''' convert json ntv_value into a NTV Dataset object (no parameters).'''
245        from tab_dataset.dataset import Ndataset
246
247        ntv = Ntv.obj(ntv_value)
248        return Ndataset.from_ntv(ntv)
249
250    @staticmethod
251    def to_json_ntv(value, name=None, typ=None):
252        ''' convert NTV Dataset object (value, name, type) into NTV json (json-value, name, type).
253
254        *Parameters*
255
256        - **typ** : string (default None) - type of the NTV object,
257        - **name** : string (default None) - name of the NTV object
258        - **value** : NTV Dataset values'''
259        return (value.to_ntv().to_obj(), name,
260                NdatasetConnec.clas_typ if not typ else typ)
261
262
263class SdatasetConnec(NtvConnector):
264    '''NTV connector for simple Dataset data'''
265
266    clas_obj = 'Sdataset'
267    clas_typ = 'tab'
268
269    @staticmethod
270    def to_obj_ntv(ntv_value, **kwargs):
271        ''' convert json ntv_value into a simple Dataset object (no parameters).'''
272        from tab_dataset.dataset import Sdataset
273
274        ntv = Ntv.obj(ntv_value)
275        return Sdataset.from_ntv(ntv)
276
277    @staticmethod
278    def to_json_ntv(value, name=None, typ=None):
279        ''' convert simple Dataset object (value, name, type) into NTV json
280        (json-value, name, type).
281
282        *Parameters*
283
284        - **typ** : string (default None) - type of the NTV object,
285        - **name** : string (default None) - name of the NTV object
286        - **value** : simple Dataset values'''
287        return (value.to_ntv().to_obj(), name,
288                SdatasetConnec.clas_typ if not typ else typ)
289
290
291class MermaidConnec(NtvConnector):
292    '''NTV connector for Mermaid diagram'''
293
294    clas_obj = 'Mermaid'
295    clas_typ = '$mermaid'
296
297    @staticmethod
298    def to_obj_ntv(ntv_value, **kwargs):
299        ''' convert ntv_value into a mermaid flowchart
300
301        *Parameters*
302
303        - **title**: String (default '') - title of the flowchart
304        - **disp**: Boolean (default False) - if true, return a display else return
305        a mermaid text diagram
306        - **row**: Boolean (default False) - if True, add the node row
307        - **leaves**: Boolean (default False) - if True, add the leaf row
308        '''
309        from base64 import b64encode
310        from IPython.display import Image, display
311
312        option = {'title': '', 'disp': False, 'row': False,
313                  'leaves': False} | kwargs
314        diagram = MermaidConnec.diagram
315        link = MermaidConnec._mermaid_link
316        ntv = Ntv.obj(ntv_value)
317        node_link = {'nodes': [], 'links': []}
318        dic_node = {}
319        if option['leaves']:
320            nodes = [node for node in NtvTree(
321                ntv) if not isinstance(node.val, list)]
322            dic_node = {node: row for row, node in enumerate(nodes)}
323        link(ntv, None, node_link, option['row'], dic_node, None)
324        mermaid_json = {option['title'] + ':$flowchart': {
325            'orientation': 'top-down',
326            'node::': {node[0]: node[1] for node in node_link['nodes']},
327            'link::': node_link['links']}}
328        if option['disp']:
329            return display(Image(url="https://mermaid.ink/img/" +
330                                 b64encode(diagram(mermaid_json).encode("ascii")).decode("ascii")))
331        return diagram(mermaid_json)
332
333    @staticmethod
334    def diagram(json_diag):
335        '''create a mermaid code from a mermaid json'''
336        ntv = Ntv.obj(json_diag)
337        erdiagram = MermaidConnec._er_diagram
338        flowchart = MermaidConnec._flowchart
339        diag_type = ntv.type_str[1:]
340        diag_txt = '---\ntitle: ' + ntv.name + '\n---\n' if ntv.name else ''
341        diag_txt += diag_type
342        match diag_type:
343            case 'erDiagram':
344                diag_txt += erdiagram(ntv)
345            case 'flowchart':
346                diag_txt += flowchart(ntv)
347        return diag_txt
348
349    @staticmethod
350    def _mermaid_node(ntv, def_typ_str, num, dic_node, ind):
351        '''create and return a node'''
352        j_name, j_sep, j_type = ntv.json_name(def_typ_str)
353        name = ''
354        if j_name:
355            name += '<b>' + j_name + '</b>\n'
356        if j_type:
357            name += j_type + '\n'
358        if ntv in dic_node:
359            num += ' ' + str(dic_node[ntv])
360        if num:
361            name += '<i>' + num + '</i>\n'
362        elif isinstance(ntv, NtvSingle):
363            if isinstance(ntv.val, str):
364                name += '<i>' + ntv.val + '</i>\n'
365            else:
366                name += '<i>' + json.dumps(ntv.val) + '</i>\n'
367            return [str(ntv.pointer(index=True, item_idx=ind)),
368                    ['rectangle', name[:-1]]]
369        if not name:
370            name = '<b>::</b>\n'
371        name = name.replace('"', "'")
372        return [str(ntv.pointer(index=True, item_idx=ind)),
373                ['roundedge', name[:-1]]]
374
375    @staticmethod
376    def _mermaid_link(ntv, def_typ_str, node_link, row, dic_node, indic):
377        '''add nodes and links from ntv in node_link '''
378        num = str(len(node_link['nodes'])) if row else ''
379        node_link['nodes'].append(MermaidConnec._mermaid_node(
380            ntv, def_typ_str, num, dic_node, indic))
381        if isinstance(ntv, NtvList):
382            for ind, ntv_val in enumerate(ntv):
383                MermaidConnec._mermaid_link(ntv_val, ntv.type_str,
384                                            node_link, row, dic_node, ind)
385                node_link['links'].append(
386                    [str(ntv.pointer(index=True)), 'normalarrow',
387                     str(ntv_val.pointer(index=True, item_idx=ind))])
388
389    @staticmethod
390    def _flowchart(ntv):
391        orientation = {'top-down': 'TD', 'top-bottom': 'TB',
392                       'bottom-top': 'BT', 'right-left': 'RL', 'left-right': 'LR'}
393        fcnode = MermaidConnec._fc_node
394        fclink = MermaidConnec._fc_link
395        flc = Ntv.obj(ntv.val)
396        diag_txt = ' ' + orientation[flc['orientation'].val]
397        for node in flc['node']:
398            diag_txt += fcnode(node)
399        for link in flc['link']:
400            diag_txt += fclink(link)
401        return diag_txt + '\n'
402
403    @staticmethod
404    def _fc_link(link):
405        link_t = {'normal': ' ---', 'normalarrow': ' -->',
406                  'dotted': ' -.-', 'dottedarrow': ' -.->'}
407        link_txt = '\n    ' + str(link[0].val) + link_t[link[1].val]
408        if len(link) == 4:
409            link_txt += '|' + link[3].val + '|'
410        return link_txt + ' ' + str(link[2].val)
411
412    @staticmethod
413    def _fc_node(node):
414        shape_l = {'rectangle': '[', 'roundedge': '(', 'stadium': '(['}
415        shape_r = {'rectangle': ']', 'roundedge': ')', 'stadium': '])'}
416        return '\n    ' + node.name + shape_l[node[0].val] + '"' + \
417               node[1].val.replace('"', "'") + '"' + shape_r[node[0].val]
418
419    @staticmethod
420    def _er_diagram(ntv):
421        erentity = MermaidConnec._er_entity
422        errelation = MermaidConnec._er_relation
423        diag_txt = ''
424        erd = Ntv.obj(ntv.val)
425        for entity in erd['entity']:
426            diag_txt += erentity(entity)
427        for relation in erd['relationship']:
428            diag_txt += errelation(relation)
429        return diag_txt
430
431    @staticmethod
432    def _er_entity(entity):
433        ent_txt = '\n    ' + entity.name + ' {'
434        for att in entity:
435            ent_txt += '\n        ' + att[0].val + ' ' + att[1].val
436            if len(att) > 2:
437                if att[2].val in ('PK', 'FK', 'UK'):
438                    ent_txt += ' ' + att[2].val
439                else:
440                    ent_txt += ' "' + att[2].val + '"'
441            if len(att) > 3:
442                ent_txt += ' "' + att[3].val + '"'
443        return ent_txt + '\n    }'
444
445    @staticmethod
446    def _er_relation(rel):
447        rel_left = {'exactly one': ' ||', 'zero or one': ' |o',
448                    'zero or more': ' }o', 'one or more': ' }|'}
449        rel_right = {'exactly one': '|| ', 'zero or one': 'o| ',
450                     'zero or more': 'o{ ', 'one or more': '|{ '}
451        identif = {'identifying': '--', 'non-identifying': '..'}
452        rel_txt = '\n    ' + rel[0].val + rel_left[rel[1].val] + \
453            identif[rel[2].val] + rel_right[rel[3].val] + rel[4].val
454        if len(rel) > 5:
455            rel_txt += ' : ' + rel[5].val
456        return rel_txt