Upload folder using huggingface_hub

ffa1f50 verified about 1 year ago

6.65 kB

import pandas as pd
import datetime
from datetime import date, timedelta
import json
import os
import yaml
from pathlib import Path

# 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_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() -> 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() -> pd.DataFrame:
    """Read work shift information"""
    path = PATHS['data']['csv']['work_shift']
    df = pd.read_csv(path)
    return df


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() -> pd.DataFrame:
    """Read work center capacity data"""
    path = PATHS['data']['csv']['work_center_capacity']
    return pd.read_csv(path)


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() -> 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]
    return df


def read_orders_data(
    start_date=None,
    # end_date=None,
) -> pd.DataFrame:
    """
    Read COOIS Released Production Orders data
    
    Args:
        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"])
    
    
    # filter by date
    if start_date is not None:    # Filter for exact start date only
        df = df[df["Basic start date"] == pd.to_datetime(start_date)]
    else:
        raise ValueError("start_date is required")
    
    return df


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)
    df["Kit"] = df["Kit"].astype(str)
    df['kits_per_hour'] = df['Kit per day']/df['Paid work hours per day']
    speeds_per_hour = dict(zip(df["Kit"], df["kits_per_hour"]))
    return speeds_per_hour

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)
    def clean_and_convert_to_float(value):
        if pd.isna(value):
            return 0.0
        
        # Convert to string and strip all kinds of whitespace (including \xa0)
        clean_value = str(value).strip()
        
        # If empty after stripping, return 0
        if clean_value == '' or clean_value == 'nan':
            return 0.0
        
        try:
            return float(clean_value)
        except ValueError as e:
            print(f"Warning: Could not convert '{repr(value)}' to float, setting to 0. Error: {e}")
            return 0.0
    
    df["Humanizer"] = df["Humanizer"].apply(clean_and_convert_to_float)
    df["UNICEF staff"] = df["UNICEF staff"].apply(clean_and_convert_to_float)
    df["Kit"] = df["Kit"].astype(str)
    
    return df


def get_production_order_data():
    """
    Extract production order information from hierarchy.
    
    Returns:
        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 = PATHS['data']['hierarchy']['kit_hierarchy']
    with open(path, 'r', encoding='utf-8') as f:
        hierarchy = json.load(f)
    
    kit_levels = {}
    dependencies = {}
    
    # Process hierarchy to extract levels and dependencies
    for master_id, master_data in hierarchy.items():
        # Master kits are level 2
        kit_levels[master_id] = 2
        dependencies[master_id] = master_data.get('dependencies', [])
        
        # Process subkits (level 1)
        for subkit_id, subkit_data in master_data.get('subkits', {}).items():
            kit_levels[subkit_id] = 1
            dependencies[subkit_id] = subkit_data.get('dependencies', [])
            
            # Process prepacks under subkits (level 0)
            for prepack_id in subkit_data.get('prepacks', []):
                if prepack_id not in kit_levels:  # Avoid overwriting if already exists
                    kit_levels[prepack_id] = 0
                    dependencies[prepack_id] = []
        
        # Process direct prepacks under master (level 0)
        for prepack_id in master_data.get('direct_prepacks', []):
            if prepack_id not in kit_levels:  # Avoid overwriting if already exists
                kit_levels[prepack_id] = 0
                dependencies[prepack_id] = []
    
    # Create priority order: prepacks first, then subkits, then masters
    priority_order = []
    
    # Level 0: Prepacks (highest priority)
    prepacks = [kit for kit, level in kit_levels.items() if level == 0]
    priority_order.extend(sorted(prepacks))
    
    # Level 1: Subkits (medium priority)
    subkits = [kit for kit, level in kit_levels.items() if level == 1]
    priority_order.extend(sorted(subkits))
    
    # Level 2: Masters (lowest priority)
    masters = [kit for kit, level in kit_levels.items() if level == 2]
    priority_order.extend(sorted(masters))
    
    return kit_levels, dependencies, priority_order



if __name__ == "__main__":
    employee_data = read_employee_data()
    print("employee data")
    print(employee_data)
    print("line speed data",read_package_speed_data())
    

	import pandas as pd
	import datetime
	from datetime import date, timedelta
	import json
	import os
	import yaml
	from pathlib import Path

	# 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_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() -> 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() -> pd.DataFrame:
	"""Read work shift information"""
	path = PATHS['data']['csv']['work_shift']
	df = pd.read_csv(path)
	return df


	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() -> pd.DataFrame:
	"""Read work center capacity data"""
	path = PATHS['data']['csv']['work_center_capacity']
	return pd.read_csv(path)


	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() -> 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]
	return df


	def read_orders_data(
	start_date=None,
	# end_date=None,
	) -> pd.DataFrame:
	"""
	Read COOIS Released Production Orders data

	Args:
	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"])


	# filter by date
	if start_date is not None: # Filter for exact start date only
	df = df[df["Basic start date"] == pd.to_datetime(start_date)]
	else:
	raise ValueError("start_date is required")

	return df


	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)
	df["Kit"] = df["Kit"].astype(str)
	df['kits_per_hour'] = df['Kit per day']/df['Paid work hours per day']
	speeds_per_hour = dict(zip(df["Kit"], df["kits_per_hour"]))
	return speeds_per_hour

	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)
	def clean_and_convert_to_float(value):
	if pd.isna(value):
	return 0.0

	# Convert to string and strip all kinds of whitespace (including \xa0)
	clean_value = str(value).strip()

	# If empty after stripping, return 0
	if clean_value == '' or clean_value == 'nan':
	return 0.0

	try:
	return float(clean_value)
	except ValueError as e:
	print(f"Warning: Could not convert '{repr(value)}' to float, setting to 0. Error: {e}")
	return 0.0

	df["Humanizer"] = df["Humanizer"].apply(clean_and_convert_to_float)
	df["UNICEF staff"] = df["UNICEF staff"].apply(clean_and_convert_to_float)
	df["Kit"] = df["Kit"].astype(str)

	return df


	def get_production_order_data():
	"""
	Extract production order information from hierarchy.

	Returns:
	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 = PATHS['data']['hierarchy']['kit_hierarchy']
	with open(path, 'r', encoding='utf-8') as f:
	hierarchy = json.load(f)

	kit_levels = {}
	dependencies = {}

	# Process hierarchy to extract levels and dependencies
	for master_id, master_data in hierarchy.items():
	# Master kits are level 2
	kit_levels[master_id] = 2
	dependencies[master_id] = master_data.get('dependencies', [])

	# Process subkits (level 1)
	for subkit_id, subkit_data in master_data.get('subkits', {}).items():
	kit_levels[subkit_id] = 1
	dependencies[subkit_id] = subkit_data.get('dependencies', [])

	# Process prepacks under subkits (level 0)
	for prepack_id in subkit_data.get('prepacks', []):
	if prepack_id not in kit_levels: # Avoid overwriting if already exists
	kit_levels[prepack_id] = 0
	dependencies[prepack_id] = []

	# Process direct prepacks under master (level 0)
	for prepack_id in master_data.get('direct_prepacks', []):
	if prepack_id not in kit_levels: # Avoid overwriting if already exists
	kit_levels[prepack_id] = 0
	dependencies[prepack_id] = []

	# Create priority order: prepacks first, then subkits, then masters
	priority_order = []

	# Level 0: Prepacks (highest priority)
	prepacks = [kit for kit, level in kit_levels.items() if level == 0]
	priority_order.extend(sorted(prepacks))

	# Level 1: Subkits (medium priority)
	subkits = [kit for kit, level in kit_levels.items() if level == 1]
	priority_order.extend(sorted(subkits))

	# Level 2: Masters (lowest priority)
	masters = [kit for kit, level in kit_levels.items() if level == 2]
	priority_order.extend(sorted(masters))

	return kit_levels, dependencies, priority_order



	if __name__ == "__main__":
	employee_data = read_employee_data()
	print("employee data")
	print(employee_data)
	print("line speed data",read_package_speed_data())