選好の有無が混在する場合のマッチング手法.py

# 選好の有無が混在する場合のマッチング手法
# https://proceedings-of-deim.github.io/DEIM2023/2b-6-5.pdf

import random
from typing import List, Tuple

def mixed_acceptance_algorithm(workers: List[Tuple[int, List[int]]], tasks: List[Tuple[int, List[int]]]) -> List[Tuple[int, int]]:
    """
    選好の有無が混在する場合のマッチングアルゴリズム

    Args:
        workers: ワーカーのリスト。各ワーカーは、タスクに対する選好リストと許容可能なタスク数を表すタプルで構成されます。
        tasks: タスクのリスト。各タスクは、ワーカーに対する選好リスト（存在する場合）と受け入れ可能なワーカー数を表すタプルで構成されます。

    Returns:
        マッチング結果のリスト。各要素は、ワーカーとマッチしたタスクのタプルです。
    """

    temlist: List[Tuple[int, int]] = []  # 仮マッチリスト
    conlist: List[Tuple[int, int]] = []  # 確定マッチリスト
    resultlist: List[Tuple[int, int]] = []  # マッチング結果リスト

    while any(w[0] > 0 and len(w[1]) > 0 for w in workers):
        for i, w in enumerate(workers):
            if w[0] > 0 and len(w[1]) > 0:  # ワーカーがまだマッチング可能かどうかを確認
                for j, t in enumerate(w[1]):  # ワーカーの選好リストを順に処理
                    if tasks[t][0] > 0:  # タスクがまだワーカーを受け入れ可能かどうかを確認
                        if tasks[t][1] is not None:  # タスクが選好リストを持つ場合
                            temlist.append((i, t))  # 仮マッチリストに追加
                            workers[i] = (workers[i][0] - 1, workers[i][1])  # ワーカーの受け入れ可能タスク数を減らす
                            tasks[t] = (tasks[t][0] - 1, tasks[t][1])  # タスクの受け入れ可能ワーカー数を減らす
                            break
                        else:  # タスクが選好リストを持たない場合
                            conlist.append((i, t))  # 確定マッチリストに追加
                            workers[i] = (workers[i][0] - 1, workers[i][1])  # ワーカーの受け入れ可能タスク数を減らす
                            tasks[t] = (tasks[t][0] - 1, tasks[t][1])  # タスクの受け入れ可能ワーカー数を減らす
                            if tasks[t][0] == 0:  # タスクの定員が満たされた場合
                                for w in workers:  # 他のワーカーの選好リストからこのタスクを削除
                                    if t in w[1]:
                                        w[1].remove(t)
                            break
                    else:
                        continue  # タスクが定員に達している場合は次のタスクへ

    resultlist = temlist + conlist  # 仮マッチリストと確定マッチリストを結合
    return resultlist

def generate_test_data(num_workers: int, num_tasks: int, num_no_preference_tasks: int, total_worker_capacity: int, total_task_capacity: int) -> Tuple[List[Tuple[int, List[int]]], List[Tuple[int, List[int]]]]:
    """
    テストデータ生成関数

    Args:
      num_workers: ワーカー数
      num_tasks: タスク数
      num_no_preference_tasks: 選好を持たないタスク数
      total_worker_capacity: ワーカー側の合計定員
      total_task_capacity: タスク側の合計定員

    Returns:
      ワーカーのリストとタスクのリスト
    """
    workers: List[Tuple[int, List[int]]] = []
    tasks: List[Tuple[int, List[int]]] = []

    # ワーカーの選好リストと定員を生成
    for i in range(num_workers):
        preference_list = list(range(num_tasks))
        random.shuffle(preference_list)
        capacity = random.randint(1, 5)
        workers.append((capacity, preference_list))

    # タスクの選好リストと定員を生成
    for i in range(num_tasks):
        if i < num_tasks - num_no_preference_tasks:
            preference_list = list(range(num_workers))
            random.shuffle(preference_list)
            capacity = random.randint(1, 5)
            tasks.append((capacity, preference_list))
        else:
            capacity = random.randint(1, 5)
            tasks.append((capacity, None))

    # 定員を調整
    worker_capacity = sum([w[0] for w in workers])
    task_capacity = sum([t[0] for t in tasks])

    while worker_capacity != total_worker_capacity:
        worker_index = random.randrange(num_workers)
        if worker_capacity > total_worker_capacity:
            if workers[worker_index][0] > 1:
                workers[worker_index] = (workers[worker_index][0] - 1, workers[worker_index][1])
                worker_capacity -= 1
        else:
            workers[worker_index] = (workers[worker_index][0] + 1, workers[worker_index][1])
            worker_capacity += 1

    while task_capacity != total_task_capacity:
        task_index = random.randrange(num_tasks)
        if task_capacity > total_task_capacity:
            if tasks[task_index][0] > 1:
                tasks[task_index] = (tasks[task_index][0] - 1, tasks[task_index][1])
                task_capacity -= 1
        else:
            tasks[task_index] = (tasks[task_index][0] + 1, tasks[task_index][1])
            task_capacity += 1

    return workers, tasks

# テストデータ生成
workers, tasks = generate_test_data(num_workers=8, num_tasks=8, num_no_preference_tasks=2, total_worker_capacity=24, total_task_capacity=24)

# マッチングの実行
result = mixed_acceptance_algorithm(workers, tasks)

# 結果の出力
print(result)