# Copyright 2021-2024 Avaiga Private Limited # # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the # specific language governing permissions and limitations under the License. import os import pathlib from datetime import datetime from time import sleep import numpy as np import pandas as pd import pytest from pandas.testing import assert_frame_equal from taipy.config.common.scope import Scope from taipy.config.config import Config from taipy.config.exceptions.exceptions import InvalidConfigurationId from taipy.core.data._data_manager import _DataManager from taipy.core.data.csv import CSVDataNode from taipy.core.data.data_node_id import DataNodeId from taipy.core.data.operator import JoinOperator, Operator from taipy.core.exceptions.exceptions import InvalidExposedType, NoData @pytest.fixture(scope="function", autouse=True) def cleanup(): yield path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/temp.csv") if os.path.isfile(path): os.remove(path) class MyCustomObject: def __init__(self, id, integer, text): self.id = id self.integer = integer self.text = text class TestCSVDataNode: def test_create(self): path = "data/node/path" dn = CSVDataNode( "foo_bar", Scope.SCENARIO, properties={"path": path, "has_header": False, "name": "super name"} ) assert isinstance(dn, CSVDataNode) assert dn.storage_type() == "csv" assert dn.config_id == "foo_bar" assert dn.name == "super name" assert dn.scope == Scope.SCENARIO assert dn.id is not None assert dn.owner_id is None assert dn.last_edit_date is None assert dn.job_ids == [] assert not dn.is_ready_for_reading assert dn.path == path assert dn.has_header is False assert dn.exposed_type == "pandas" with pytest.raises(InvalidConfigurationId): dn = CSVDataNode( "foo bar", Scope.SCENARIO, properties={"path": path, "has_header": False, "name": "super name"} ) def test_modin_deprecated_in_favor_of_pandas(self): path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") # Create CSVDataNode with modin exposed_type csv_data_node_as_modin = CSVDataNode("bar", Scope.SCENARIO, properties={"path": path, "exposed_type": "modin"}) assert csv_data_node_as_modin.properties["exposed_type"] == "pandas" data_modin = csv_data_node_as_modin.read() assert isinstance(data_modin, pd.DataFrame) def test_get_user_properties(self, csv_file): dn_1 = CSVDataNode("dn_1", Scope.SCENARIO, properties={"path": "data/node/path"}) assert dn_1._get_user_properties() == {} dn_2 = CSVDataNode( "dn_2", Scope.SCENARIO, properties={ "exposed_type": "numpy", "default_data": "foo", "default_path": csv_file, "has_header": False, "foo": "bar", }, ) # exposed_type, default_data, default_path, path, has_header, sheet_name are filtered out assert dn_2._get_user_properties() == {"foo": "bar"} def test_new_csv_data_node_with_existing_file_is_ready_for_reading(self): not_ready_dn_cfg = Config.configure_data_node("not_ready_data_node_config_id", "csv", path="NOT_EXISTING.csv") not_ready_dn = _DataManager._bulk_get_or_create([not_ready_dn_cfg])[not_ready_dn_cfg] assert not not_ready_dn.is_ready_for_reading path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") ready_dn_cfg = Config.configure_data_node("ready_data_node_config_id", "csv", path=path) ready_dn = _DataManager._bulk_get_or_create([ready_dn_cfg])[ready_dn_cfg] assert ready_dn.is_ready_for_reading @pytest.mark.parametrize( ["properties", "exists"], [ ({}, False), ({"default_data": ["foo", "bar"]}, True), ], ) def test_create_with_default_data(self, properties, exists): dn = CSVDataNode("foo", Scope.SCENARIO, DataNodeId("dn_id"), properties=properties) assert os.path.exists(dn.path) is exists def test_read_with_header_pandas(self): not_existing_csv = CSVDataNode("foo", Scope.SCENARIO, properties={"path": "WRONG.csv", "has_header": True}) with pytest.raises(NoData): assert not_existing_csv.read() is None not_existing_csv.read_or_raise() path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") # # Create CSVDataNode without exposed_type (Default is pandas.DataFrame) csv_data_node_as_pandas = CSVDataNode("bar", Scope.SCENARIO, properties={"path": path}) data_pandas = csv_data_node_as_pandas.read() assert isinstance(data_pandas, pd.DataFrame) assert len(data_pandas) == 10 assert np.array_equal(data_pandas.to_numpy(), pd.read_csv(path).to_numpy()) def test_read_with_header_numpy(self): path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") # Create CSVDataNode with numpy exposed_type csv_data_node_as_numpy = CSVDataNode( "bar", Scope.SCENARIO, properties={"path": path, "has_header": True, "exposed_type": "numpy"} ) data_numpy = csv_data_node_as_numpy.read() assert isinstance(data_numpy, np.ndarray) assert len(data_numpy) == 10 assert np.array_equal(data_numpy, pd.read_csv(path).to_numpy()) def test_read_with_header_custom_exposed_type(self): path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") csv_data_node_as_pandas = CSVDataNode("bar", Scope.SCENARIO, properties={"path": path}) data_pandas = csv_data_node_as_pandas.read() path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") # Create the same CSVDataNode but with custom exposed_type csv_data_node_as_custom_object = CSVDataNode( "bar", Scope.SCENARIO, properties={"path": path, "exposed_type": MyCustomObject} ) data_custom = csv_data_node_as_custom_object.read() assert isinstance(data_custom, list) assert len(data_custom) == 10 for (_, row_pandas), row_custom in zip(data_pandas.iterrows(), data_custom): assert isinstance(row_custom, MyCustomObject) assert row_pandas["id"] == row_custom.id assert str(row_pandas["integer"]) == row_custom.integer assert row_pandas["text"] == row_custom.text def test_read_without_header(self): not_existing_csv = CSVDataNode("foo", Scope.SCENARIO, properties={"path": "WRONG.csv", "has_header": False}) with pytest.raises(NoData): assert not_existing_csv.read() is None not_existing_csv.read_or_raise() path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") # Create CSVDataNode without exposed_type (Default is pandas.DataFrame) csv_data_node_as_pandas = CSVDataNode("bar", Scope.SCENARIO, properties={"path": path, "has_header": False}) data_pandas = csv_data_node_as_pandas.read() assert isinstance(data_pandas, pd.DataFrame) assert len(data_pandas) == 11 assert np.array_equal(data_pandas.to_numpy(), pd.read_csv(path, header=None).to_numpy()) # Create CSVDataNode with numpy exposed_type csv_data_node_as_numpy = CSVDataNode( "qux", Scope.SCENARIO, properties={"path": path, "has_header": False, "exposed_type": "numpy"} ) data_numpy = csv_data_node_as_numpy.read() assert isinstance(data_numpy, np.ndarray) assert len(data_numpy) == 11 assert np.array_equal(data_numpy, pd.read_csv(path, header=None).to_numpy()) # Create the same CSVDataNode but with custom exposed_type csv_data_node_as_custom_object = CSVDataNode( "quux", Scope.SCENARIO, properties={"path": path, "has_header": False, "exposed_type": MyCustomObject} ) data_custom = csv_data_node_as_custom_object.read() assert isinstance(data_custom, list) assert len(data_custom) == 11 for (_, row_pandas), row_custom in zip(data_pandas.iterrows(), data_custom): assert isinstance(row_custom, MyCustomObject) assert row_pandas[0] == row_custom.id assert str(row_pandas[1]) == row_custom.integer assert row_pandas[2] == row_custom.text @pytest.mark.parametrize( "content", [ ([{"a": 11, "b": 22, "c": 33}, {"a": 44, "b": 55, "c": 66}]), (pd.DataFrame([{"a": 11, "b": 22, "c": 33}, {"a": 44, "b": 55, "c": 66}])), ([[11, 22, 33], [44, 55, 66]]), ], ) def test_append(self, csv_file, default_data_frame, content): csv_dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": csv_file}) assert_frame_equal(csv_dn.read(), default_data_frame) csv_dn.append(content) assert_frame_equal( csv_dn.read(), pd.concat([default_data_frame, pd.DataFrame(content, columns=["a", "b", "c"])]).reset_index(drop=True), ) @pytest.mark.parametrize( "content,columns", [ ([{"a": 11, "b": 22, "c": 33}, {"a": 44, "b": 55, "c": 66}], None), ([[11, 22, 33], [44, 55, 66]], None), ([[11, 22, 33], [44, 55, 66]], ["e", "f", "g"]), ], ) def test_write(self, csv_file, default_data_frame, content, columns): csv_dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": csv_file}) assert np.array_equal(csv_dn.read().values, default_data_frame.values) if not columns: csv_dn.write(content) df = pd.DataFrame(content) else: csv_dn.write_with_column_names(content, columns) df = pd.DataFrame(content, columns=columns) assert np.array_equal(csv_dn.read().values, df.values) csv_dn.write(None) assert len(csv_dn.read()) == 0 def test_write_with_different_encoding(self, csv_file): data = pd.DataFrame([{"≥a": 1, "b": 2}]) utf8_dn = CSVDataNode("utf8_dn", Scope.SCENARIO, properties={"default_path": csv_file}) utf16_dn = CSVDataNode("utf16_dn", Scope.SCENARIO, properties={"default_path": csv_file, "encoding": "utf-16"}) # If a file is written with utf-8 encoding, it can only be read with utf-8, not utf-16 encoding utf8_dn.write(data) assert np.array_equal(utf8_dn.read(), data) with pytest.raises(UnicodeError): utf16_dn.read() # If a file is written with utf-16 encoding, it can only be read with utf-16, not utf-8 encoding utf16_dn.write(data) assert np.array_equal(utf16_dn.read(), data) with pytest.raises(UnicodeError): utf8_dn.read() def test_set_path(self): dn = CSVDataNode("foo", Scope.SCENARIO, properties={"default_path": "foo.csv"}) assert dn.path == "foo.csv" dn.path = "bar.csv" assert dn.path == "bar.csv" def test_read_write_after_modify_path(self): path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") new_path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/temp.csv") dn = CSVDataNode("foo", Scope.SCENARIO, properties={"default_path": path}) read_data = dn.read() assert read_data is not None dn.path = new_path with pytest.raises(FileNotFoundError): dn.read() dn.write(read_data) assert dn.read().equals(read_data) def test_pandas_exposed_type(self): path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": path, "exposed_type": "pandas"}) assert isinstance(dn.read(), pd.DataFrame) def test_filter_pandas_exposed_type(self, csv_file): dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": csv_file, "exposed_type": "pandas"}) dn.write( [ {"foo": 1, "bar": 1}, {"foo": 1, "bar": 2}, {"foo": 1}, {"foo": 2, "bar": 2}, {"bar": 2}, ] ) # Test datanode indexing and slicing assert dn["foo"].equals(pd.Series([1, 1, 1, 2, None])) assert dn["bar"].equals(pd.Series([1, 2, None, 2, 2])) assert dn[:2].equals(pd.DataFrame([{"foo": 1.0, "bar": 1.0}, {"foo": 1.0, "bar": 2.0}])) # Test filter data filtered_by_filter_method = dn.filter(("foo", 1, Operator.EQUAL)) filtered_by_indexing = dn[dn["foo"] == 1] expected_data = pd.DataFrame([{"foo": 1.0, "bar": 1.0}, {"foo": 1.0, "bar": 2.0}, {"foo": 1.0}]) assert_frame_equal(filtered_by_filter_method.reset_index(drop=True), expected_data) assert_frame_equal(filtered_by_indexing.reset_index(drop=True), expected_data) filtered_by_filter_method = dn.filter(("foo", 1, Operator.NOT_EQUAL)) filtered_by_indexing = dn[dn["foo"] != 1] expected_data = pd.DataFrame([{"foo": 2.0, "bar": 2.0}, {"bar": 2.0}]) assert_frame_equal(filtered_by_filter_method.reset_index(drop=True), expected_data) assert_frame_equal(filtered_by_indexing.reset_index(drop=True), expected_data) filtered_by_filter_method = dn.filter(("bar", 2, Operator.EQUAL)) filtered_by_indexing = dn[dn["bar"] == 2] expected_data = pd.DataFrame([{"foo": 1.0, "bar": 2.0}, {"foo": 2.0, "bar": 2.0}, {"bar": 2.0}]) assert_frame_equal(filtered_by_filter_method.reset_index(drop=True), expected_data) assert_frame_equal(filtered_by_indexing.reset_index(drop=True), expected_data) filtered_by_filter_method = dn.filter([("bar", 1, Operator.EQUAL), ("bar", 2, Operator.EQUAL)], JoinOperator.OR) filtered_by_indexing = dn[(dn["bar"] == 1) | (dn["bar"] == 2)] expected_data = pd.DataFrame( [ {"foo": 1.0, "bar": 1.0}, {"foo": 1.0, "bar": 2.0}, {"foo": 2.0, "bar": 2.0}, {"bar": 2.0}, ] ) assert_frame_equal(filtered_by_filter_method.reset_index(drop=True), expected_data) assert_frame_equal(filtered_by_indexing.reset_index(drop=True), expected_data) def test_filter_numpy_exposed_type(self, csv_file): dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": csv_file, "exposed_type": "numpy"}) dn.write( [ [1, 1], [1, 2], [1, 3], [2, 1], [2, 2], [2, 3], ] ) # Test datanode indexing and slicing assert np.array_equal(dn[0], np.array([1, 1])) assert np.array_equal(dn[1], np.array([1, 2])) assert np.array_equal(dn[:3], np.array([[1, 1], [1, 2], [1, 3]])) assert np.array_equal(dn[:, 0], np.array([1, 1, 1, 2, 2, 2])) assert np.array_equal(dn[1:4, :1], np.array([[1], [1], [2]])) # Test filter data assert np.array_equal(dn.filter((0, 1, Operator.EQUAL)), np.array([[1, 1], [1, 2], [1, 3]])) assert np.array_equal(dn[dn[:, 0] == 1], np.array([[1, 1], [1, 2], [1, 3]])) assert np.array_equal(dn.filter((0, 1, Operator.NOT_EQUAL)), np.array([[2, 1], [2, 2], [2, 3]])) assert np.array_equal(dn[dn[:, 0] != 1], np.array([[2, 1], [2, 2], [2, 3]])) assert np.array_equal(dn.filter((1, 2, Operator.EQUAL)), np.array([[1, 2], [2, 2]])) assert np.array_equal(dn[dn[:, 1] == 2], np.array([[1, 2], [2, 2]])) assert np.array_equal( dn.filter([(1, 1, Operator.EQUAL), (1, 2, Operator.EQUAL)], JoinOperator.OR), np.array([[1, 1], [1, 2], [2, 1], [2, 2]]), ) assert np.array_equal(dn[(dn[:, 1] == 1) | (dn[:, 1] == 2)], np.array([[1, 1], [1, 2], [2, 1], [2, 2]])) def test_raise_error_invalid_exposed_type(self): path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv") with pytest.raises(InvalidExposedType): CSVDataNode("foo", Scope.SCENARIO, properties={"path": path, "exposed_type": "foo"}) def test_get_system_modified_date_instead_of_last_edit_date(self, tmpdir_factory): temp_file_path = str(tmpdir_factory.mktemp("data").join("temp.csv")) pd.DataFrame([]).to_csv(temp_file_path) dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": temp_file_path, "exposed_type": "pandas"}) dn.write(pd.DataFrame([1, 2, 3])) previous_edit_date = dn.last_edit_date sleep(0.1) pd.DataFrame([4, 5, 6]).to_csv(temp_file_path) new_edit_date = datetime.fromtimestamp(os.path.getmtime(temp_file_path)) assert previous_edit_date < dn.last_edit_date assert new_edit_date == dn.last_edit_date sleep(0.1) dn.write(pd.DataFrame([7, 8, 9])) assert new_edit_date < dn.last_edit_date os.unlink(temp_file_path)