# 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.
import warnings
import sklearn.exceptions
import sklearn.model_selection
import dask.dataframe as dd
import numpy as np
from sympathy.api import node
from sympathy.api.nodeconfig import Port, Ports, Tag, Tags
from sympathy.api.exceptions import SyDataError, sywarn
from sympathy.utils.pip_util import import_optional
from sylib.machinelearning.model import ModelPort
from sylib_aml.dataset import DatasetPort, prefix_rename
from sylib.machinelearning.utility import table_to_array
import inspect
import functools
from ast import literal_eval
from sympathy.api import qt2 as qt
QtWidgets = qt.QtWidgets
def _torch():
return import_optional("torch", group="torch")
@functools.lru_cache()
def mem_rows(ds, model):
ds = literal_eval(ds)
reader_config = dict(ds["reader_config"])
prefix = reader_config.pop('prefix', None)
df = dd.read_csv(
ds["paths"],
**reader_config,
include_path_column=True,
assume_missing=True,
blocksize=1000e6,
usecols=[
col
for col in ds["column_config"].keys()
if ds["column_config"][col]["orig"]
]
if reader_config["header"]
else [
i
for i in range(len(ds["column_config"].keys()))
if ds["column_config"][list(ds["column_config"].keys())[i]]["orig"]
],
)
prefix_rename(df, prefix)
# Repartition step
if "iterator_train" in model.__dict__ and df.npartitions >= 100:
df = df.repartition(partition_size="1000MB")
# # Persist changes to speed up future computations
df = df.persist()
return (df, df.npartitions, len(df.get_partition(0)),
df[ds["output_col"]].nunique())
[docs]
class FitDataSet(node.Node):
"""
Creates a fitted model object from model input and dataset representation
"""
name = "Fit Dataset (Experimental)"
nodeid = "com.sympathyfordata.advancedmachinelearning.fitdataset"
author = "Jannes Germishuys"
description = "Trains a model by fitting to dataset."
icon = "fit_ds.svg"
tags = Tags(Tag.MachineLearning.Apply)
inputs = Ports(
[
ModelPort("Input model", "in-model"),
DatasetPort("Input dataset", "X"),
Port.Custom("table", "sample_weights", name="sample_weights",
n=(0, 1)),
]
)
outputs = Ports([ModelPort("Output model", "out-model")])
def execute(self, node_context):
from sylib_aml.amlnets import ImgDataSet, TabDataSet, SyConcatDataset, SubsetRandomSampler
data = node_context.input["X"]
in_model = node_context.input["in-model"]
out_model = node_context.output["out-model"]
sample_weights = node_context.input.group("sample_weights")
if len(sample_weights) > 0:
sample_weight = table_to_array(sample_weights[0], unitary=True)
else:
sample_weight = None
out_model.source(in_model)
out_model.load()
model = out_model.get_skl()
data.load()
ds = data.get_ds()
if ds is None or model is None:
raise SyDataError("Empty dataset and/or model")
if ds["dstype"] == "table":
df, npartitions, nrows, ncategories = mem_rows(str(ds), model)
datasets = list(
map(
lambda x: TabDataSet(
x,
df,
ds["reader_config"],
ds["column_config"],
label_col=ds["output_col"],
),
ds["paths"],
)
)
cds = SyConcatDataset(datasets)
X = cds
Y = None
if not hasattr(model, 'train_split'):
model.train_split = None
if model.train_split is not None:
sampler = SubsetRandomSampler(
indices=tuple(ds["indices"]), block_size=nrows, cv=True)
model.train_split.cv = sampler
else:
sampler = SubsetRandomSampler(
indices=tuple(ds["indices"]), block_size=nrows, cv=False)
model.iterator_train__sampler = sampler
model.module__output_dim = ncategories.compute()
model.module__virtual_batch_size = nrows
elif ds["dstype"] == "image":
if "To Tensor" not in ds["transforms"]:
ds["transforms"].extend(["To Tensor"])
ds["transforms_values"].extend([{}])
cds = ImgDataSet(
ds["paths"], ds["transforms"], ds["transforms_values"],
ds["labels"],
preview=False
)
X = _torch().utils.data.Subset(cds, indices=ds["indices"])
Y = None
npartitions = 0
else:
raise TypeError("This data type is not supported.")
# Check if we can fit in a progress_update function
kwargs = {}
args = inspect.getfullargspec(model.fit).args
if "progress_fn" in args:
kwargs["progress_fn"] = lambda i: self.set_progress(i)
with warnings.catch_warnings():
warnings.simplefilter(
"ignore", sklearn.exceptions.ConvergenceWarning)
# Filter user warning related to named tensors
warnings.filterwarnings("ignore", category=UserWarning)
if Y is None:
if sample_weight is None:
if "iterator_train" in model.__dict__:
if ds["dstype"] == "table":
model.module__input_dim = (
len(ds["column_config"].keys())
- sum(
ds["column_config"][col]["exclude"]
for col in ds["column_config"].keys()
)
- 1
)
if npartitions > 1:
model.fit(X, Y)
else:
try:
X_train, Y = cds[tuple(ds["indices"])]
model.iterator_train__sampler = None
model.batch_size = 1024
model.fit(X_train, Y)
except MemoryError(
"Data too large to be loaded into "
"memory."):
pass
else:
model.fit(X, Y)
else:
if ds["dstype"] == "table":
model.module__input_dim = (
len(ds["column_config"].keys())
- sum(
ds["column_config"][col]["exclude"]
for col in ds["column_config"].keys()
)
- 1
)
# Add Y if possible
sywarn(
"This model requires an explicit Y, "
"attempting to load into memory...")
try:
X_train, Y = cds[tuple(ds["indices"])]
model.iterator_train__sampler = None
model.batch_size = 1024
model.fit(X_train, Y)
except MemoryError(
"Data too large to be loaded into "
"memory."):
pass
else:
# Add Y if possible
sywarn(
"This model requires an explicit Y, "
"attempting to load into memory...")
try:
X_train, Y = cds[ds["indices"]]
model.fit(X_train, Y)
except MemoryError(
"Data too large to be loaded into "
"memory."):
pass
else:
model.fit(X, sample_weight=sample_weight)
else:
if sample_weight is None:
model.fit(X, Y)
else:
model.fit(X, Y, sample_weight=sample_weight)
desc = out_model.get_desc()
# desc.set_x_names(X_tbl.column_names())
# if Y is not None:
# desc.set_y_names(Y_tbl.column_names())
desc.post_fit(model)
out_model.save()
[docs]
class PredictDataset(node.Node):
name = "Predict Dataset (Experimental)"
nodeid = "com.sympathyfordata.advancedmachinelearning.predict_dataset"
author = "Jannes Germishuys"
icon = "predict_ds.svg"
tags = Tags(Tag.MachineLearning.Apply)
description = "Uses a model to predict Y given X"
parameters = node.parameters()
inputs = Ports(
[ModelPort("Input model", "in-model"),
DatasetPort("Dataset", name="dataset")]
)
outputs = Ports(
[
Port.Table("Table Containing Predictions", "predictions"),
Port.Table("Table Containing True Values", "ground_truth"),
]
)
def execute(self, node_context):
from sylib_aml.amlnets import ImgDataSet, TabDataSet, SyConcatDataset, SubsetRandomSampler
in_model = node_context.input["in-model"]
pred_table = node_context.output["predictions"]
true_table = node_context.output["ground_truth"]
data = node_context.input["dataset"]
in_model.load()
model = in_model.get_skl()
data.load()
ds = data.get_ds()
if ds is None or model is None:
raise SyDataError("Empty dataset and/or model")
if ds["dstype"] == "table":
df, npartitions, nrows, ncategories = mem_rows(str(ds), model)
datasets = list(
map(
lambda x: TabDataSet(
x,
df,
ds["reader_config"],
ds["column_config"],
label_col=ds["output_col"],
),
ds["paths"],
)
)
cds = SyConcatDataset(datasets)
X = cds
# Y = None
sampler = SubsetRandomSampler(
indices=tuple(ds["indices"]), block_size=nrows, train=False
)
model.iterator_valid__sampler = sampler
elif ds["dstype"] == "image":
if "To Tensor" not in ds["transforms"]:
ds["transforms"].extend(["To Tensor"])
ds["transforms_values"].extend([{}])
cds = ImgDataSet(
ds["paths"], ds["transforms"], ds["transforms_values"],
ds["labels"], preview=False
)
if len(ds["indices"]) > 0:
test_ds = _torch().utils.data.Subset(cds, indices=ds["indices"])
else:
test_ds = _torch().utils.data.Subset(cds, indices=ds["indices"])
X = test_ds
# Y = None
else:
raise TypeError("This data type is not supported.")
with warnings.catch_warnings():
warnings.simplefilter(
"ignore", sklearn.exceptions.ConvergenceWarning)
# Filter user warning related to named tensors
warnings.filterwarnings("ignore", category=UserWarning)
if "iterator_valid" in model.__dict__:
model.batch_size = None
Y_pred = model.predict(X)
if ds["dstype"] == "table":
_, Y_test = cds[tuple(ds["indices"])]
Y_test = Y_test.numpy()
else:
Y_test = np.array([item[1] for item in iter(X)])
else:
# Add Y if possible
sywarn(
"This model requires an explicit Y, attempting to load "
"into memory...")
try:
model.iterator_valid__sampler = None
model.batch_size = 1024
X, Y_test = cds[tuple(ds["indices"])]
Y_test = Y_test.numpy()
Y_pred = model.predict(X)
except TypeError as exc:
raise SyDataError(
"Model does not implement the 'predict' function"
) from exc
return
if len(Y_pred.shape) < 2:
Y_pred = Y_pred.reshape(Y_pred.shape + (1,))
if len(Y_test.shape) < 2:
Y_test = Y_test.reshape(Y_test.shape + (1,))
if "labels" in ds:
y_names = [str(ds["labels"])]
else:
y_names = [str(ds["output_col"])]
if y_names is None:
y_names = ["y{0}".format(i) for i in range(Y_pred.shape[1])]
for i, name in enumerate(y_names):
pred_table.set_column_from_array(name, Y_pred[:, i])
true_table.set_column_from_array(name, Y_test[:, i])
# json = node_context.output["dataset"]
# json.set_ds(ds)
# json.save()
