move from unitests to pytests

Author: ypriverol

fsspark/fs/fdataframe.py

@@ -1,10 +1,10 @@
 import logging
-from typing import Optional, Union, List, Set, Tuple
+from typing import List, Tuple
 
 import numpy
 import numpy as np
 import pandas as pd
-from pandas import DataFrame, Series
+from pandas import DataFrame
 from sklearn.preprocessing import MinMaxScaler, MaxAbsScaler, StandardScaler, RobustScaler, LabelEncoder
 
 logging.basicConfig(format="%(levelname)s (%(name)s %(lineno)s): %(message)s")

fsspark/tests/generate_big_tests.py

@@ -0,0 +1,51 @@
+import logging
+
+import pandas as pd
+import numpy as np
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+def test_generate_big_dataset():
+    # Parameters for the dataset
+    n_samples = 1200
+    n_features = 10_000
+    chunk_size = 100  # Adjust chunk size for memory efficiency
+
+    # Generate sample IDs and labels
+    sample_ids = np.arange(1, n_samples + 1)
+    labels = np.random.choice(['LV', 'RV', 'LA', 'RA'], size=n_samples)
+
+    # Parquet schema definition
+    schema = pa.schema([pa.field('sample_id', pa.int32()), pa.field('label', pa.string())] +
+                       [pa.field(f'feature{i}', pa.float32()) for i in range(1, n_features + 1)])
+
+    # Create an empty Parquet file
+    output_file = 'large_dataset_optimized_samples_{}_features_{}.parquet'.format(n_samples, n_features)
+    with pq.ParquetWriter(output_file, schema, compression='snappy') as writer:
+        # Process in chunks to reduce memory usage
+        for chunk_start in range(0, n_samples, chunk_size):
+            chunk_end = min(chunk_start + chunk_size, n_samples)
+
+            # Generate chunk of samples and labels
+            chunk_sample_ids = sample_ids[chunk_start:chunk_end]
+            chunk_labels = labels[chunk_start:chunk_end]
+
+            # Generate chunk of features
+            chunk_features = {f'feature{i}': np.random.rand(chunk_end - chunk_start) for i in range(1, n_features + 1)}
+
+            # Create DataFrame chunk
+            chunk_data = {
+                'sample_id': chunk_sample_ids,
+                'label': chunk_labels
+            }
+            chunk_data.update(chunk_features)
+
+            df_chunk = pd.DataFrame(chunk_data)
+
+            # Convert to PyArrow Table and write chunk to Parquet file
+            table_chunk = pa.Table.from_pandas(df_chunk, schema=schema)
+            writer.write_table(table_chunk)
+            logging.info(f'Processed samples {chunk_start + 1} to {chunk_end}')
+
+    print("Optimized Parquet file created successfully!")
+

fsspark/tests/test_fsdataframe.py

@@ -1,4 +1,3 @@
-import pytest
 import pandas as pd
 from fsspark.fs.fdataframe import FSDataFrame
 

fsspark/tests/test_univariate_methods.py

@@ -1,44 +1,27 @@
-import unittest
-
 import pandas as pd
 from fsspark.utils.datasets import get_tnbc_data_path
 from fsspark.fs.fdataframe import FSDataFrame
 
 from fsspark.fs.univariate import univariate_filter
 
-
-class UnivariateMethodsTest(unittest.TestCase):
+def test_univariate_filter_corr():
     """
-    Define testing methods for FSDataFrame class.
+    Test univariate_filter method with 'u_corr' method.
+    :return: None
     """
 
-    def setUp(self) -> None:
-        # import tsv as pandas DataFrame
-        self.df = pd.read_csv(get_tnbc_data_path(), sep='\t')
-
-        # create FSDataFrame instance
-        self.fsdf = FSDataFrame(df=self.df,
-                                sample_col='Sample',
-                                label_col='label')
-
-    def tearDown(self) -> None:
-        pass
+    # import tsv as pandas DataFrame
+    df = pd.read_csv(get_tnbc_data_path(), sep='\t')
 
-    def test_univariate_filter_corr(self):
-        """
-        Test univariate_filter method with 'u_corr' method.
-        :return: None
-        """
+    # create FSDataFrame instance
+    fs_df = FSDataFrame(df=df,sample_col='Sample',label_col='label')
 
-        fsdf = self.fsdf
-        fsdf_filtered = univariate_filter(fsdf,
-                                          univariate_method='u_corr',
-                                          corr_threshold=0.3)
+    fsdf_filtered = univariate_filter(fs_df,univariate_method='u_corr', corr_threshold=0.3)
 
-        self.assertEqual(fsdf.count_features(), 500)
-        self.assertEqual(fsdf_filtered.count_features(), 211)
+    assert fs_df.count_features() == 500
+    assert fsdf_filtered.count_features() == 211
 
-        # Export the filtered DataFrame as Pandas DataFrame
-        df_filtered = fsdf_filtered.to_pandas()
-        df_filtered.to_csv('filtered_tnbc_data.csv', index=False)
+    # Export the filtered DataFrame as Pandas DataFrame
+    df_filtered = fsdf_filtered.to_pandas()
+    df_filtered.to_csv('filtered_tnbc_data.csv', index=False)