automatic file detection

Author: Cyprien Ricque

pyproject.toml

@@ -7,7 +7,7 @@ py-modules = ["stdflow"]
 
 [project]
 name = "stdflow"
-version = "0.0.20"
+version = "0.0.21"
 description = "[alpha] A package that transform your notebooks and python files into pipeline steps by standardizing the data input / output."
 readme = "README.md"
 authors = [

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name="stdflow",
-    version="0.0.20",
+    version="0.0.21",
     description="[alpha] A package that transform your notebooks and python files into pipeline steps by standardizing the data input / output.",
     long_description=open("README.md").read(),
     long_description_content_type="text/markdown",

stdflow/__init__.py

@@ -10,7 +10,7 @@
 
 import pandas as pd
 
-__version__ = "0.0.20"
+__version__ = "0.0.21"
 
 import logging
 import sys

stdflow/stdflow_path/data_path.py

@@ -4,6 +4,7 @@
 import os
 
 from stdflow.stdflow_path import Path
+from stdflow_utils.listing import list_csv_files
 
 try:
     from typing import Literal, Optional
@@ -26,7 +27,7 @@ def __init__(
         attrs: list | None | str = None,
         step_name: str | None = None,
         version: str | Literal[":last", ":first"] = ":last",
-        file_name: str = None,
+        file_name: str | Literal[":auto"] = None,
     ):
         """
         At this stage all information are present except the version which is to be detected if not specified
@@ -36,8 +37,9 @@ def __init__(
         :param version: last part of the full_path. one of [":last", ":first", "<version_name>", None]
         :param file_name: file name (optional)
         """
-        # if step is str and contains step_, remove it
         super().__init__(root, file_name)
+
+        # if step is str and contains step_, remove it
         if isinstance(step_name, str) and step_name.startswith(STEP_PREFIX):
             step_name = step_name[len(STEP_PREFIX) :]
         # if version is str and contains v_, remove it
@@ -55,6 +57,20 @@ def __init__(
         elif version is not None:
             self.version = version
 
+        if file_name == ":auto":
+            self.file_name = self.detect_file_name()
+
+    def detect_file_name(self):
+        if not os.path.isdir(self.dir_path):
+            logger.error(f"Path {self.dir_path} does not exist")
+        files = list_csv_files(self.dir_path)
+        if len(files) == 1:
+            logger.debug(f"Using file {files[0]}")
+            return files[0]
+        else:
+            logger.warning(f"Multiple files found in {self.dir_path}: {files}")
+            return None
+
     def detect_version(self, path, version_type):
         if version_type not in [":last", ":first"]:
             logger.warning(f"Unknown version type: {version_type}")

stdflow/stdflow_utils/__init__.py

@@ -144,6 +144,9 @@ def string_to_uuid(input_string):
 #         )
 
 
+
+
+
 if __name__ == "__main__":
     to_html(
         dest="./",

stdflow/stdflow_utils/listing.py

@@ -0,0 +1,56 @@
+import os
+import re
+
+
+def list_files_regex_all_depth(directory, pattern):
+    matched_files = []
+
+    # Walk through each file in the directory
+    for root, dirs, files in os.walk(directory):
+        for file in files:
+            # If the file matches the pattern, add it to the list
+            if re.search(pattern, file):
+                full_path = os.path.join(root, file)
+                relative_path = os.path.relpath(full_path, directory)
+                matched_files.append(relative_path)
+
+    return matched_files
+
+
+def list_files_regex(directory, pattern):
+    matched_files = []
+
+    # List each file in the directory
+    for file in os.listdir(directory):
+        # Construct the full path
+        full_path = os.path.join(directory, file)
+        # Check if it's a file (not a subdirectory)
+        if os.path.isfile(full_path):
+            # If the file matches the pattern, add it to the list
+            if re.search(pattern, file):
+                relative_path = os.path.relpath(full_path, directory)
+                matched_files.append(relative_path)
+
+    return matched_files
+
+
+# using glob
+def list_files_ext(directory, pattern):
+    import glob
+
+    paths = glob.glob(os.path.join(directory, pattern))
+    return [os.path.relpath(path, directory) for path in paths]
+
+
+def list_excel_files(directory):
+    return list_files_ext(directory, "*.xlsx") + list_files_ext(directory, "*.xls")
+
+
+def list_csv_files(directory):
+    return list_files_ext(directory, "*.csv")
+
+
+if __name__ == "__main__":
+    print(list_files_regex_all_depth(directory="./src", pattern=".*\.py$"))
+    print(list_files_regex(directory="./src/data_analysis", pattern=".*\.py"))
+    print(list_files_ext(directory="./src/data_analysis", pattern="*.py"))

stdflow/step.py

@@ -73,31 +73,57 @@ def __new__(cls, *args, **kwargs):
 
 
 class Step(ModuleType):
-    def __init__(self, name: str = "Step"):
-        super().__init__(name)
+    def __init__(
+        self,
+        step_in: str | None = None,
+        version_in: str | None = ":last",
+        attrs_in: str | list[str] | None = ":default",
+        file_name_in: str | None = ":default",
+        method_in: str | object | None = ":auto",
+        root_in: str | None = ":default",
+
+        step_out: str | None = None,
+        version_out: str | None = DEFAULT_DATE_VERSION_FORMAT,
+        attrs_out: str | list[str] | None = ":default",
+        file_name_out: str | None = ":default",
+        method_out: str | object | None = ":auto",
+        root_out: str | None = ":default",
+
+        root: str | None = "./data",
+        attrs: str | list[str] | None = None,
+        file_name: str | None = ":auto",
+
+        data_l: list[MetaData] = None,
+        data_l_in: list[MetaData] = None,
+
+    ):
+        super().__init__("Step")
+
         # === Exported === #
-        self.data_l: list[MetaData] = []
-        self.data_l_in: list[MetaData] = []  # direct input to this step file
+        # all inputs to this step
+        self.data_l = data_l if data_l is not None else []
+        # direct input to this step
+        self.data_l_in = data_l_in if data_l_in is not None else []
         # ================ #
 
         # Default values of load and save functions
-        self._step_in: str | None = None
-        self._version_in: str | None = ":last"
-        self._attrs_in: str | list[str] | None = ":default"
-        self._file_name_in: str | None = ":default"  # TODO
-        self._method_in: str | object | None = ":auto"  # TODO
-        self._root_in: str | None = ":default"
-
-        self._step_out: str | None = None
-        self._version_out: str | None = DEFAULT_DATE_VERSION_FORMAT
-        self._attrs_out: str | list[str] | None = ":default"
-        self._file_name_out: str | None = ":default"  # TODO
-        self._method_out: str | object | None = ":auto"
-        self._root_out: str | None = ":default"
+        self._step_in = step_in
+        self._version_in = version_in
+        self._attrs_in = attrs_in
+        self._file_name_in = file_name_in
+        self._method_in = method_in
+        self._root_in = root_in
+
+        self._step_out = step_out
+        self._version_out = version_out
+        self._attrs_out = attrs_out
+        self._file_name_out = file_name_out
+        self._method_out = method_out
+        self._root_out = root_out
 
-        self._root: str | None = "./data"
-        self._file_name: str | None = ":auto"  # TODO
-        self._attrs: str | list[str] | None = None
+        self._root = root
+        self._file_name = file_name
+        self._attrs = attrs
 
     def load(
         self,
@@ -125,6 +151,10 @@ def load(
         :param kwargs: kwargs to send to the method
         :return:
         """
+        if verbose:
+            logger.setLevel(logging.INFO)
+        else:
+            logger.setLevel(logging.WARNING)
         caller_file_name, caller_function, caller_package = get_caller_metadata()
         if "ipykernel" in caller_file_name:
             notebook_path, notebook_name = get_notebook_path()

tests/old/test_version.py

@@ -2,7 +2,7 @@
 
 
 def test_exact_version():
-    assert st.__version__ == "0.0.20"
+    assert st.__version__ == "0.0.21"
 
 
 def test_version():

tests/test_file_name_auto.py

@@ -0,0 +1,39 @@
+import os
+import re
+
+import stdflow as sf
+
+import pandas as pd
+import os
+import shutil
+import pytest
+
+
+import stdflow as sf
+
+
+def test_sf_load():
+    # Define a test directory name
+    test_dir = "test_dir"
+
+    # Create the test directory
+    os.makedirs(test_dir, exist_ok=True)
+
+    # Create a test dataframe
+    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
+
+    # Save the test dataframe to csv in the test directory
+    df.to_csv(os.path.join(test_dir, "test.csv"), index=False)
+
+    # Load the dataframe using sf.load
+    loaded_df = sf.load(root='./', attrs=test_dir)
+
+    # Check if the loaded dataframe is equal to the original one
+    pd.testing.assert_frame_equal(df, loaded_df)
+
+    # Remove the test directory after test
+    shutil.rmtree(test_dir)
+
+
+# Run the test
+pytest.main(["-v", "your_test_file.py"])

tests/test_multiple_steps.py

@@ -6,7 +6,7 @@
 
 def load_and_process_digi_sentiments_indonesia(catalog, brand_dict=None):
     print("load_and_process_digi_sentiments_indonesia")
-    conf = catalog["digimind_sentiments"]
+    conf = catalog["twitter_sentiments"]
 
     step = sf.Step()
     step.reset()
@@ -36,7 +36,7 @@ def load_digi_mentions_indonesia(step):
 
 def load_and_process_digi_mentions_indonesia(catalog, brand_dict=None):
     print("load_and_process_digi_mentions_indonesia")
-    conf = catalog["digimind_mentions"]
+    conf = catalog["twitter_mentions"]
 
     step = sf.Step()
     step.reset()
@@ -54,7 +54,7 @@ def load_and_process_digi_mentions_indonesia(catalog, brand_dict=None):
 
 def load_and_process_digi_indonesia(catalog, brand_dict=None):
     print("load_and_process_digi_indonesia")
-    conf = catalog["digimind"]
+    conf = catalog["twitter"]
 
     step = sf.Step()
     step.reset()
@@ -69,8 +69,8 @@ def load_and_process_digi_indonesia(catalog, brand_dict=None):
 def test_ms():
     load_and_process_digi_indonesia(
         {
-            "digimind": {"attrs_out": "oui", "step_out": "non"},
-            "digimind_mentions": {"attrs": "oui", "step_out": "non", "step_in": "non"},
-            "digimind_sentiments": {"attrs": "oui", "step_out": "non", "step_in": "non"},
+            "twitter": {"attrs_out": "oui", "step_out": "non"},
+            "twitter_mentions": {"attrs": "oui", "step_out": "non", "step_in": "non"},
+            "twitter_sentiments": {"attrs": "oui", "step_out": "non", "step_in": "non"},
         }
     )