人間-ロボット協調 — 宇宙ステーションでの共同作業

import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
from dataclasses import dataclass, field
from typing import Optional
import heapq

@dataclass
class Task:
    """宇宙ステーションのタスクを表すクラス"""
    task_id: str
    name: str
    duration: float        # 時間（時間単位）
    priority: int          # 1(低) - 5(高)
    executor: str          # 'human', 'robot', 'both'
    dependencies: list = field(default_factory=list)
    category: str = 'maintenance'

# ISSの典型的な1日のタスクリスト
tasks = [
    Task('T01', 'Morning health check', 0.5, 5, 'human', [], 'health'),
    Task('T02', 'Air quality monitoring', 0.5, 4, 'robot', [], 'maintenance'),
    Task('T03', 'Equipment inventory scan', 1.0, 3, 'robot', [], 'maintenance'),
    Task('T04', 'Crystal growth experiment', 2.0, 5, 'human', ['T01'], 'science'),
    Task('T05', 'Solar panel inspection (ext)', 1.5, 4, 'robot', [], 'maintenance'),
    Task('T06', 'Water recycler maintenance', 1.0, 4, 'both', [], 'maintenance'),
    Task('T07', 'Protein analysis', 1.5, 5, 'human', ['T04'], 'science'),
    Task('T08', 'Module cleaning', 1.0, 2, 'robot', [], 'maintenance'),
    Task('T09', 'Communication system check', 0.5, 4, 'both', [], 'maintenance'),
    Task('T10', 'Fluid physics experiment', 2.0, 5, 'human', ['T01'], 'science'),
    Task('T11', 'Cargo manifest update', 0.5, 3, 'robot', ['T03'], 'logistics'),
    Task('T12', 'Exercise session', 2.0, 5, 'human', [], 'health'),
    Task('T13', 'Cable routing inspection', 1.0, 3, 'robot', [], 'maintenance'),
    Task('T14', 'Microscopy sample prep', 1.0, 4, 'both', [], 'science'),
    Task('T15', 'Thermal control check', 0.5, 3, 'robot', [], 'maintenance'),
    Task('T16', 'Earth observation photography', 0.5, 2, 'human', [], 'science'),
    Task('T17', 'Radiation monitor readout', 0.5, 4, 'robot', [], 'maintenance'),
    Task('T18', 'Experiment data downlink prep', 1.0, 4, 'both', ['T07', 'T10'], 'science'),
]

class CooperativeScheduler:
    """人間-ロボット協調タスクスケジューラ（貪欲法）"""

    def __init__(self, tasks: list, work_hours: float = 10.0):
        self.tasks = {t.task_id: t for t in tasks}
        self.work_hours = work_hours
        self.human_schedule = []   # [(start, end, task)]
        self.robot_schedule = []   # [(start, end, task)]
        self.human_time = 0.0
        self.robot_time = 0.0
        self.completed = set()
        self.unscheduled = []

    def _dependencies_met(self, task: Task) -> bool:
        """依存タスクがすべて完了しているかチェック"""
        return all(dep in self.completed for dep in task.dependencies)

    def _dependency_completion_time(self, task: Task) -> float:
        """依存タスクの最遅完了時刻を計算"""
        max_time = 0.0
        for dep_id in task.dependencies:
            for schedule in [self.human_schedule, self.robot_schedule]:
                for start, end, t in schedule:
                    if t.task_id == dep_id:
                        max_time = max(max_time, end)
        return max_time

    def schedule(self):
        """貪欲法でタスクをスケジューリング"""
        # 優先度降順 → 依存関係の深さ昇順でソート
        remaining = list(self.tasks.values())
        iteration = 0
        max_iterations = len(remaining) * 10  # 無限ループ防止

        while remaining and iteration < max_iterations:
            iteration += 1
            # 現在スケジュール可能なタスクを選択
            schedulable = []
            for task in remaining:
                if self._dependencies_met(task):
                    dep_time = self._dependency_completion_time(task)
                    schedulable.append((task, dep_time))

            if not schedulable:
                # 依存関係が解決されていないタスクは後回し
                # 完了済みタスクが増えるまで待つが、進捗がなければ打ち切り
                break

            # 優先度が高い順にソート
            schedulable.sort(key=lambda x: (-x[0].priority, x[1]))

            scheduled_this_round = False
            for task, dep_time in schedulable:
                if task not in remaining:
                    continue

                assigned = False
                if task.executor == 'human':
                    start = max(self.human_time, dep_time)
                    if start + task.duration <= self.work_hours:
                        self.human_schedule.append((start, start + task.duration, task))
                        self.human_time = start + task.duration
                        assigned = True
                elif task.executor == 'robot':
                    start = max(self.robot_time, dep_time)
                    if start + task.duration <= self.work_hours:
                        self.robot_schedule.append((start, start + task.duration, task))
                        self.robot_time = start + task.duration
                        assigned = True
                else:  # 'both' — 早く開始できる方に割り当て
                    human_start = max(self.human_time, dep_time)
                    robot_start = max(self.robot_time, dep_time)
                    if human_start <= robot_start and human_start + task.duration <= self.work_hours:
                        self.human_schedule.append((human_start, human_start + task.duration, task))
                        self.human_time = human_start + task.duration
                        assigned = True
                    elif robot_start + task.duration <= self.work_hours:
                        self.robot_schedule.append((robot_start, robot_start + task.duration, task))
                        self.robot_time = robot_start + task.duration
                        assigned = True
                    elif human_start + task.duration <= self.work_hours:
                        self.human_schedule.append((human_start, human_start + task.duration, task))
                        self.human_time = human_start + task.duration
                        assigned = True

                if assigned:
                    self.completed.add(task.task_id)
                    remaining.remove(task)
                    scheduled_this_round = True
                    break  # 1タスク割り当てたら再評価

            if not scheduled_this_round:
                # スケジュール不能なタスクを記録
                self.unscheduled = remaining
                break

        if remaining and not self.unscheduled:
            self.unscheduled = remaining

    def print_summary(self):
        """スケジュールのサマリーを表示"""
        print("=" * 60)
        print("Human-Robot Cooperative Schedule Summary")
        print("=" * 60)
        print(f"\n--- Astronaut Schedule ---")
        for start, end, task in self.human_schedule:
            print(f"  [{start:5.1f}h - {end:5.1f}h] {task.name} (P{task.priority})")
        print(f"  Total: {self.human_time:.1f}h / {self.work_hours:.1f}h")

        print(f"\n--- Robot Schedule ---")
        for start, end, task in self.robot_schedule:
            print(f"  [{start:5.1f}h - {end:5.1f}h] {task.name} (P{task.priority})")
        print(f"  Total: {self.robot_time:.1f}h / {self.work_hours:.1f}h")

        if self.unscheduled:
            print(f"\n--- Unscheduled Tasks ---")
            for task in self.unscheduled:
                print(f"  {task.name} (P{task.priority}, {task.executor})")

        # 統計
        human_science = sum(e - s for s, e, t in self.human_schedule if t.category == 'science')
        human_maint = sum(e - s for s, e, t in self.human_schedule if t.category == 'maintenance')
        robot_maint = sum(e - s for s, e, t in self.robot_schedule if t.category == 'maintenance')
        print(f"\n--- Statistics ---")
        print(f"  Astronaut science time: {human_science:.1f}h")
        print(f"  Astronaut maintenance time: {human_maint:.1f}h")
        print(f"  Robot maintenance time: {robot_maint:.1f}h")
        print(f"  Astronaut utilization: {self.human_time / self.work_hours * 100:.1f}%")
        print(f"  Robot utilization: {self.robot_time / self.work_hours * 100:.1f}%")

# スケジューリング実行
scheduler = CooperativeScheduler(tasks, work_hours=10.0)
scheduler.schedule()
scheduler.print_summary()

def plot_schedule(scheduler):
    """協調スケジュールをGanttチャートで可視化"""
    fig, axes = plt.subplots(2, 1, figsize=(14, 7), sharex=True)

    # カテゴリごとの色
    colors = {
        'science': '#00bcd4',     # cyan
        'health': '#4caf50',      # green
        'maintenance': '#ff9800', # orange
        'logistics': '#9c27b0',   # purple
    }

    # 宇宙飛行士のスケジュール
    ax = axes[0]
    ax.set_title('Astronaut Schedule', fontsize=13, fontweight='bold', color='#e0e0e0')
    for i, (start, end, task) in enumerate(scheduler.human_schedule):
        color = colors.get(task.category, '#607d8b')
        ax.barh(0, end - start, left=start, height=0.6, color=color,
                edgecolor='white', linewidth=0.5, alpha=0.9)
        if end - start >= 0.4:
            ax.text((start + end) / 2, 0, task.name,
                    ha='center', va='center', fontsize=7, color='white', fontweight='bold')

    # ロボットのスケジュール
    ax = axes[1]
    ax.set_title('Robot Schedule', fontsize=13, fontweight='bold', color='#e0e0e0')
    for i, (start, end, task) in enumerate(scheduler.robot_schedule):
        color = colors.get(task.category, '#607d8b')
        ax.barh(0, end - start, left=start, height=0.6, color=color,
                edgecolor='white', linewidth=0.5, alpha=0.9)
        if end - start >= 0.4:
            ax.text((start + end) / 2, 0, task.name,
                    ha='center', va='center', fontsize=7, color='white', fontweight='bold')

    for ax in axes:
        ax.set_xlim(0, scheduler.work_hours)
        ax.set_ylim(-0.5, 0.5)
        ax.set_yticks([])
        ax.set_facecolor('#1a1a2e')
        ax.grid(axis='x', alpha=0.3, color='gray')
        ax.tick_params(colors='#e0e0e0')

    axes[1].set_xlabel('Time (hours)', fontsize=11, color='#e0e0e0')

    # 凡例
    legend_patches = [mpatches.Patch(color=c, label=cat.capitalize())
                      for cat, c in colors.items()]
    fig.legend(handles=legend_patches, loc='upper right', fontsize=9,
              facecolor='#16213e', edgecolor='#e0e0e0', labelcolor='#e0e0e0')

    fig.set_facecolor('#0f0f23')
    plt.tight_layout()
    plt.savefig('hrc_schedule.png', dpi=150, bbox_inches='tight',
                facecolor='#0f0f23')
    plt.show()

plot_schedule(scheduler)

def analyze_collaboration_levels(tasks, work_hours=10.0):
    """
    協調レベルを変えたときのスケジュール効率を比較
    Level 1: ロボットなし（人間が全タスク実行）
    Level 2: 共存（ロボットは独立作業のみ）
    Level 3: 協調（'both'タスクを最適配分）
    """
    results = {}

    # Level 1: 人間のみ（ロボットなし）
    human_only_time = sum(t.duration for t in tasks)
    human_science_time_l1 = sum(t.duration for t in tasks if t.category == 'science')
    results['No Robot'] = {
        'total_time': human_only_time,
        'science_time': human_science_time_l1,
        'feasible': human_only_time <= work_hours
    }

    # Level 2: 共存（'both'タスクは全て人間が実行）
    tasks_l2 = []
    for t in tasks:
        t_copy = Task(t.task_id, t.name, t.duration, t.priority,
                      'human' if t.executor == 'both' else t.executor,
                      t.dependencies, t.category)
        tasks_l2.append(t_copy)
    sched_l2 = CooperativeScheduler(tasks_l2, work_hours)
    sched_l2.schedule()
    human_science_l2 = sum(e - s for s, e, t in sched_l2.human_schedule if t.category == 'science')
    results['Coexistence'] = {
        'human_util': sched_l2.human_time,
        'robot_util': sched_l2.robot_time,
        'science_time': human_science_l2,
        'unscheduled': len(sched_l2.unscheduled)
    }

    # Level 3: 協調（'both'タスクを最適配分）
    sched_l3 = CooperativeScheduler(tasks, work_hours)
    sched_l3.schedule()
    human_science_l3 = sum(e - s for s, e, t in sched_l3.human_schedule if t.category == 'science')
    results['Collaboration'] = {
        'human_util': sched_l3.human_time,
        'robot_util': sched_l3.robot_time,
        'science_time': human_science_l3,
        'unscheduled': len(sched_l3.unscheduled)
    }

    return results

results = analyze_collaboration_levels(tasks)

# 結果の可視化
fig, axes = plt.subplots(1, 3, figsize=(14, 5))

# 1. 宇宙飛行士の科学活動時間
labels = list(results.keys())
science_times = [results[l].get('science_time', 0) for l in labels]
bar_colors = ['#ff6b6b', '#ffd93d', '#6bcb77']
axes[0].bar(labels, science_times, color=bar_colors, edgecolor='white', linewidth=0.5)
axes[0].set_ylabel('Hours', color='#e0e0e0')
axes[0].set_title('Astronaut Science Time', fontsize=12, fontweight='bold', color='#e0e0e0')
axes[0].set_facecolor('#1a1a2e')
axes[0].tick_params(colors='#e0e0e0')
for i, v in enumerate(science_times):
    axes[0].text(i, v + 0.1, f'{v:.1f}h', ha='center', color='#e0e0e0', fontweight='bold')

# 2. 利用率
human_utils = [results[l].get('human_util', results[l].get('total_time', 0)) for l in labels]
robot_utils = [results[l].get('robot_util', 0) for l in labels]
x = np.arange(len(labels))
width = 0.35
axes[1].bar(x - width/2, human_utils, width, label='Astronaut', color='#00bcd4', edgecolor='white')
axes[1].bar(x + width/2, robot_utils, width, label='Robot', color='#ff9800', edgecolor='white')
axes[1].set_xticks(x)
axes[1].set_xticklabels(labels)
axes[1].set_ylabel('Hours', color='#e0e0e0')
axes[1].set_title('Utilization', fontsize=12, fontweight='bold', color='#e0e0e0')
axes[1].legend(facecolor='#16213e', edgecolor='#e0e0e0', labelcolor='#e0e0e0')
axes[1].set_facecolor('#1a1a2e')
axes[1].tick_params(colors='#e0e0e0')

# 3. 未スケジュールタスク数
unsched = [results[l].get('unscheduled', 0) if isinstance(results[l].get('unscheduled', 0), int)
           else (1 if not results[l].get('feasible', True) else 0) for l in labels]
axes[2].bar(labels, unsched, color=['#ff6b6b', '#ffd93d', '#6bcb77'], edgecolor='white')
axes[2].set_ylabel('Count', color='#e0e0e0')
axes[2].set_title('Unscheduled Tasks', fontsize=12, fontweight='bold', color='#e0e0e0')
axes[2].set_facecolor('#1a1a2e')
axes[2].tick_params(colors='#e0e0e0')
axes[2].yaxis.set_major_locator(plt.MaxNLocator(integer=True))

fig.set_facecolor('#0f0f23')
plt.tight_layout()
plt.savefig('hrc_comparison.png', dpi=150, bbox_inches='tight',
            facecolor='#0f0f23')
plt.show()

# 数値サマリー
print("\n=== Collaboration Level Comparison ===")
for level, data in results.items():
    print(f"\n{level}:")
    for key, val in data.items():
        print(f"  {key}: {val}")

def simulate_trust_dynamics(n_interactions=50, alpha=0.3, noise_std=0.1):
    """
    Lee & Moray (1994) の動的信頼モデルをシミュレーション
    T(t+1) = T(t) + alpha * [P(t) - T(t)]
    """
    np.random.seed(42)
    time_steps = np.arange(n_interactions)

    # ロボットのパフォーマンス（0-1, 1が完璧）
    # 学習により徐々に改善し、途中で故障イベント発生
    base_performance = 0.5 + 0.4 * (1 - np.exp(-time_steps / 20))
    noise = np.random.normal(0, noise_std, n_interactions)
    performance = np.clip(base_performance + noise, 0, 1)

    # 故障イベント: タスク25と40で大きなパフォーマンス低下
    performance[25] = 0.1
    performance[40] = 0.2

    # 信頼度のシミュレーション
    trust = np.zeros(n_interactions)
    trust[0] = 0.5  # 初期信頼度

    for t in range(n_interactions - 1):
        trust[t + 1] = trust[t] + alpha * (performance[t] - trust[t])
        trust[t + 1] = np.clip(trust[t + 1], 0, 1)

    # 適切な信頼の範囲を計算
    optimal_trust_lower = performance - 0.15
    optimal_trust_upper = performance + 0.15

    # 可視化
    fig, ax = plt.subplots(figsize=(12, 6))
    ax.fill_between(time_steps, optimal_trust_lower, optimal_trust_upper,
                    alpha=0.15, color='#00bcd4', label='Appropriate trust zone')
    ax.plot(time_steps, performance, 'o-', markersize=3, color='#ff9800',
            alpha=0.7, label='Robot performance P(t)', linewidth=1)
    ax.plot(time_steps, trust, 's-', markersize=4, color='#00bcd4',
            label=f'Trust T(t), α={alpha}', linewidth=2)

    # 故障イベントを注釈
    ax.annotate('Failure event', xy=(25, performance[25]),
                xytext=(28, 0.25), fontsize=9, color='#ff6b6b',
                arrowprops=dict(arrowstyle='->', color='#ff6b6b'))
    ax.annotate('Failure event', xy=(40, performance[40]),
                xytext=(43, 0.35), fontsize=9, color='#ff6b6b',
                arrowprops=dict(arrowstyle='->', color='#ff6b6b'))

    ax.set_xlabel('Interaction number', fontsize=11, color='#e0e0e0')
    ax.set_ylabel('Trust / Performance', fontsize=11, color='#e0e0e0')
    ax.set_title('Dynamic Trust Model: Astronaut-Robot Interaction',
                fontsize=13, fontweight='bold', color='#e0e0e0')
    ax.legend(fontsize=10, facecolor='#16213e', edgecolor='#e0e0e0', labelcolor='#e0e0e0')
    ax.set_xlim(0, n_interactions - 1)
    ax.set_ylim(0, 1.05)
    ax.set_facecolor('#1a1a2e')
    ax.tick_params(colors='#e0e0e0')
    ax.grid(alpha=0.2, color='gray')
    fig.set_facecolor('#0f0f23')
    plt.tight_layout()
    plt.savefig('trust_dynamics.png', dpi=150, bbox_inches='tight',
                facecolor='#0f0f23')
    plt.show()

simulate_trust_dynamics()