Source code for node_table_to_dataset

# This file is part of Sympathy for Data.
# Copyright (c) 2021 Combine Control Systems
#
# SYMPATHY FOR DATA COMMERCIAL LICENSE
# You should have received a link to the License with Sympathy for Data.

from sympathy.api import node, exceptions
from sympathy.api.nodeconfig import Port, Ports, Tag, Tags
from sylib_aml.dataset import DatasetPort




class TablesToDataset(node.Node):
    """
    Convert Table(s) to Dataset.
    """

    name = "Tables to Dataset (Experimental)"
    nodeid = "com.sympathyfordata.advancedmachinelearning.tablestodataset"
    author = "Jannes Germishuys"
    icon = "tables2ds.svg"
    tags = Tags(Tag.MachineLearning.Processing)

    inputs = Ports(
        [
            Port.Tables("Tables", name="in_tables"),
            Port.Datasources("Datasources", name="source"),
        ]
    )
    outputs = Ports([DatasetPort("Dataset", "dataset")])

    def execute(self, node_context):

        input_tables = node_context.input["in_tables"]
        output_paths = [s.decode_path() for s in node_context.input["source"]]
        tab_names = []

        if len(input_tables) == 0:
            raise exceptions.SyDataError("Empty tables")

        assert len(input_tables) == len(
            output_paths
        ), "Length of input tables should match length of output paths"

        for i, tab in enumerate(input_tables):
            tab.to_csv(output_paths[i], delimiter=",")
            columns = tab.column_names()
            tab_names.append(tab.name)

        column_config = {
            col_name: {"dtype": "", "transforms": [],
                       "transforms_values": []}
            for col_name in columns
        }

        ds = {
            "dstype": "table",
            "paths": output_paths,
            "tab_names": tab_names,
            "reader_config": {},
            "column_config": column_config,
        }

        json = node_context.output["dataset"]
        json.set_ds(ds)
        json.save()