Spaces:
No application file
No application file
import os | |
from processor import ManufacturingProcessor, get_file_preview | |
from utils import prompt_weights | |
def main(): | |
print("π Manufacturing Priority Decision Helper") | |
# Get file path | |
file_path = input("Enter the full path to your Excel file: ").strip() | |
if not os.path.exists(file_path): | |
print("β File not found.") | |
return | |
# Initialize processor | |
try: | |
processor = ManufacturingProcessor() | |
file_info = processor.get_file_info(file_path) | |
except Exception as e: | |
print(f"β Unable to read Excel file: {e}") | |
return | |
# Show available sheets | |
print(f"\nπ Available sheets in '{file_info['file_name']}':") | |
for i, sheet_name in enumerate(file_info['sheets'], start=1): | |
print(f" {i}. {sheet_name}") | |
# Sheet selection | |
while True: | |
try: | |
idx = int(input("Select a sheet by number: ")) | |
if 1 <= idx <= len(file_info['sheets']): | |
selected_sheet = file_info['sheets'][idx-1] | |
break | |
except ValueError: | |
pass | |
print("β οΈ Invalid selection, try again.") | |
print(f"β Selected sheet: {selected_sheet}") | |
# Preview data and validate | |
print("\nπ Analyzing data...") | |
try: | |
preview = get_file_preview(file_path, selected_sheet) | |
validation = preview['validation'] | |
print(f"π Data Summary:") | |
print(f" - Rows: {validation['row_count']}") | |
print(f" - Available columns: {len(validation['available_columns'])}") | |
if not validation['valid']: | |
print(f"\nβ Data validation failed:") | |
print(f" Missing required columns: {validation['missing_columns']}") | |
return | |
if validation['data_issues']: | |
print(f"\nβ οΈ Data quality issues found:") | |
for issue in validation['data_issues']: | |
print(f" - {issue}") | |
continue_anyway = input("\nContinue processing anyway? (y/N): ").strip().lower() | |
if continue_anyway != 'y': | |
return | |
print("β Data validation passed!") | |
except Exception as e: | |
print(f"β Error analyzing data: {e}") | |
return | |
# Weight adjustment (optional) | |
print(f"\nβοΈ Current weights: Age={processor.weights['AGE_WEIGHT']}%, " | |
f"Component={processor.weights['COMPONENT_WEIGHT']}%, " | |
f"Manual={processor.weights['MANUAL_WEIGHT']}%") | |
adjust_weights = input("Would you like to adjust weights? (y/N): ").strip().lower() | |
if adjust_weights == 'y': | |
try: | |
new_weights = prompt_weights(processor.weights.copy()) | |
processor.weights = new_weights | |
print(f"β Updated weights: {new_weights}") | |
except Exception as e: | |
print(f"β οΈ Error setting weights, using defaults: {e}") | |
# Quantity threshold | |
try: | |
min_qty_input = input("Enter minimum quantity threshold for FIFO (default 50): ").strip() | |
min_qty = int(min_qty_input) if min_qty_input else 50 | |
except ValueError: | |
min_qty = 50 | |
print("β οΈ Invalid input, using default threshold of 50") | |
# Process the data | |
print(f"\nπ Processing data with minimum quantity threshold: {min_qty}") | |
try: | |
processed_df, processing_info = processor.process_file( | |
file_path, selected_sheet, min_qty | |
) | |
print("β Priority calculation completed!") | |
print(f"π Results summary:") | |
print(f" - Total products: {processing_info['total_products']}") | |
print(f" - Products above threshold: {processing_info['products_above_threshold']}") | |
print(f" - Highest priority score: {processing_info['highest_priority_score']:.4f}") | |
except Exception as e: | |
print(f"β Error processing data: {e}") | |
return | |
# Show preview of results | |
print(f"\nπ Top 10 Priority Results:") | |
display_cols = [c for c in ["Name of Product", "Components Used", "Quantity of Each Component", | |
"Oldest Product Required First", "Priority Assigned", | |
"OrderAgeDays", "ComponentCount", "QtyThresholdOK", "PriorityScore"] | |
if c in processed_df.columns] | |
print(processed_df[display_cols].head(10).to_string(index=False, max_colwidth=20)) | |
# Save results | |
base_name = os.path.splitext(os.path.basename(file_path))[0] | |
output_dir = os.path.dirname(file_path) | |
output_path = os.path.join(output_dir, f"{base_name}_PRIORITY.xlsx") | |
try: | |
final_output = processor.save_results(processed_df, output_path, processing_info) | |
print(f"\nπΎ Results saved to: {final_output}") | |
print(f"\nπ Output includes:") | |
print(f" - Priority_Results: Ranked manufacturing data") | |
print(f" - Instructions: Methodology and column explanations") | |
print(f" - Processing_Log: Detailed processing information") | |
except Exception as e: | |
print(f"β Failed to save results: {e}") | |
return | |
print(f"\nπ Processing complete! Check the output file for detailed results.") | |
if __name__ == "__main__": | |
main() |