lean-manufacturing

リーン製造

リーン製造とオペレーショナルエクセレンスの専門家です。組織からムダを排除し、フローを最適化し、リードタイムを短縮し、品質を向上させ、継続的改善の文化を築くことを支援します。

初期評価

リーンの取り組みを実施する前に、以下を把握してください。

1. 現状

   • どのような問題や非効率が存在しますか？
   • 現在のプロセスフローとサイクルタイムはどのくらいですか？
   • 既知のムダやボトルネックはありますか？
   • 品質上の問題や不良率はどのような状況ですか？

2. 組織の状況

   • 製造形態は何ですか？（個別生産、プロセス生産、組立、受注生産）
   • 生産量と製品の種類はどのくらいですか？
   • リーンの原則に関する現在の習熟度はどの程度ですか？
   • 変革に対する経営陣の支援はありますか？

3. 改善目標

   • 主な目標は何ですか？（リードタイムの短縮、コスト削減、品質向上）
   • 目標指標と改善目標値は何ですか？
   • 利用可能な期間とリソースはどのくらいですか？
   • クイックウィンと戦略的変革のどちらを目指しますか？

4. 対象範囲

   • 対象となるプロセスまたはバリューストリームはどれですか？
   • 工場全体ですか、それともパイロットエリアですか？
   • サプライチェーンも関与しますか？
   • 組織文化は変革に対応できる状態ですか？

---

リーン製造のフレームワーク

リーンの基本原則

1. 価値

• 顧客の視点から価値を定義する
• 顧客は何に対価を支払う意思がありますか？
• それ以外のものはすべて排除する

2. バリューストリーム

• 価値を生み出す工程と生み出さない工程を含む、すべての工程をマッピングする
• ムダを特定して排除する
• エンドツーエンドのフローを最適化する

3. フロー

• 価値を生み出す工程がスムーズに流れるようにする
• 中断、遅延、バッチ処理を排除する
• 可能な場合は一個流しにする

4. プル

• 顧客の需要があるものだけを生産する
• 実際の消費を生産のトリガーにする
• 過剰生産を避ける

5. 完璧

• 継続的改善（カイゼン）
• 不良ゼロ、ムダゼロを目指す
• すべての従業員を改善活動に巻き込む

7つのムダ（ムダ）

1. 運搬

• 材料の不要な移動
• 複数回の取り扱い、長距離の移動
• 例：工場内での材料の移動、往復移動

2. 在庫

• 原材料、仕掛品、完成品の過剰在庫
• 問題（品質、信頼性、計画）を覆い隠す
• 資金を拘束し、保管スペースを必要とする

3. 動作

• 人の不要な動き
• 手を伸ばす、歩く、探す
• 例：レイアウトの不備、整理整頓の不足、工具の不足

4. 待ち

• 人や機械のアイドル時間
• 材料、情報、承認を待つこと
• 作業負荷の偏り、ボトルネック

5. 過剰生産

• 必要量を超える、または必要な時期より早い生産
• 最悪のムダ（ほかのムダを引き起こす）
• 例：大ロット生産、需要ではなく予測に基づく生産

6. 過剰加工

• 顧客が求める以上の作業を行うこと
• 過剰な品質、不必要な工程
• 例：追加承認、重複検査、過剰設計

7. 不良

• エラー、手直し、廃棄、返品
• 検査や修正にかかる労力
• 顧客の不満

追加のムダ（8つ目のムダ）：

• 人材の未活用：従業員のアイデアやスキルを活用しないこと

---

バリューストリームマッピング（VSM）

現状のVSM

目的： 現在のプロセスフローを記録し、ムダを特定する

手順：

1. 製品ファミリーまたはバリューストリームを選択する
2. プロセスを歩いて確認する（現地・現物・現実）
3. 各プロセス工程をマッピングする
4. データを収集する（サイクルタイム、段取り替え、稼働率、不良、在庫）
5. 物と情報の流れをマッピングする
6. 付加価値時間と非付加価値時間を算出する
7. 改善の機会を特定する

VSM分析コード：

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches

class ValueStreamMap:
    """
    Value Stream Mapping analysis and visualization
    Calculate lead time, value-added ratio, and identify waste
    """

    def __init__(self, process_steps):
        """
        process_steps: list of dicts with process details

        Example:
        {
            'name': 'Cutting',
            'cycle_time': 45,  # seconds
            'changeover_time': 30,  # minutes
            'uptime': 0.85,
            'batch_size': 500,
            'operators': 2,
            'inventory_days': 2.5,
            'distance_to_next': 50  # meters
        }
        """
        self.process_steps = process_steps
        self.df = pd.DataFrame(process_steps)

    def calculate_metrics(self):
        """Calculate key VSM metrics"""

        # Total cycle time (value-added time)
        total_cycle_time = self.df['cycle_time'].sum()

        # Total lead time (includes waiting/inventory)
        total_lead_time_days = self.df['inventory_days'].sum()
        total_lead_time_seconds = total_lead_time_days * 24 * 3600

        # Value-added ratio
        va_ratio = (total_cycle_time / total_lead_time_seconds) * 100

        # Total distance traveled
        total_distance = self.df['distance_to_next'].sum()

        # Total operators
        total_operators = self.df['operators'].sum()

        # OEE (Overall Equipment Effectiveness)
        # Simplified: uptime * performance * quality (assume 100% quality for now)
        avg_uptime = self.df['uptime'].mean()

        metrics = {
            'total_cycle_time_sec': total_cycle_time,
            'total_cycle_time_min': total_cycle_time / 60,
            'total_lead_time_days': total_lead_time_days,
            'total_lead_time_hours': total_lead_time_days * 24,
            'value_added_ratio_pct': va_ratio,
            'total_distance_m': total_distance,
            'total_operators': total_operators,
            'avg_uptime_pct': avg_uptime * 100
        }

        return metrics

    def identify_waste(self):
        """Identify and quantify waste in the value stream"""

        waste_analysis = []

        # Transportation waste
        total_transport = self.df['distance_to_next'].sum()
        waste_analysis.append({
            'waste_type': 'Transportation',
            'quantity': total_transport,
            'unit': 'meters',
            'impact': 'High' if total_transport > 200 else 'Medium' if total_transport > 100 else 'Low'
        })

        # Inventory waste
        total_inventory_days = self.df['inventory_days'].sum()
        waste_analysis.append({
            'waste_type': 'Inventory',
            'quantity': total_inventory_days,
            'unit': 'days',
            'impact': 'High' if total_inventory_days > 10 else 'Medium' if total_inventory_days > 5 else 'Low'
        })

        # Waiting waste (from low uptime)
        avg_uptime = self.df['uptime'].mean()
        downtime_pct = (1 - avg_uptime) * 100
        waste_analysis.append({
            'waste_type': 'Waiting (Downtime)',
            'quantity': downtime_pct,
            'unit': 'percent',
            'impact': 'High' if downtime_pct > 20 else 'Medium' if downtime_pct > 10 else 'Low'
        })

        # Overproduction (large batches)
        avg_batch = self.df['batch_size'].mean()
        waste_analysis.append({
            'waste_type': 'Overproduction',
            'quantity': avg_batch,
            'unit': 'units',
            'impact': 'High' if avg_batch > 1000 else 'Medium' if avg_batch > 500 else 'Low'
        })

        # Value-added ratio (overall waste indicator)
        metrics = self.calculate_metrics()
        va_ratio = metrics['value_added_ratio_pct']
        waste_analysis.append({
            'waste_type': 'Overall VA Ratio',
            'quantity': va_ratio,
            'unit': 'percent',
            'impact': 'Excellent' if va_ratio > 10 else 'Good' if va_ratio > 5 else 'Poor'
        })

        return pd.DataFrame(waste_analysis)

    def calculate_takt_time(self, customer_demand_per_day, available_work_time_minutes):
        """
        Calculate takt time = available time / customer demand

        Takt time is the pace of customer demand
        Process should be designed to match takt time
        """

        takt_time_minutes = available_work_time_minutes / customer_demand_per_day
        takt_time_seconds = takt_time_minutes * 60

        # Compare to current cycle time
        total_cycle_time = self.df['cycle_time'].sum()

        return {
            'takt_time_seconds': takt_time_seconds,
            'takt_time_minutes': takt_time_minutes,
            'current_cycle_time_seconds': total_cycle_time,
            'cycle_time_vs_takt': 'OK' if total_cycle_time <= takt_time_seconds else 'TOO SLOW',
            'margin_seconds': takt_time_seconds - total_cycle_time
        }

    def generate_future_state_recommendations(self):
        """Generate improvement recommendations for future state VSM"""

        recommendations = []

        metrics = self.calculate_metrics()

        # Inventory reduction
        if metrics['total_lead_time_days'] > 5:
            recommendations.append({
                'area': 'Inventory Reduction',
                'current': f"{metrics['total_lead_time_days']:.1f} days",
                'target': f"{metrics['total_lead_time_days'] * 0.5:.1f} days",
                'actions': [
                    'Implement pull/kanban system',
                    'Reduce batch sizes',
                    'Improve supplier delivery frequency',
                    'Eliminate safety stock where possible'
                ]
            })

        # Layout optimization
        if metrics['total_distance_m'] > 100:
            recommendations.append({
                'area': 'Layout Optimization',
                'current': f"{metrics['total_distance_m']:.0f} meters",
                'target': f"{metrics['total_distance_m'] * 0.3:.0f} meters",
                'actions': [
                    'Cellular manufacturing layout',
                    'Collocate sequential processes',
                    'Point-of-use material storage',
                    'Eliminate backtracking'
                ]
            })

        # Flow improvement
        if metrics['value_added_ratio_pct'] < 5:
            recommendations.append({
                'area': 'Flow Improvement',
                'current': f"{metrics['value_added_ratio_pct']:.2f}% VA ratio",
                'target': '>10% VA ratio',
                'actions': [
                    'Implement continuous flow (one-piece flow)',
                    'Eliminate batching where possible',
                    'Balance workloads across processes',
                    'Create FIFO lanes between processes'
                ]
            })

        # Equipment reliability
        if metrics['avg_uptime_pct'] < 85:
            recommendations.append({
                'area': 'Equipment Reliability',
                'current': f"{metrics['avg_uptime_pct']:.1f}% uptime",
                'target': '>95% uptime',
                'actions': [
                    'Implement TPM (Total Productive Maintenance)',
                    'Autonomous maintenance by operators',
                    'Preventive maintenance schedule',
                    'Root cause analysis on breakdowns'
                ]
            })

        return recommendations

# Example usage
process_steps = [
    {
        'name': 'Raw Material Storage',
        'cycle_time': 0,
        'changeover_time': 0,
        'uptime': 1.0,
        'batch_size': 5000,
        'operators': 0,
        'inventory_days': 5.0,
        'distance_to_next': 80
    },
    {
        'name': 'Cutting',
        'cycle_time': 45,
        'changeover_time': 30,
        'uptime': 0.85,
        'batch_size': 500,
        'operators': 2,
        'inventory_days': 2.5,
        'distance_to_next': 50
    },
    {
        'name': 'Welding',
        'cycle_time': 120,
        'changeover_time': 45,
        'uptime': 0.78,
        'batch_size': 500,
        'operators': 3,
        'inventory_days': 3.0,
        'distance_to_next': 30
    },
    {
        'name': 'Assembly',
        'cycle_time': 180,
        'changeover_time': 20,
        'uptime': 0.92,
        'batch_size': 100,
        'operators': 4,
        'inventory_days': 1.5,
        'distance_to_next': 25
    },
    {
        'name': 'Inspection',
        'cycle_time': 60,
        'changeover_time': 0,
        'uptime': 0.95,
        'batch_size': 100,
        'operators': 1,
        'inventory_days': 0.5,
        'distance_to_next': 15
    },
    {
        'name': 'Finished Goods',
        'cycle_time': 0,
        'changeover_time': 0,
        'uptime': 1.0,
        'batch_size': 100,
        'operators': 0,
        'inventory_days': 4.0,
        'distance_to_next': 0
    }
]

vsm = ValueStreamMap(process_steps)

# Calculate metrics
metrics = vsm.calculate_metrics()
print("Value Stream Metrics:")
print(f"  Total Cycle Time: {metrics['total_cycle_time_min']:.1f} minutes")
print(f"  Total Lead Time: {metrics['total_lead_time_days']:.1f} days")
print(f"  Value-Added Ratio: {metrics['value_added_ratio_pct']:.2f}%")
print(f"  Total Distance: {metrics['total_distance_m']:.0f} meters")

# Identify waste
print("\nWaste Analysis:")
waste_df = vsm.identify_waste()
print(waste_df)

# Takt time
takt = vsm.calculate_takt_time(customer_demand_per_day=400, available_work_time_minutes=450)
print(f"\nTakt Time: {takt['takt_time_seconds']:.1f} seconds")
print(f"Current Cycle Time: {takt['current_cycle_time_seconds']:.0f} seconds")
print(f"Status: {takt['cycle_time_vs_takt']}")

# Recommendations
print("\nFuture State Recommendations:")
recommendations = vsm.generate_future_state_recommendations()
for rec in recommendations:
    print(f"\n{rec['area']}:")
    print(f"  Current: {rec['current']}")
    print(f"  Target: {rec['target']}")
    print(f"  Actions:")
    for action in rec['actions']:
        print(f"    - {action}")

---

5S職場管理

5S システム

1. 整理（Seiri）

• 必要なものと不要なものを分けます
• 不要なものを取り除きます
• 赤札作戦を実施します

2. 整頓（Seiton）

• 残ったものを整理します
• すべてのものに置き場所を決め、決めた場所に置きます
• 目で見る管理、ラベル、影絵ボードを活用します

3. 清掃（Seiso）

• 清掃と点検を行います
• 清掃を点検の一環として行います
• 異常を特定します

4. 清潔（Seiketsu）

• 標準と手順を作成します
• 5S を維持するために視覚的な管理を行います
• チェックリストと監査を活用します

5. 躾（Shitsuke）

• 規律と責任感を持ちます
• 定期的に監査します
• 標準を継続的に改善します

5S 評価ツール

class FiveSAssessment:
    """
    5S workplace assessment and scoring
    """

    def __init__(self):
        self.criteria = {
            'Sort': [
                'Only needed items present in work area',
                'No unnecessary materials, tools, or equipment',
                'Clear distinction between needed and unneeded items',
                'Regular red tag events conducted'
            ],
            'Set in Order': [
                'Everything has a designated location',
                'Visual controls and labels in place',
                'Items stored near point of use',
                'Easy to find and return items'
            ],
            'Shine': [
                'Work area is clean and free of debris',
                'Equipment is clean and well-maintained',
                'Cleaning standards are defined',
                'Cleaning is part of daily routine'
            ],
            'Standardize': [
                'Standard procedures documented',
                'Visual management tools in place',
                'Responsibilities are clearly assigned',
                'Standards are easy to follow'
            ],
            'Sustain': [
                'Regular 5S audits conducted',
                'All team members follow 5S practices',
                'Continuous improvement of standards',
                'Management support visible'
            ]
        }

    def conduct_assessment(self, scores):
        """
        Conduct 5S assessment

        Parameters:
        - scores: dict {S: [ratings]} where rating is 1-5 for each criterion

        Example:
        scores = {
            'Sort': [4, 5, 3, 4],
            'Set in Order': [3, 4, 4, 3],
            ...
        }

        Returns assessment results with scores and recommendations
        """

        results = {}

        for s, ratings in scores.items():
            avg_score = np.mean(ratings)
            max_score = 5.0

            results[s] = {
                'score': avg_score,
                'percentage': (avg_score / max_score) * 100,
                'level': self._get_level(avg_score),
                'criteria_scores': list(zip(self.criteria[s], ratings))
            }

        # Overall score
        overall = np.mean([r['score'] for r in results.values()])
        results['Overall'] = {
            'score': overall,
            'percentage': (overall / 5.0) * 100,
            'level': self._get_level(overall)
        }

        return results

    def _get_level(self, score):
        """Determine 5S maturity level"""
        if score >= 4.5:
            return 'World Class'
        elif score >= 4.0:
            return 'Excellent'
        elif score >= 3.5:
            return 'Good'
        elif score >= 3.0:
            return 'Fair'
        elif score >= 2.0:
            return 'Needs Improvement'
        else:
            return 'Poor'

    def generate_action_plan(self, results):
        """Generate improvement action plan based on assessment"""

        actions = []

        for s, result in results.items():
            if s == 'Overall':
                continue

            if result['score'] < 3.5:
                # Priority improvement area
                actions.append({
                    'area': s,
                    'priority': 'High' if result['score'] < 2.5 else 'Medium',
                    'current_score': result['score'],
                    'target_score': 4.0,
                    'gap': 4.0 - result['score'],
                    'focus_areas': [
                        criterion for criterion, score in result['criteria_scores']
                        if score < 3
                    ]
                })

        return sorted(actions, key=lambda x: x['gap'], reverse=True)

# Example usage
assessment = FiveSAssessment()

scores = {
    'Sort': [4, 4, 3, 4],
    'Set in Order': [3, 3, 4, 3],
    'Shine': [5, 4, 5, 4],
    'Standardize': [2, 3, 2, 3],
    'Sustain': [2, 2, 3, 2]
}

results = assessment.conduct_assessment(scores)

print("5S Assessment Results:\n")
for s, result in results.items():
    print(f"{s}: {result['score']:.2f} / 5.00 ({result['percentage']:.1f}%) - {result['level']}")

print("\n\nAction Plan:")
action_plan = assessment.generate_action_plan(results)
for action in action_plan:
    print(f"\n{action['area']} - Priority: {action['priority']}")
    print(f"  Current Score: {action['current_score']:.2f}")
    print(f"  Target Score: {action['target_score']:.2f}")
    print(f"  Focus Areas:")
    for area in action['focus_areas']:
        print(f"    - {area}")

---

かんばんプルシステム

かんばんの計算

class KanbanSystem:
    """
    Kanban system design and calculation
    Determine number of kanban cards needed
    """

    def __init__(self, demand_rate, lead_time, container_size, safety_factor=1.1):
        """
        Parameters:
        - demand_rate: average daily demand (units/day)
        - lead_time: replenishment lead time (days)
        - container_size: units per kanban container
        - safety_factor: buffer factor (typically 1.0 - 1.2)
        """
        self.demand_rate = demand_rate
        self.lead_time = lead_time
        self.container_size = container_size
        self.safety_factor = safety_factor

    def calculate_number_of_kanbans(self):
        """
        Calculate number of kanban cards required

        Formula: N = (D × L × (1 + S)) / C
        Where:
        - N = number of kanbans
        - D = demand rate
        - L = lead time
        - S = safety factor (0.1 = 10% buffer)
        - C = container size
        """

        n_kanbans = (self.demand_rate * self.lead_time * self.safety_factor) / self.container_size

        # Round up
        n_kanbans = int(np.ceil(n_kanbans))

        # Calculate inventory levels
        max_inventory = n_kanbans * self.container_size
        avg_inventory = max_inventory / 2
        safety_stock = max_inventory - (self.demand_rate * self.lead_time)

        return {
            'number_of_kanbans': n_kanbans,
            'container_size': self.container_size,
            'max_inventory': max_inventory,
            'average_inventory': avg_inventory,
            'safety_stock': safety_stock,
            'inventory_days': max_inventory / self.demand_rate,
            'turnover_rate': self.demand_rate * 365 / avg_inventory
        }

    def optimize_container_size(self, min_size=10, max_size=200, step=10):
        """
        Find optimal container size to minimize inventory
        while meeting service level
        """

        results = []

        for container_size in range(min_size, max_size + 1, step):
            self.container_size = container_size
            kanban_calc = self.calculate_number_of_kanbans()

            results.append({
                'container_size': container_size,
                'num_kanbans': kanban_calc['number_of_kanbans'],
                'avg_inventory': kanban_calc['average_inventory'],
                'inventory_days': kanban_calc['inventory_days'],
                'turnover_rate': kanban_calc['turnover_rate']
            })

        df = pd.DataFrame(results)

        # Find optimal (minimum average inventory while ensuring coverage)
        optimal = df.loc[df['avg_inventory'].idxmin()]

        return {
            'optimal_container_size': optimal['container_size'],
            'optimal_num_kanbans': optimal['num_kanbans'],
            'analysis': df
        }

    def simulate_kanban_system(self, days=30, demand_variation=0.1):
        """
        Simulate kanban system performance

        Parameters:
        - days: simulation period
        - demand_variation: coefficient of variation for demand
        """

        n_kanbans = self.calculate_number_of_kanbans()['number_of_kanbans']

        # Initialize
        inventory = n_kanbans * self.container_size
        stockouts = 0
        daily_inventory = []

        for day in range(days):
            # Random demand (normal distribution)
            daily_demand = max(0, np.random.normal(
                self.demand_rate,
                self.demand_rate * demand_variation
            ))

            # Consume inventory
            inventory -= daily_demand

            # Replenishment (if kanban triggered and lead time elapsed)
            # Simplified: replenish when below reorder point
            reorder_point = self.demand_rate * self.lead_time
            if inventory < reorder_point:
                # Replenish (assuming lead time already passed)
                inventory += self.container_size

            # Check for stockout
            if inventory < 0:
                stockouts += 1
                inventory = 0

            daily_inventory.append(inventory)

        return {
            'avg_inventory': np.mean(daily_inventory),
            'min_inventory': np.min(daily_inventory),
            'max_inventory': np.max(daily_inventory),
            'stockout_days': stockouts,
            'service_level': (days - stockouts) / days * 100,
            'daily_inventory': daily_inventory
        }

# Example usage
kanban = KanbanSystem(
    demand_rate=100,      # 100 units per day
    lead_time=3,          # 3 days replenishment time
    container_size=50,    # 50 units per container
    safety_factor=1.1     # 10% buffer
)

result = kanban.calculate_number_of_kanbans()
print("Kanban System Design:")
print(f"  Number of Kanbans: {result['number_of_kanbans']}")
print(f"  Container Size: {result['container_size']} units")
print(f"  Maximum Inventory: {result['max_inventory']} units")
print(f"  Average Inventory: {result['average_inventory']:.0f} units")
print(f"  Inventory Days: {result['inventory_days']:.1f} days")
print(f"  Turnover Rate: {result['turnover_rate']:.1f}x per year")

# Optimize container size
print("\nOptimizing Container Size...")
optimization = kanban.optimize_container_size(min_size=20, max_size=100, step=10)
print(f"  Optimal Container Size: {optimization['optimal_container_size']} units")
print(f"  Optimal Number of Kanbans: {optimization['optimal_num_kanbans']}")

# Simulate
print("\nSimulating Kanban System (30 days)...")
simulation = kanban.simulate_kanban_system(days=30)
print(f"  Average Inventory: {simulation['avg_inventory']:.1f} units")
print(f"  Service Level: {simulation['service_level']:.1f}%")
print(f"  Stockout Days: {simulation['stockout_days']}")

---

SMED（シングル段取り）

段取り替え削減の分析

class SMEDAnalysis:
    """
    SMED (Single Minute Exchange of Die) changeover reduction
    Analyze and reduce setup/changeover times
    """

    def __init__(self, changeover_steps):
        """
        changeover_steps: list of dicts with step details

        Example:
        {
            'step': 'Remove old die',
            'type': 'Internal',  # Internal or External
            'duration_minutes': 15,
            'category': 'Removal'
        }
        """
        self.steps = pd.DataFrame(changeover_steps)

    def analyze_current_state(self):
        """Analyze current changeover process"""

        # Total time by internal/external
        time_by_type = self.steps.groupby('type')['duration_minutes'].sum()

        # Total changeover time
        total_internal = time_by_type.get('Internal', 0)
        total_external = time_by_type.get('External', 0)
        total_time = total_internal + total_external

        # Internal time is the bottleneck (machine downtime)
        # External can be done while machine is running

        analysis = {
            'total_changeover_time': total_time,
            'internal_time': total_internal,
            'external_time': total_external,
            'internal_pct': (total_internal / total_time * 100) if total_time > 0 else 0,
            'steps_breakdown': self.steps.groupby(['type', 'category'])['duration_minutes'].sum()
        }

        return analysis

    def apply_smed_principles(self):
        """
        Apply SMED principles to reduce changeover time

        SMED Stages:
        1. Separate internal from external activities
        2. Convert internal to external
        3. Streamline internal activities
        4. Streamline external activities
        """

        recommendations = []

        # Stage 1: Identify internal that should be external
        internal_steps = self.steps[self.steps['type'] == 'Internal']

        for idx, step in internal_steps.iterrows():
            if 'Fetch' in step['step'] or 'Prepare' in step['step'] or 'Get' in step['step']:
                recommendations.append({
                    'step': step['step'],
                    'current_time': step['duration_minutes'],
                    'stage': 'Stage 2: Convert to External',
                    'action': 'Perform this step before stopping machine',
                    'time_saved': step['duration_minutes'],
                    'priority': 'High'
                })

        # Stage 3: Streamline internal
        for idx, step in internal_steps.iterrows():
            if step['duration_minutes'] > 5:
                recommendations.append({
                    'step': step['step'],
                    'current_time': step['duration_minutes'],
                    'stage': 'Stage 3: Streamline Internal',
                    'action': 'Use quick-release fasteners, standardize tools, use guides',
                    'time_saved': step['duration_minutes'] * 0.5,  # Assume 50% reduction
                    'priority': 'High' if step['duration_minutes'] > 10 else 'Medium'
                })

        # Stage 4: Standardize and parallel operations
        recommendations.append({
            'step': 'General',
            'current_time': 0,
            'stage': 'Stage 4: Standardize',
            'action': 'Create changeover checklist, use multiple operators in parallel',
            'time_saved': self.steps['duration_minutes'].sum() * 0.2,  # Assume 20% improvement
            'priority': 'Medium'
        })

        return pd.DataFrame(recommendations)

    def calculate_improvement_impact(self, recommendations_df, annual_changeovers):
        """Calculate business impact of SMED improvements"""

        total_time_saved = recommendations_df['time_saved'].sum()

        current_analysis = self.analyze_current_state()
        current_time = current_analysis['total_changeover_time']
        improved_time = current_time - total_time_saved

        # Calculate productivity impact
        annual_time_saved_hours = (total_time_saved / 60) * annual_changeovers
        annual_time_saved_days = annual_time_saved_hours / 8

        # Assuming $100/hour downtime cost
        cost_per_hour = 100
        annual_cost_savings = annual_time_saved_hours * cost_per_hour

        return {
            'current_changeover_time_min': current_time,
            'improved_changeover_time_min': improved_time,
            'time_reduction_min': total_time_saved,
            'improvement_pct': (total_time_saved / current_time * 100) if current_time > 0 else 0,
            'annual_changeovers': annual_changeovers,
            'annual_time_saved_hours': annual_time_saved_hours,
            'annual_time_saved_days': annual_time_saved_days,
            'annual_cost_savings': annual_cost_savings
        }

# Example usage
changeover_steps = [
    {'step': 'Fetch new die from storage', 'type': 'Internal', 'duration_minutes': 10, 'category': 'Preparation'},
    {'step': 'Get tools and equipment', 'type': 'Internal', 'duration_minutes': 5, 'category': 'Preparation'},
    {'step': 'Loosen bolts on old die', 'type': 'Internal', 'duration_minutes': 15, 'category': 'Removal'},
    {'step': 'Remove old die', 'type': 'Internal', 'duration_minutes': 20, 'category': 'Removal'},
    {'step': 'Clean machine surface', 'type': 'Internal', 'duration_minutes': 8, 'category': 'Cleaning'},
    {'step': 'Position new die', 'type': 'Internal', 'duration_minutes': 12, 'category': 'Installation'},
    {'step': 'Align and center die', 'type': 'Internal', 'duration_minutes': 18, 'category': 'Installation'},
    {'step': 'Tighten bolts', 'type': 'Internal', 'duration_minutes': 15, 'category': 'Installation'},
    {'step': 'Run test pieces', 'type': 'Internal', 'duration_minutes': 20, 'category': 'Adjustment'},
    {'step': 'Make adjustments', 'type': 'Internal', 'duration_minutes': 12, 'category': 'Adjustment'},
]

smed = SMEDAnalysis(changeover_steps)

# Current state
current = smed.analyze_current_state()
print("Current State Analysis:")
print(f"  Total Changeover Time: {current['total_changeover_time']:.0f} minutes")
print(f"  Internal Time (downtime): {current['internal_time']:.0f} minutes")
print(f"  External Time: {current['external_time']:.0f} minutes")

# Recommendations
print("\nSMED Recommendations:")
recommendations = smed.apply_smed_principles()
print(recommendations[['step', 'stage', 'action', 'time_saved', 'priority']])

# Impact
impact = smed.calculate_improvement_impact(recommendations, annual_changeovers=250)
print("\nImprovement Impact:")
print(f"  Current Changeover: {impact['current_changeover_time_min']:.0f} minutes")
print(f"  Improved Changeover: {impact['improved_changeover_time_min']:.0f} minutes")
print(f"  Improvement: {impact['improvement_pct']:.1f}%")
print(f"  Annual Time Saved: {impact['annual_time_saved_days']:.0f} days")
print(f"  Annual Cost Savings: ${impact['annual_cost_savings']:,.0f}")

---

カイゼン（継続的改善）

カイゼンイベントのフレームワーク

class KaizenEvent:
    """
    Kaizen event planning and tracking
    Rapid improvement workshop (typically 3-5 days)
    """

    def __init__(self, focus_area, current_metrics, target_metrics):
        """
        Parameters:
        - focus_area: description of improvement area
        - current_metrics: dict of baseline metrics
        - target_metrics: dict of target metrics
        """
        self.focus_area = focus_area
        self.current_metrics = current_metrics
        self.target_metrics = target_metrics
        self.improvements = []

    def calculate_improvement_potential(self):
        """Calculate improvement gap and potential"""

        improvements = {}

        for metric, current_value in self.current_metrics.items():
            target_value = self.target_metrics.get(metric)

            if target_value is not None:
                # For metrics where lower is better (time, cost, defects)
                if 'time' in metric.lower() or 'cost' in metric.lower() or 'defect' in metric.lower():
                    improvement_pct = ((current_value - target_value) / current_value) * 100
                else:
                    # For metrics where higher is better (OEE, quality, output)
                    improvement_pct = ((target_value - current_value) / current_value) * 100

                improvements[metric] = {
                    'current': current_value,
                    'target': target_value,
                    'gap': target_value - current_value if 'time' not in metric.lower() else current_value - target_value,
                    'improvement_pct': improvement_pct
                }

        return improvements

    def track_action_items(self, action_items):
        """
        Track kaizen action items

        action_items: list of dicts with action details
        {
            'action': 'Install shadow board for tools',
            'responsible': 'John Smith',
            'due_date': '2025-02-15',
            'status': 'In Progress',
            'impact': 'Reduce tool search time by 5 min/day'
        }
        """

        self.action_items = pd.DataFrame(action_items)

        # Calculate completion rate
        if not self.action_items.empty:
            completed = (self.action_items['status'] == 'Completed').sum()
            total = len(self.action_items)
            completion_rate = (completed / total) * 100
        else:
            completion_rate = 0

        return {
            'total_actions': len(self.action_items),
            'completed': (self.action_items['status'] == 'Completed').sum(),
            'in_progress': (self.action_items['status'] == 'In Progress').sum(),
            'not_started': (self.action_items['status'] == 'Not Started').sum(),
            'completion_rate': completion_rate
        }

    def measure_results(self, actual_metrics):
        """
        Measure actual results after kaizen implementation

        actual_metrics: dict of post-kaizen metrics
        """

        results = {}

        for metric, current_value in self.current_metrics.items():
            actual_value = actual_metrics.get(metric)
            target_value = self.target_metrics.get(metric)

            if actual_value is not None and target_value is not None:
                # Calculate actual improvement
                if 'time' in metric.lower() or 'cost' in metric.lower() or 'defect' in metric.lower():
                    actual_improvement = ((current_value - actual_value) / current_value) * 100
                    target_achieved = (current_value - actual_value) >= (current_value - target_value)
                else:
                    actual_improvement = ((actual_value - current_value) / current_value) * 100
                    target_achieved = (actual_value - current_value) >= (target_value - current_value)

                results[metric] = {
                    'baseline': current_value,
                    'target': target_value,
                    'actual': actual_value,
                    'improvement_pct': actual_improvement,
                    'target_achieved': target_achieved
                }

        return results

# Example usage
kaizen = KaizenEvent(
    focus_area="Assembly line efficiency improvement",
    current_metrics={
        'cycle_time_seconds': 180,
        'defect_rate_ppm': 5000,
        'changeover_time_minutes': 45,
        'oee_pct': 72
    },
    target_metrics={
        'cycle_time_seconds': 150,
        'defect_rate_ppm': 2000,
        'changeover_time_minutes': 20,
        'oee_pct': 85
    }
)

# Calculate potential
potential = kaizen.calculate_improvement_potential()
print("Improvement Potential:")
for metric, data in potential.items():
    print(f"  {metric}:")
    print(f"    Current: {data['current']}")
    print(f"    Target: {data['target']}")
    print(f"    Improvement: {data['improvement_pct']:.1f}%")

# Track actions
action_items = [
    {'action': 'Implement 5S in assembly area', 'responsible': 'Team Lead', 'due_date': '2025-02-15', 'status': 'Completed', 'impact': 'Reduce search time'},
    {'action': 'Install quick-release fixtures', 'responsible': 'Maintenance', 'due_date': '2025-02-20', 'status': 'In Progress', 'impact': 'Reduce changeover time'},
    {'action': 'Create standard work instructions', 'responsible': 'Quality', 'due_date': '2025-02-18', 'status': 'Completed', 'impact': 'Reduce defects'},
    {'action': 'Train operators on new process', 'responsible': 'Supervisor', 'due_date': '2025-02-22', 'status': 'Not Started', 'impact': 'Improve cycle time'},
]

tracking = kaizen.track_action_items(action_items)
print(f"\nAction Item Tracking:")
print(f"  Completion Rate: {tracking['completion_rate']:.1f}%")
print(f"  Completed: {tracking['completed']}")
print(f"  In Progress: {tracking['in_progress']}")
print(f"  Not Started: {tracking['not_started']}")

# Measure results (after implementation)
actual_metrics = {
    'cycle_time_seconds': 155,
    'defect_rate_ppm': 2500,
    'changeover_time_minutes': 25,
    'oee_pct': 80
}

results = kaizen.measure_results(actual_metrics)
print("\nKaizen Results:")
for metric, data in results.items():
    print(f"  {metric}:")
    print(f"    Baseline: {data['baseline']}")
    print(f"    Target: {data['target']}")
    print(f"    Actual: {data['actual']}")
    print(f"    Improvement: {data['improvement_pct']:.1f}%")
    print(f"    Target Achieved: {'✓' if data['target_achieved'] else '✗'}")

---

ツールとライブラリ

Python ライブラリ

分析と最適化：

• pandas：データ分析と操作
• numpy：数値計算
• matplotlib、seaborn、plotly：可視化と VSM 図
• scipy：プロセス改善のための統計分析
• simpy：プロセスフローの離散事象シミュレーション

リーン特化ツール：

• カスタム VSM ツール（上記の例を参照）
• 改善スケジュール用のガントチャート

商用リーンソフトウェア

バリューストリームマッピング：

• VSM ソフトウェア：LeanKit、Lucidchart、Microsoft Visio
• eVSM：電子バリューストリームマッピング
• SIMUL8：VSM によるプロセスシミュレーション

継続的改善プラットフォーム：

• KaiNexus：改善管理ソフトウェア
• Ingage：リーン日常管理システム
• Weever：デジタル日常管理と監査
• Lean Focus：視覚的管理ソフトウェア

製造実行システム（MES）：

• SAP ME：リーン機能を備えた製造実行システム
• Apriso：リーン製造実行システム
• Plex：リーンツールを備えたクラウド MES

---

よくある課題と解決策

課題：変化への抵抗

問題：

• 従業員がリーンの導入に消極的です
• 「これまでもずっとこのやり方でした」
• 失職への不安があります

解決策：

• 教育とトレーニングから始めます
• 改善に従業員を巻き込みます（ボトムアップ）
• 効果を示すため、早期に成果を上げます
• リーダーシップのコミットメントと模範的な行動を示します
• 雇用の安定に関する懸念に、透明性をもって対応します

課題：改善が定着しない

問題：

• 当初の改善が元に戻ります
• 5S の実施エリアが再び乱雑になります
• カイゼンの成果が維持されません

解決策：

• すべての改善に定着の段階を組み込みます
• 定期的な監査と視覚的な管理を行います
• 正しい行動を簡単にできるようにします（ポカヨケ）
• パフォーマンスを説明責任と結び付けます
• 継続的な働きかけと評価を行います

課題：指標の不整合

問題：

• リーンの目標（在庫削減）が、ほかの目標（機械稼働率の向上）と衝突します
• 局所最適化がシステム全体に悪影響を及ぼします

解決策：

• 指標をリーンの原則に合わせます
• システム全体の視点を持ちます（制約理論）
• 稼働率ではなく、フローに重点を置きます
• 在庫と過剰生産の真のコストについて理解を深めます
• インセンティブの仕組みを変更します

課題：複雑な製品構成

問題：

• 製品の種類が多く、標準化が困難です
• タクトタイムの設定が困難です
• 段取り替えが頻繁に発生します

解決策：

• 製品群ごとにまとめます
• 類似製品のセル生産方式を導入します
• SMED で段取り替えの負担を軽減します
• 混流生産のスケジューリングを行います
• モジュール設計と延期戦略を活用します

課題：サプライチェーンの不安定性

問題：

• JIT には信頼できるサプライヤーが必要です
• 供給のばらつきがフローを妨げます
• サプライヤーからのリードタイムが長くなります

解決策：

• サプライヤーの育成とパートナーシップの構築
• 戦略的バッファーを設けたプルシステム
• 重要品目の複数購買先確保
• ベンダー管理在庫（VMI）
• サプライヤーの可視性とコミュニケーションを改善します

---

出力形式

リーン評価レポート

エグゼクティブサマリー：

• 現状の成熟度レベル
• 特定された主なムダ
• 改善の優先事項
• 期待される効果

バリューストリーム分析：

指標	現状	将来の状態	改善
リードタイム	16.5 日	5.0 日	70% 削減
付加価値比率	2.4%	8.5%	6.1 ポイント増加
移動距離	200 m	60 m	70% 削減
在庫（日数）	12 日	3 日	75% 削減
不良率	5,000 ppm	1,000 ppm	80% 削減

ムダの分析：

ムダの種類	影響	根本原因	推奨する対策
作りすぎ	高	ロットサイズが大きい	カンバンを導入し、ロットサイズを 1/5 に削減する
待ち	高	ラインのバランスが取れていない	ラインバランスの調整、クロストレーニング
運搬	中	レイアウトが不適切	セル生産方式のレイアウト、工程の近接配置
在庫	高	安全在庫	プルシステム、信頼性の向上

導入ロードマップ：

フェーズ 1（1～3 か月目）：基盤づくり

• パイロットエリアで 5S を導入する
• 全従業員向けにリーンのトレーニングを実施する
• VSM を行うバリューストリームを選定する

フェーズ 2（4～6 か月目）：フロー

• パイロットエリアで連続フローを導入する
• 部品向けにカンバンシステムを導入する
• 視覚的な管理ボードを設置する

フェーズ 3（7～9 か月目）：プル

• カンバンをすべての資材に拡大する
• 生産計画を平準化する
• サプライヤー向けのプルシステムを導入する

フェーズ 4（10～12 か月目）：完璧の追求

• カイゼン文化を醸成し、イベントを実施する
• 他のバリューストリームにも拡大する
• 標準化し、定着させる

期待される効果：

• リードタイムの短縮：60～70%
• 在庫の削減：40～60%
• 生産性の向上：20～30%
• 品質の改善：不良を 50% 削減
• スペースの削減：30～40%
• コスト削減効果：年間 200 万～500 万ドル

---

質問事項

さらに詳しい情報が必要な場合：

1. 主な問題または改善の機会は何ですか？（品質、リードタイム、コスト、フロー）
2. 製造環境の種類は何ですか？（ジョブショップ、フロー、組立、プロセス）
3. 現在のリーン成熟度レベルはどの程度ですか？（リーンに取り組み始めたばかり、一部で取り組み中、成熟している）
4. 対象とするバリューストリームまたは工程エリアはどこですか？
5. 現在、どの指標を追跡していますか？
6. 経営陣の支援や利用可能なリソースはありますか？
7. 改善のスケジュールはどのようになっていますか？
8. 変化に対応できる文化的な準備は整っていますか？

---

関連スキル

• production-scheduling：生産計画と順序付けに使用
• quality-management：品質改善とシックスシグマに使用
• maintenance-planning：TPM と設備の信頼性向上に使用
• process-optimization：プロセスの分析と改善に使用
• assembly-line-balancing：ラインバランスの調整とタクトタイムの管理に使用
• inventory-optimization：在庫削減戦略に使用
• supply-chain-analytics：パフォーマンス測定に使用
• value-analysis：価値工学とコスト削減に使用