neversight/learn-skills.dev/automotive-supply-chain

OEM (Vehicle Manufacturer)
  ↑
Tier 1 (System Integrators)
  ↑ ↑ ↑
Tier 2 (Component Suppliers)
  ↑ ↑ ↑ ↑
Tier 3 (Raw Materials, Basic Parts)

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class AutomotiveJITScheduler:
    """
    Manage Just-In-Time delivery schedules for automotive assembly
    """
    def __init__(self, assembly_schedule, takt_time_minutes):
        """
        Initialize JIT scheduler
        Parameters:
        - assembly_schedule: vehicle build schedule
        - takt_time_minutes: time per vehicle (e.g., 60 seconds = 1 vehicle/min)
        """
        self.assembly_schedule = assembly_schedule
        self.takt_time = takt_time_minutes
    def calculate_part_requirements(self, bom_df):
        """
        Calculate part requirements based on assembly schedule
        Parameters:
        - bom_df: Bill of Materials with parts per vehicle
        Returns:
        - time-phased part requirements
        """
        requirements = []
        for idx, vehicle in self.assembly_schedule.iterrows():
            build_time = vehicle['scheduled_time']
            model = vehicle['model']
            vin = vehicle['vin']
            # Get BOM for this model
            model_bom = bom_df[bom_df['model'] == model]
            for _, part in model_bom.iterrows():
                requirements.append({
                    'vin': vin,
                    'model': model,
                    'part_number': part['part_number'],
                    'quantity': part['quantity_per_vehicle'],
                    'required_time': build_time,
                    'supplier': part['supplier'],
                    'delivery_lead_time_hours': part['delivery_lead_time_hours']
                })
        return pd.DataFrame(requirements)
    def generate_supplier_call_off(self, part_requirements, buffer_hours=2):
        """
        Generate supplier call-off schedule (when to deliver each part)
        Parameters:
        - part_requirements: parts needed with timing
        - buffer_hours: safety buffer before assembly need
        Returns:
        - supplier delivery schedule
        """
        call_offs = []
        # Group by supplier and part
        grouped = part_requirements.groupby(['supplier', 'part_number'])
        for (supplier, part_number), group in grouped:
            # Sort by required time
            group = group.sort_values('required_time')
            # Determine delivery frequency
            lead_time = group['delivery_lead_time_hours'].iloc[0]
            # Calculate delivery windows
            for idx, row in group.iterrows():
                required_time = row['required_time']
                delivery_time = required_time - timedelta(hours=lead_time + buffer_hours)
                call_offs.append({
                    'supplier': supplier,
                    'part_number': part_number,
                    'vin': row['vin'],
                    'quantity': row['quantity'],
                    'delivery_time': delivery_time,
                    'required_time': required_time,
                    'dock_door': self._assign_dock_door(supplier)
                })
        call_off_df = pd.DataFrame(call_offs)
        return call_off_df.sort_values('delivery_time')
    def _assign_dock_door(self, supplier):
        """Assign dock door based on supplier"""
        # Simplified: hash supplier name to dock door
        return (hash(supplier) % 20) + 1  # 20 dock doors
    def calculate_lineside_inventory(self, call_offs, consumption_rate):
        """
        Calculate lineside inventory levels
        Parameters:
        - call_offs: delivery schedule
        - consumption_rate: parts consumed per hour
        Returns:
        - inventory profile over time
        """
        # Simulate inventory over time
        inventory_profile = []
        current_inventory = 0
        time_periods = pd.date_range(
            start=call_offs['delivery_time'].min(),
            end=call_offs['required_time'].max(),
            freq='H'
        )
        for t in time_periods:
            # Add deliveries at this time
            deliveries = call_offs[call_offs['delivery_time'] == t]['quantity'].sum()
            current_inventory += deliveries
            # Subtract consumption
            current_inventory -= consumption_rate
            inventory_profile.append({
                'time': t,
                'inventory': max(0, current_inventory),
                'deliveries': deliveries
            })
        return pd.DataFrame(inventory_profile)
# Example usage
assembly_schedule = pd.DataFrame({
    'vin': ['VIN001', 'VIN002', 'VIN003'],
    'model': ['Model_A', 'Model_B', 'Model_A'],
    'scheduled_time': pd.to_datetime([
        '2025-01-20 08:00',
        '2025-01-20 09:00',
        '2025-01-20 10:00'
    ])
})
bom = pd.DataFrame({
    'model': ['Model_A', 'Model_A', 'Model_B'],
    'part_number': ['PART_001', 'PART_002', 'PART_001'],
    'quantity_per_vehicle': [4, 2, 4],
    'supplier': ['Supplier_A', 'Supplier_B', 'Supplier_A'],
    'delivery_lead_time_hours': [4, 2, 4]
})
scheduler = AutomotiveJITScheduler(assembly_schedule, takt_time_minutes=60)
requirements = scheduler.calculate_part_requirements(bom)
call_offs = scheduler.generate_supplier_call_off(requirements, buffer_hours=2)
print("Supplier Call-Off Schedule:")
print(call_offs[['supplier', 'part_number', 'delivery_time', 'quantity']])

class SequencedPartsManager:
    """
    Manage sequenced parts delivery (e.g., seats, cockpits)
    """
    def __init__(self, build_sequence):
        self.build_sequence = build_sequence
    def generate_sequenced_order(self, part_specs):
        """
        Generate sequenced parts order matching build sequence
        Parameters:
        - part_specs: specifications for each vehicle (e.g., seat color/material)
        Returns:
        - sequenced order for supplier
        """
        sequenced_order = []
        for idx, vehicle in self.build_sequence.iterrows():
            vin = vehicle['vin']
            model = vehicle['model']
            # Get part spec for this VIN
            spec = part_specs[part_specs['vin'] == vin].iloc[0]
            sequenced_order.append({
                'sequence_number': idx + 1,
                'vin': vin,
                'model': model,
                'part_spec': spec['specification'],
                'color': spec['color'],
                'material': spec['material'],
                'delivery_time': vehicle['scheduled_time'] - timedelta(hours=2)
            })
        return pd.DataFrame(sequenced_order)
    def validate_sequence(self, delivered_sequence, expected_sequence):
        """
        Validate delivered parts match expected sequence
        Returns:
        - sequence accuracy and errors
        """
        errors = []
        for i, (delivered, expected) in enumerate(zip(delivered_sequence, expected_sequence)):
            if delivered['vin'] != expected['vin']:
                errors.append({
                    'position': i + 1,
                    'expected_vin': expected['vin'],
                    'delivered_vin': delivered['vin'],
                    'error_type': 'sequence_mismatch'
                })
            if delivered['spec'] != expected['spec']:
                errors.append({
                    'position': i + 1,
                    'vin': delivered['vin'],
                    'expected_spec': expected['spec'],
                    'delivered_spec': delivered['spec'],
                    'error_type': 'specification_mismatch'
                })
        accuracy = 1 - (len(errors) / len(expected_sequence))
        return {
            'accuracy_pct': accuracy * 100,
            'errors': errors,
            'error_count': len(errors)
        }

class AutomotiveQualityManager:
    """
    Track supplier quality performance (PPM defects)
    """
    def __init__(self, target_ppm=50):
        """
        Initialize quality manager
        Parameters:
        - target_ppm: target defect rate (parts per million)
        """
        self.target_ppm = target_ppm
        self.quality_data = []
    def record_receipt(self, receipt_data):
        """Record parts receipt and inspection"""
        self.quality_data.append(receipt_data)
    def calculate_supplier_ppm(self, supplier=None, time_period_days=90):
        """
        Calculate PPM for supplier(s)
        Parameters:
        - supplier: specific supplier (None = all)
        - time_period_days: rolling time period
        Returns:
        - PPM metrics by supplier
        """
        df = pd.DataFrame(self.quality_data)
        # Filter time period
        cutoff_date = datetime.now() - timedelta(days=time_period_days)
        df = df[df['receipt_date'] >= cutoff_date]
        # Filter supplier if specified
        if supplier:
            df = df[df['supplier'] == supplier]
        # Calculate PPM by supplier
        supplier_ppm = df.groupby('supplier').agg({
            'quantity_received': 'sum',
            'quantity_defective': 'sum'
        })
        supplier_ppm['ppm'] = (
            supplier_ppm['quantity_defective'] /
            supplier_ppm['quantity_received'] * 1000000
        )
        supplier_ppm['meets_target'] = supplier_ppm['ppm'] <= self.target_ppm