Source code for node_json_convert

# This file is part of Sympathy for Data.
# Copyright (c) 2019, Combine Control Systems AB
#
# SYMPATHY FOR DATA COMMERCIAL LICENSE
# You should have received a link to the License with Sympathy for Data.
from sympathy.api import node as synode
from sympathy.api.nodeconfig import Port, Ports, Tag, Tags
from sympathy.api import exceptions
from sympathy.api import json


CONVERSION_NODES = [
    'org.sysess.sympathy.json.jsontolist',
    'org.sysess.sympathy.json.jsontodict',
    'org.sysess.sympathy.json.listtojson',
    'org.sysess.sympathy.json.dicttojson',
]




class JsonToDict(synode.Node):
    author = 'Erik der Hagopian'
    icon = 'json_to_dict.svg'
    name = 'Json to Dict'
    description = (
        'Convert Json with a top-level dictionary to a Sympathy dictionary '
        'of Json elements.')
    nodeid = 'org.sysess.sympathy.json.jsontodict'
    tags = Tags(Tag.DataProcessing.Convert)
    related = CONVERSION_NODES

    inputs = Ports([Port.Json('Json', name='json')])
    outputs = Ports([Port.Custom('{json}', 'Json dict', name='json_dict')])

    def execute(self, node_context):
        data = node_context.input['json'].get()
        if not isinstance(data, dict):
            raise exceptions.SyDataError(
                'Conversion from json to {json} requires top-level dict')
        else:
            output = node_context.output['json_dict']
            for k, v in data.items():
                item = json.File()
                item.set(v)
                output[k] = item





class DictToJson(synode.Node):
    author = 'Erik der Hagopian'
    icon = 'dict_to_json.svg'
    name = 'Dict to Json'
    description = (
        'Convert Sympathy dictionary of Json elements to Json with a '
        'top-level dictionary.')
    nodeid = 'org.sysess.sympathy.json.dicttojson'
    tags = Tags(Tag.DataProcessing.Convert)
    related = CONVERSION_NODES

    inputs = Ports([Port.Custom('{json}', 'Json dict', name='json_dict')])
    outputs = Ports([Port.Json('Json', name='json')])

    def execute(self, node_context):
        node_context.output['json'].set(
            {k: v.get() for k, v in node_context.input['json_dict'].items()})





class JsonToList(synode.Node):
    author = 'Erik der Hagopian'
    icon = 'json_to_list.svg'
    name = 'Json to List'
    description = (
        'Convert Json containing a top-level list to a Sympathy list '
        'of Json elements.')
    nodeid = 'org.sysess.sympathy.json.jsontolist'
    tags = Tags(Tag.DataProcessing.Convert)
    related = CONVERSION_NODES

    inputs = Ports([Port.Json('Json', name='json')])
    outputs = Ports([Port.Custom('[json]', 'Json list', name='json_list')])

    def execute(self, node_context):
        data = node_context.input['json'].get()
        if not isinstance(data, list):
            raise exceptions.SyDataError(
                'Conversion from json to [json] requires top-level list')
        else:
            output = node_context.output['json_list']
            for v in data:
                item = json.File()
                item.set(v)
                output.append(item)





class ListToJson(synode.Node):
    author = 'Erik der Hagopian'
    icon = 'list_to_json.svg'
    name = 'List to Json'
    description = (
        'Convert Sympathy list of Json elements to Json with a '
        'top-level list.')
    nodeid = 'org.sysess.sympathy.json.listtojson'
    tags = Tags(Tag.DataProcessing.Convert)
    related = CONVERSION_NODES

    inputs = Ports([Port.Custom('[json]', 'Json list', name='json_list')])
    outputs = Ports([Port.Json('Json', name='json')])

    def execute(self, node_context):
        node_context.output['json'].set(
            [v.get() for v in node_context.input['json_list']])