Recover and restore preprocessing improvements from d54de4e

Restored all preprocessing work that was done today:
- Add data_preprocess.py: Helper functions for data preprocessing
- Improve extract.py: Better data extraction logic (108 lines modified)
- Enhance kit_composition_cleaner.py: Major improvements (260 lines, 361 lines modified)
- Update transform.py: Better transformation logic (36 lines modified)
- Add paths.yaml: Configuration for data paths

Total: 505 lines changed across preprocessing module
This represents a full day's work on improving the preprocessing pipeline.

src/config/paths.yaml

@@ -0,0 +1,19 @@
+# Data Paths Configuration
+# All paths are relative to the project root directory
+
+data:
+  # CSV data files
+  csv:
+    demand: "data/real_data_excel/converted_csv/COOIS_Planned_and_Released.csv"
+    kit_composition: "data/real_data_excel/converted_csv/Kit_Composition_and_relation_cleaned_with_line_type.csv"
+    workforce_pay_scale: "data/real_data_excel/converted_csv/WH_Workforce_Hourly_Pay_Scale_processed.csv"
+    work_shift: "data/real_data_excel/converted_csv/work_shift.csv"
+    work_center_capacity: "data/real_data_excel/converted_csv/Work_Centre_Capacity.csv"
+    work_center_capacity_processed: "data/real_data_excel/converted_csv/Work_Centre_Capacity_processed.csv"
+    material_master: "data/real_data_excel/converted_csv/Material_Master_WMS.csv"
+    kits_calculation: "data/real_data_excel/converted_csv/Kits__Calculation.csv"
+  
+  # Hierarchy data
+  hierarchy:
+    kit_hierarchy: "data/hierarchy_exports/kit_hierarchy.json"
+

src/preprocess/data_preprocess.py

@@ -0,0 +1,39 @@
+import pandas as pd
+
+
+def process_Kit_Composition_and_relation(output_csv_path: str = 'data/real_data_excel/converted_csv/Kit_Composition_and_relation_cleaned_with_line_type_and_id.csv') -> pd.DataFrame:
+    """
+    Process the Kit_Composition_and_relation.csv file to clean the data and add line type and id.
+    
+    Returns:
+        saves to csv path
+        cleaned_df: pd.DataFrame
+    """
+    df = pd.read_csv('data/real_data_excel/converted_csv/Kit_Composition_and_relation.csv')
+    # df.dropna(inplace=True)
+    master = df[["Master Kit", "Master Kit  Description"]]
+    master["kit_type"] = "master"
+    master.rename(columns={"Master Kit": "kit_name", "Master Kit  Description": "kit_description"}, inplace=True)
+
+    subkit = df[["Sub kit", "Sub kit description"]]
+    subkit["kit_type"] = "subkit"
+    subkit.rename(columns={"Sub kit": "kit_name", "Sub kit Description": "kit_description"}, inplace=True)
+    subkit.columns = ["kit_name", "kit_description", "kit_type"]
+
+    prepack = df[["Prepack", "Prepack Description"]]
+    prepack["kit_type"] = "prepack"
+    prepack.rename(columns={"Prepack": "kit_name", "Prepack Description": "kit_description"}, inplace=True)
+
+
+    cleaned_df = pd.concat([master, subkit, prepack])
+    cleaned_df[['kit_name','kit_description','kit_type']].drop_duplicates()
+    tmp = cleaned_df.groupby('kit_name').count()['kit_type'].reset_index()
+    standalone_masterkit_list = tmp.loc[tmp['kit_type']==1,'kit_name']
+
+    cleaned_df.loc[cleaned_df['kit_name'].isin(standalone_masterkit_list),'line_type'] = 'long line'
+    cleaned_df.loc[cleaned_df['kit_type']=='prepack','line_type'] = 'mini load'
+    cleaned_df.loc[cleaned_df['kit_type']=='subkit','line_type'] = 'long line'
+    cleaned_df.loc[cleaned_df['line_type']=='mini load', 'line_id'] = 7
+    cleaned_df.loc[cleaned_df['line_type']=='long line', 'line_id'] = 6
+    cleaned_df.to_csv(output_csv_path, index=False)
+    return cleaned_df

src/preprocess/extract.py

@@ -3,77 +3,54 @@ import datetime
 from datetime import date, timedelta
 import json
 import os
-# Default dates - will be overridden by optimization_config.py
-START_DATE = pd.Timestamp(2025, 7, 7)
-END_DATE = pd.Timestamp(2025, 7, 11)
+import yaml
+from pathlib import Path
 
-def set_global_dates(start_date, end_date):
-    """Update global START_DATE and END_DATE variables"""
-    global START_DATE, END_DATE
-    START_DATE = pd.Timestamp(start_date)
-    END_DATE = pd.Timestamp(end_date)
-    print(f"Updated global dates: {START_DATE} to {END_DATE}")
+# Load paths configuration
+_config_dir = Path(__file__).parent.parent / "config"
+_paths_file = _config_dir / "paths.yaml"
+with open(_paths_file, 'r', encoding='utf-8') as f:
+    PATHS = yaml.safe_load(f)
 
 
-def read_excel(path: str) -> pd.DataFrame:
-    return pd.read_excel(path, dtype={"id": "Int64"})
-
-
-def read_demand_data(
-    path="data/real_data_excel/converted_csv/COOIS_Planned_and_Released.csv",
-    start_date=None,
-    end_date=None,
-) -> pd.DataFrame:
-    df = pd.read_csv(path)
-    df["Basic start date"] = pd.to_datetime(df["Basic start date"])
-    # df["Basic finish date"] = pd.to_datetime(df["Basic finish date"])
-    
-    # Use provided dates or fall back to module defaults
-    filter_start_date = start_date if start_date is not None else START_DATE
-    filter_end_date = end_date if end_date is not None else END_DATE
-    
-    df = df[(df["Basic start date"] == filter_start_date)]
-
-    return df
-
-def read_kit_line_match_data(
-    path="data/real_data_excel/converted_csv/Kit_Composition_and_relation_cleaned_with_line_type.csv",
-) -> pd.DataFrame:
+def read_kit_line_match_data() -> pd.DataFrame:
+    """Read kit composition and relation data"""
+    path = PATHS['data']['csv']['kit_composition']
     return pd.read_csv(path)
 
 
-def read_employee_data(
-    path="data/real_data_excel/converted_csv/WH_Workforce_Hourly_Pay_Scale_processed.csv",
-) -> pd.DataFrame:
+def read_employee_data() -> pd.DataFrame:
+    """Read employee workforce hourly pay scale data"""
+    path = PATHS['data']['csv']['workforce_pay_scale']
     return pd.read_csv(path)
 
-def get_shift_info(
-        path = "data/real_data_excel/converted_csv/work_shift.csv"
-) -> pd.DataFrame:
+def get_shift_info() -> pd.DataFrame:
+    """Read work shift information"""
+    path = PATHS['data']['csv']['work_shift']
     df = pd.read_csv(path)
     return df
 
 
-def read_shift_cost_data(
-    path="data/real_data_excel/converted_csv/WH_Workforce_Hourly_Pay_Scale_processed.csv",
-) -> pd.DataFrame:
+def read_shift_cost_data() -> pd.DataFrame:
+    """Read shift cost data from workforce pay scale"""
+    path = PATHS['data']['csv']['workforce_pay_scale']
     return pd.read_csv(path)
 
 
-def read_work_center_capacity(
-    path="data/real_data_excel/converted_csv/Work_Centre_Capacity.csv",
-) -> pd.DataFrame:
+def read_work_center_capacity() -> pd.DataFrame:
+    """Read work center capacity data"""
+    path = PATHS['data']['csv']['work_center_capacity']
     return pd.read_csv(path)
 
 
-def read_material_master(
-    path="data/real_data_excel/converted_csv/Material_Master_WMS.csv",
-) -> pd.DataFrame:
+def read_material_master() -> pd.DataFrame:
+    """Read material master WMS data"""
+    path = PATHS['data']['csv']['material_master']
     return pd.read_csv(path)
 
-def read_packaging_line_data(
-    path="data/real_data_excel/converted_csv/Work_Centre_Capacity_processed.csv",
-) -> pd.DataFrame:
+def read_packaging_line_data() -> pd.DataFrame:
+    """Read packaging line data (filtered work center capacity)"""
+    path = PATHS['data']['csv']['work_center_capacity_processed']
     df = pd.read_csv(path)
     # Filter for packaging lines only
     df = df[df["line_for_packaging"] == True]
@@ -81,26 +58,23 @@ def read_packaging_line_data(
 
 
 def read_orders_data(
-    path="data/real_data_excel/converted_csv/COOIS_Planned_and_Released.csv",
     start_date=None,
     # end_date=None,
 ) -> pd.DataFrame:
     """
-    COOIS_Released_Prod_Orders.csv 
+    Read COOIS Released Production Orders data
     
     Args:
-        path: path to the csv file
         start_date: start date (pd.Timestamp or datetime)
     
-
     Returns:
         pd.DataFrame: filtered dataframe by date
     """
+    path = PATHS['data']['csv']['demand']
     df = pd.read_csv(path)
     assert len(df) > 0, "No data found in the file"
     # convert date column to datetime
     df["Basic start date"] = pd.to_datetime(df["Basic start date"])
-    # df["Basic finish date"] = pd.to_datetime(df["Basic finish date"])
     
     
     # filter by date
@@ -112,9 +86,9 @@ def read_orders_data(
     return df
 
 
-def read_package_speed_data(
-    path="data/real_data_excel/converted_csv/Kits__Calculation.csv",
-):
+def read_package_speed_data():
+    """Read package speed data from Kits Calculation"""
+    path = PATHS['data']['csv']['kits_calculation']
     df = pd.read_csv(path, usecols=["Kit", "Kit per day","Paid work hours per day"])
     df["Kit per day"] = df["Kit per day"].astype(float)
     df["Paid work hours per day"] = df["Paid work hours per day"].astype(float)
@@ -123,9 +97,9 @@ def read_package_speed_data(
     speeds_per_hour = dict(zip(df["Kit"], df["kits_per_hour"]))
     return speeds_per_hour
 
-def read_personnel_requirement_data(
-    path="data/real_data_excel/converted_csv/Kits__Calculation.csv",
-):
+def read_personnel_requirement_data():
+    """Read personnel requirement data from Kits Calculation"""
+    path = PATHS['data']['csv']['kits_calculation']
     df = pd.read_csv(path, usecols=["Kit", "Humanizer", "UNICEF staff"])
     
     # Clean the data by handling special whitespace characters like \xa0 (non-breaking space)
@@ -156,12 +130,14 @@ def read_personnel_requirement_data(
 def get_production_order_data():
     """
     Extract production order information from hierarchy.
+    
     Returns:
-        - kit_levels: {kit_id: level} where level 0=prepack, 1=subkit, 2=master
-        - dependencies: {kit_id: [dependency_list]}
-        - priority_order: [kit_ids] sorted by production priority
+        tuple: (kit_levels, dependencies, priority_order)
+            - kit_levels: {kit_id: level} where level 0=prepack, 1=subkit, 2=master
+            - dependencies: {kit_id: [dependency_list]}
+            - priority_order: [kit_ids] sorted by production priority
     """
-    path = "data/hierarchy_exports/kit_hierarchy.json"
+    path = PATHS['data']['hierarchy']['kit_hierarchy']
     with open(path, 'r', encoding='utf-8') as f:
         hierarchy = json.load(f)
     

src/preprocess/kit_composition_cleaner.py

@@ -22,161 +22,206 @@ import os
 from typing import Tuple
 
 
-def load_kit_composition_data(file_path: str) -> pd.DataFrame:
-    """Load the Kit Composition and relation CSV file."""
-    if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
-    
-    df = pd.read_csv(file_path)
-    print(f"Loaded {len(df)} rows from {file_path}")
-    return df
-
-
-def process_master_kits(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    Process Master Kits according to business rules:
-    - Standalone masters (no subkits/prepacks, only components): line_type = "long line"
-    - Non-standalone masters (have subkits/prepacks): line_type = "" (empty - no production needed)
-    """
-    print("Processing Master Kits...")
-    
-    # Identify masters with hierarchy (subkits or prepacks)
-    masters_with_subkits = set(df[df['Sub kit'].notna()]['Master Kit'].unique())
-    masters_with_prepacks = set(df[df['Prepack'].notna()]['Master Kit'].unique())
-    masters_with_hierarchy = masters_with_subkits.union(masters_with_prepacks)
-    
-    # All masters
-    all_masters = set(df['Master Kit'].unique())
-    
-    # Standalone masters are those WITHOUT subkits/prepacks (only have components)
-    standalone_masters = all_masters - masters_with_hierarchy
-    
-    print(f"Total unique Master Kits: {len(all_masters)}")
-    print(f"Masters with subkits/prepacks: {len(masters_with_hierarchy)}")
-    print(f"Standalone masters (only components): {len(standalone_masters)}")
-    
-    # Create master kit records
-    master_data = []
-    
-    # Get unique master kits with descriptions
-    unique_masters = df[['Master Kit', 'Master Kit  Description']].drop_duplicates()
-    
-    for _, row in unique_masters.iterrows():
-        master_kit = row['Master Kit']
-        master_desc = row['Master Kit  Description']
-        
-        # Determine line_type based on standalone status
-        if master_kit in standalone_masters:
-            line_type = "long line"
-        else:
-            line_type = ""  # Empty for non-standalone (theoretical)
-        
-        master_data.append({
-            'kit_name': master_kit,
-            'kit_description': master_desc,
-            'kit_type': 'master',
-            'line_type': line_type
-        })
-    
-    master_df = pd.DataFrame(master_data)
-    print(f"Created {len(master_df)} master kit records")
-    print(f"Standalone masters with 'long line': {sum(master_df['line_type'] == 'long line')}")
-    
-    return master_df
-
-
-def process_sub_kits(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    Process Sub Kits according to business rules:
-    - All sub kits get line_type = "long line"
-    - Remove duplicates
-    """
-    print("Processing Sub Kits...")
-    
-    # Filter rows that have sub kits
-    subkit_df = df[df['Sub kit'].notna()].copy()
-    
-    if len(subkit_df) == 0:
-        print("No sub kits found")
-        return pd.DataFrame(columns=['kit_name', 'kit_description', 'kit_type', 'line_type'])
-    
-    # Get unique sub kits with descriptions
-    unique_subkits = subkit_df[['Sub kit', 'Sub kit description']].drop_duplicates()
-    
-    subkit_data = []
-    for _, row in unique_subkits.iterrows():
-        subkit_data.append({
-            'kit_name': row['Sub kit'],
-            'kit_description': row['Sub kit description'],
-            'kit_type': 'subkit',
-            'line_type': 'long line'
-        })
-    
-    subkit_result = pd.DataFrame(subkit_data)
-    print(f"Created {len(subkit_result)} sub kit records")
-    
-    return subkit_result
-
-
-def process_prepacks(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    Process Prepacks according to business rules:
-    - All prepacks get line_type = "miniload"
-    - Remove duplicates
+class KitCompositionCleaner:
     """
-    print("Processing Prepacks...")
-    
-    # Filter rows that have prepacks
-    prepack_df = df[df['Prepack'].notna()].copy()
-    
-    if len(prepack_df) == 0:
-        print("No prepacks found")
-        return pd.DataFrame(columns=['kit_name', 'kit_description', 'kit_type', 'line_type'])
+    Cleans and processes kit composition data with line type assignments.
     
-    # Get unique prepacks with descriptions
-    unique_prepacks = prepack_df[['Prepack', 'Prepack Description']].drop_duplicates()
-    
-    prepack_data = []
-    for _, row in unique_prepacks.iterrows():
-        prepack_data.append({
-            'kit_name': row['Prepack'],
-            'kit_description': row['Prepack Description'],
-            'kit_type': 'prepack',
-            'line_type': 'miniload'
-        })
-    
-    prepack_result = pd.DataFrame(prepack_data)
-    print(f"Created {len(prepack_result)} prepack records")
-    
-    return prepack_result
-
-
-def concatenate_and_save(master_df: pd.DataFrame, subkit_df: pd.DataFrame, 
-                        prepack_df: pd.DataFrame, output_path: str) -> pd.DataFrame:
+    This class maintains state across processing steps, allowing for:
+    - Single data load
+    - Step-by-step processing
+    - Intermediate result storage
     """
-    Concatenate all processed dataframes and save to output file.
-    """
-    print("Concatenating results...")
-    
-    # Concatenate all dataframes
-    final_df = pd.concat([master_df, subkit_df, prepack_df], ignore_index=True)
     
-    # Ensure empty strings instead of NaN for line_type
-    final_df['line_type'] = final_df['line_type'].fillna('')
+    def __init__(self, input_file: str, output_file: str = None):
+        """
+        Initialize the cleaner with file paths.
+        
+        Args:
+            input_file: Path to input CSV file (Kit_Composition_and_relation.csv)
+            output_file: Path to output CSV file (optional, can be set later)
+        """
+        self.input_file = input_file
+        self.output_file = output_file
+        
+        # State variables for processing pipeline
+        self.df = None
+        self.master_df = None
+        self.subkit_df = None
+        self.prepack_df = None
+        self.final_df = None
+        
+    def load_data(self) -> pd.DataFrame:
+        """Load the Kit Composition and relation CSV file."""
+        if not os.path.exists(self.input_file):
+            raise FileNotFoundError(f"File not found: {self.input_file}")
+        
+        self.df = pd.read_csv(self.input_file)
+        print(f"Loaded {len(self.df)} rows from {self.input_file}")
+        return self.df
     
-    # Sort by kit_type for better organization
-    final_df = final_df.sort_values(['kit_type', 'kit_name']).reset_index(drop=True)
+    def process_master_kits(self) -> pd.DataFrame:
+        """
+        Process Master Kits according to business rules:
+        - Standalone masters (no subkits/prepacks, only components): line_type = "long line"
+        - Non-standalone masters (have subkits/prepacks): line_type = "" (empty - no production needed)
+        """
+        if self.df is None:
+            raise ValueError("Data not loaded. Call load_data() first.")
+        
+        print("Processing Master Kits...")
+        
+        # Identify masters with hierarchy (subkits or prepacks)
+        masters_with_subkits = set(self.df[self.df['Sub kit'].notna()]['Master Kit'].unique())
+        masters_with_prepacks = set(self.df[self.df['Prepack'].notna()]['Master Kit'].unique())
+        masters_with_hierarchy = masters_with_subkits.union(masters_with_prepacks)
+        
+        # All masters
+        all_masters = set(self.df['Master Kit'].unique())
+        
+        # Standalone masters are those WITHOUT subkits/prepacks (only have components)
+        standalone_masters = all_masters - masters_with_hierarchy
+        
+        print(f"Total unique Master Kits: {len(all_masters)}")
+        print(f"Masters with subkits/prepacks: {len(masters_with_hierarchy)}")
+        print(f"Standalone masters (only components): {len(standalone_masters)}")
+        
+        # Create master kit records
+        master_data = []
+        
+        # Get unique master kits with descriptions
+        unique_masters = self.df[['Master Kit', 'Master Kit  Description']].drop_duplicates()
+        
+        for _, row in unique_masters.iterrows():
+            master_kit = row['Master Kit']
+            master_desc = row['Master Kit  Description']
+            
+            # Determine line_type based on standalone status
+            if master_kit in standalone_masters:
+                line_type = "long line"
+            else:
+                line_type = ""  # Empty for non-standalone (theoretical)
+            
+            master_data.append({
+                'kit_name': master_kit,
+                'kit_description': master_desc,
+                'kit_type': 'master',
+                'line_type': line_type
+            })
+        
+        self.master_df = pd.DataFrame(master_data)
+        print(f"Created {len(self.master_df)} master kit records")
+        print(f"Standalone masters with 'long line': {sum(self.master_df['line_type'] == 'long line')}")
+        
+        return self.master_df
     
-    print(f"Final dataset contains {len(final_df)} records:")
-    print(f"  - Masters: {len(master_df)}")
-    print(f"  - Subkits: {len(subkit_df)}")
-    print(f"  - Prepacks: {len(prepack_df)}")
+    def process_sub_kits(self) -> pd.DataFrame:
+        """
+        Process Sub Kits according to business rules:
+        - All sub kits get line_type = "long line"
+        - Remove duplicates
+        """
+        if self.df is None:
+            raise ValueError("Data not loaded. Call load_data() first.")
+        
+        print("Processing Sub Kits...")
+        
+        # Filter rows that have sub kits
+        subkit_df = self.df[self.df['Sub kit'].notna()].copy()
+        
+        if len(subkit_df) == 0:
+            print("No sub kits found")
+            self.subkit_df = pd.DataFrame(columns=['kit_name', 'kit_description', 'kit_type', 'line_type'])
+            return self.subkit_df
+        
+        # Get unique sub kits with descriptions
+        unique_subkits = subkit_df[['Sub kit', 'Sub kit description']].drop_duplicates()
+        
+        subkit_data = []
+        for _, row in unique_subkits.iterrows():
+            subkit_data.append({
+                'kit_name': row['Sub kit'],
+                'kit_description': row['Sub kit description'],
+                'kit_type': 'subkit',
+                'line_type': 'long line'
+            })
+        
+        self.subkit_df = pd.DataFrame(subkit_data)
+        print(f"Created {len(self.subkit_df)} sub kit records")
+        
+        return self.subkit_df
     
-    # Save to file (keep empty strings as empty, not NaN)
-    final_df.to_csv(output_path, index=False, na_rep='')
-    print(f"Saved cleaned data to: {output_path}")
+    def process_prepacks(self) -> pd.DataFrame:
+        """
+        Process Prepacks according to business rules:
+        - All prepacks get line_type = "miniload"
+        - Remove duplicates
+        """
+        if self.df is None:
+            raise ValueError("Data not loaded. Call load_data() first.")
+        
+        print("Processing Prepacks...")
+        
+        # Filter rows that have prepacks
+        prepack_df = self.df[self.df['Prepack'].notna()].copy()
+        
+        if len(prepack_df) == 0:
+            print("No prepacks found")
+            self.prepack_df = pd.DataFrame(columns=['kit_name', 'kit_description', 'kit_type', 'line_type'])
+            return self.prepack_df
+        
+        # Get unique prepacks with descriptions
+        unique_prepacks = prepack_df[['Prepack', 'Prepack Description']].drop_duplicates()
+        
+        prepack_data = []
+        for _, row in unique_prepacks.iterrows():
+            prepack_data.append({
+                'kit_name': row['Prepack'],
+                'kit_description': row['Prepack Description'],
+                'kit_type': 'prepack',
+                'line_type': 'miniload'
+            })
+        
+        self.prepack_df = pd.DataFrame(prepack_data)
+        print(f"Created {len(self.prepack_df)} prepack records")
+        
+        return self.prepack_df
     
-    return final_df
+    def concatenate_and_save(self, output_path: str = None) -> pd.DataFrame:
+        """
+        Concatenate all processed dataframes and save to output file.
+        
+        Args:
+            output_path: Path to save the output file (uses self.output_file if not provided)
+        """
+        if self.master_df is None or self.subkit_df is None or self.prepack_df is None:
+            raise ValueError("Processing not complete. Run process_master_kits(), process_sub_kits(), and process_prepacks() first.")
+        
+        print("Concatenating results...")
+        
+        # Concatenate all dataframes
+        self.final_df = pd.concat([self.master_df, self.subkit_df, self.prepack_df], ignore_index=True)
+        
+        # Ensure empty strings instead of NaN for line_type
+        self.final_df['line_type'] = self.final_df['line_type'].fillna('')
+        
+        # Sort by kit_type for better organization
+        self.final_df = self.final_df.sort_values(['kit_type', 'kit_name']).reset_index(drop=True)
+        
+        print(f"Final dataset contains {len(self.final_df)} records:")
+        print(f"  - Masters: {len(self.master_df)}")
+        print(f"  - Subkits: {len(self.subkit_df)}")
+        print(f"  - Prepacks: {len(self.prepack_df)}")
+        
+        # Determine output path
+        save_path = output_path or self.output_file
+        if save_path is None:
+            raise ValueError("No output path provided. Specify output_path parameter or set self.output_file")
+        
+        # Save to file (keep empty strings as empty, not NaN)
+        self.final_df.to_csv(save_path, index=False, na_rep='')
+        print(f"Saved cleaned data to: {save_path}")
+        
+        return self.final_df
 
 
 def main():
@@ -187,19 +232,17 @@ def main():
     output_file = os.path.join(base_dir, "data/real_data_excel/converted_csv/Kit_Composition_and_relation_cleaned_with_line_type.csv")
     
     try:
-        # Load the original data
-        df = load_kit_composition_data(input_file)
-        
-        # Process each type of kit
-        master_df = process_master_kits(df)
-        subkit_df = process_sub_kits(df)
-        prepack_df = process_prepacks(df)
+        # Initialize cleaner with class
+        cleaner = KitCompositionCleaner(input_file, output_file)
         
-        # Concatenate and save
-        final_df = concatenate_and_save(master_df, subkit_df, prepack_df, output_file)
+        # Execute pipeline step by step
+        cleaner.load_data()
+        cleaner.process_master_kits()
+        cleaner.process_sub_kits()
+        cleaner.process_prepacks()
+        final_df = cleaner.concatenate_and_save()
         
         # Display summary statistics
-        print("\n=== SUMMARY ===")
         print("Line type distribution:")
         print(final_df['line_type'].value_counts(dropna=False))
         print("\nKit type distribution:")
@@ -207,9 +250,9 @@ def main():
         
         print("\nSample of final data:")
         print(final_df.head(10))
-        
+
     except Exception as e:
-        print(f"Error processing kit composition data: {e}")
+        print(f"❌ Error processing kit composition data: {e}")
         raise
 
 

src/preprocess/transform.py

@@ -2,8 +2,14 @@ import pandas as pd
 import src.preprocess.extract as ex
 
 
-def get_product_list():
-    demand = ex.read_demand_data()
+def get_product_list(start_date=None):
+    """
+    Get unique product list from demand data
+    
+    Args:
+        start_date: start date to filter data. Required.
+    """
+    demand = ex.read_orders_data(start_date=start_date)
     print(demand["Material Number"].unique())
     return demand["Material Number"].unique()
 
@@ -16,13 +22,15 @@ def get_employee_list():
 
 
 def get_released_product_list(start_date=None):
+    """
+    get released product list from COOIS_Released_Prod_Orders.csv
     
+    Args:
+        start_date: start date to filter data. Required.
+    """
     released_orders = ex.read_orders_data(
         start_date=start_date, 
-        # end_date=end_date
     )
-    
-    
     product_list = released_orders["Material Number"].unique().tolist()
     print(f"Released products for date range {start_date}: {len(product_list)} products")
     return product_list
@@ -30,23 +38,19 @@ def get_released_product_list(start_date=None):
 
 def get_available_dates():
     """
-    COOIS_Released_Prod_Orders.csv에서 사용 가능한 모든 날짜를 가져옵니다.
+    get available all dates from COOIS_Released_Prod_Orders.csv
     
     Returns:
-        tuple: (start_dates, end_dates) - 고유한 시작 날짜와 종료 날짜 리스트
+        tuple: (start_dates, end_dates) - unique start dates and end dates list
     """
-    # 모든 데이터를 가져옴 (날짜 필터링 없이)
     released_orders = ex.read_orders_data()
     
-    # 날짜 컬럼을 datetime으로 변환 (이미 extract.py에서 변환되지만 확실히 하기 위해)
     released_orders["Basic start date"] = pd.to_datetime(released_orders["Basic start date"])
     released_orders["Basic finish date"] = pd.to_datetime(released_orders["Basic finish date"])
     
-    # 고유한 시작 날짜와 종료 날짜 추출
     start_dates = sorted(released_orders["Basic start date"].dt.date.unique())
     end_dates = sorted(released_orders["Basic finish date"].dt.date.unique())
     
-    # 모든 고유한 날짜들 (시작날짜 + 종료날짜)
     all_dates = sorted(set(start_dates + end_dates))
     
     return all_dates, start_dates, end_dates
@@ -54,25 +58,21 @@ def get_available_dates():
 
 def get_date_ranges():
     """
-    COOIS_Released_Prod_Orders.csv에서 날짜 범위 조합을 가져옵니다.
-    
+    get available (start_date, end_date) combinations
     Returns:
-        list: 사용 가능한 (start_date, end_date) 조합 리스트
+        list : available (start_date, end_date) combinations
     """
     released_orders = ex.read_orders_data()
     
-    # 날짜 컬럼을 datetime으로 변환
     released_orders["Basic start date"] = pd.to_datetime(released_orders["Basic start date"])
     released_orders["Basic finish date"] = pd.to_datetime(released_orders["Basic finish date"])
     
-    # 고유한 날짜 범위 조합 추출
     date_ranges = released_orders[["Basic start date", "Basic finish date"]].drop_duplicates()
     date_ranges["start_date"] = date_ranges["Basic start date"].dt.date
     date_ranges["end_date"] = date_ranges["Basic finish date"].dt.date
     
-    # (start_date, end_date) 튜플 리스트로 반환
     ranges = [(row["start_date"], row["end_date"]) for _, row in date_ranges.iterrows()]
-    ranges = sorted(set(ranges))  # 중복 제거 및 정렬
+    ranges = sorted(set(ranges))
     
     return ranges