Source code for node_json_convert

# This file is part of Sympathy for Data.
# Copyright (c) 2019, Combine Control Systems AB
#
# Sympathy for Data is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# Sympathy for Data is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Sympathy for Data.  If not, see <http://www.gnu.org/licenses/>.
from sympathy.api import node as synode
from sympathy.api import node_helper
from sympathy.api.nodeconfig import Port, Ports, Tag, Tags, deprecated_node
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



@deprecated_node(repl='Json to Dict inside a Lambda + Map')
@node_helper.list_node_decorator(['json'], ['json_dict'])
class JsonsToDicts(JsonToDict):
    nodeid = 'org.sysess.sympathy.json.jsonstodicts'
    name = 'Jsons to Dicts'




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)



@deprecated_node(repl='Json to List inside a Lambda + Map')
@node_helper.list_node_decorator(['json'], ['json_list'])
class JsonsToLists(JsonToList):
    nodeid = 'org.sysess.sympathy.json.jsonstolists'
    name = 'Jsons to Lists'




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']])