Source code for node_convert_column_tabledataset

# 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 qt2 as qt_compat
from sympathy.api import node
from sympathy.api.nodeconfig import Ports, Tag, Tags, adjust
from sylib_aml.dataset import DatasetPort
from sylib.converters import TYPE_NAMES

QtCore = qt_compat.QtCore
QtGui = qt_compat.import_module("QtGui")
QtWidgets = qt_compat.import_module("QtWidgets")




class ConvertColumnTypeInTableDataset(node.Node):
    """
    Convert selected columns to specified data type in tabular dataset.
    """

    name = "Convert column type in Table Dataset (Experimental)"
    nodeid = (
        "com.sympathyfordata.advancedmachinelearning."
        "convertcolumntypeintabledataset"
    )
    author = "Jannes Germishuys"
    icon = "convert_column_types_table_ds.svg"
    tags = Tags(Tag.MachineLearning.Processing)

    inputs = Ports([DatasetPort("Dataset", "dataset")])
    outputs = Ports([DatasetPort("Dataset", "dataset")])

    parameters = node.parameters()
    editor = node.editors.multilist_editor(edit=True, mode=False)
    parameters.set_list(
        "columns",
        label="Columns",
        description="Columns that should be converted.",
        value=[],
        editor=editor,
    )
    editor = node.editors.combo_editor(
        options=list(sorted(TYPE_NAMES.values())))
    parameters.set_string(
        "types",
        label="Target type",
        description="The type that these columns should be converted to.",
        value_names=["text"],
        editor=editor,
    )

    def adjust_parameters(self, node_context):
        adjust(node_context.parameters["columns"], node_context.input[0])

    def execute(self, node_context):
        input_ds = node_context.input["dataset"]
        input_ds.load()
        input_ds = input_ds.get_ds()
        all_cols = node_context.input[0].names(kind='cols')
        columns_choice = node_context.parameters['columns'].selected_names(
            all_cols)
        if input_ds is not None:
            update_dict = dict(
                zip(
                    columns_choice,
                    [node_context.parameters["types"].value]
                    * len(columns_choice),
                )
            )

            for col, typ in update_dict.items():
                input_ds["column_config"][col]["dtype"] = typ
        else:
            input_ds = {}

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