シフトスケジューリング問題を解いてみた

はじめに

下記の記事のシフトスケジューリング問題を、pandasを使った数理モデルで解いてみました。

元ネタは、下記の「Pythonによる実務で役立つ最適化問題100+ (3)
―配送計画・パッキング・スケジューリング―」の31章の問題です。

変数表

変数は変数表で管理します。変数表は2つあります。

変数表df_staff（スタッフ一覧）

列名	意味
Staff	スタッフ番号
VarS	スタッフを使うかどうか（0-1変数）

変数表df（スタッフとジョブの可能な組み合わせ）

列名	意味
Staff	スタッフ番号
Job	ジョブ番号
Start	ジョブの開始時刻
End	ジョブの終了時刻
Var	ジョブをスタッフに割り当てるかどうか（0-1変数）
VarS	スタッフを使うかどうか（df_staff.VarSの参照）

数理モデル

数理モデルは、元記事と一緒ですが、表現方法だけ変えています。

• 変数：
  • df.Var：ジョブをスタッフに割り当てるかどうか（0-1変数）
  • df_staff.VarS：スタッフを使うかどうか（0-1変数）
• 目的関数：スタッフ数（df_staff.VarSの和） → 最小化
• 制約条件：
  • ジョブごとに、いずれかのスタッフに割り当てる
  • スタッフごと、クリークごとに、対象のVarの和がVarS以下

※ クリークとは、1人のスタッフが同時にできないジョブの集合です。

データの取得

下記のOR-LibraryからZIPファイルをダウンロードして、解凍してください。

データの説明は、下記を参照してください。

pip install mip pandasで必要なライブラリをインストールしてください。

今回は、元記事と同じptask/data_1_23_40_66.datを使います。

データの読み込み

df_staffとdfとcliquesを作成します。cliquesは（極大）クリークのリストです。

from pathlib import Path
import pandas as pd
from mip import Model, minimize, xsum

def load(dat):
    lines = dat.read_text().splitlines()
    n_job = int(lines[4].split()[-1])
    n_staff = int(lines[n_job + 5].split()[-1])
    job_lines = lines[5:n_job+5]
    qualifications = lines[n_job+6:n_job+n_staff+6]
    df_job = pd.DataFrame(map(str.split, job_lines), columns=["Start", "End"]).astype(int)
    df_staff = pd.DataFrame(range(n_staff), columns=["Staff"])
    # 全クリークの計算
    sorted_start = df_job.Start.sort_values()
    sorted_end = df_job.End.sort_values()
    next_start = sorted_start[i_start := 0]
    next_end = sorted_end[i_end := 0]
    added = False  # 追加されたか
    clique = set()
    cliques = []  # 全クリーク
    while i_start < n_job or i_end < n_job:
        if next_start < next_end:
            added = True
            clique.add(sorted_start.index[i_start])
            i_start += 1
            next_start = sorted_start.iloc[i_start] if i_start < n_job else 1e99
            continue
        if added:
            cliques.append(set(clique))
        added = False
        clique.remove(sorted_end.index[i_end])
        i_end += 1
        next_end = sorted_end.iloc[i_end] if i_end < n_job else 1e99
    df = pd.DataFrame(
        ((i, int(j))
        for i, s in enumerate(qualifications)
        for j in s.split()[1:]),
        columns=["Staff", "Job"]
    )
    df = df.join(df_job, "Job")
    return df_staff, df, cliques

df_staff, df, cliques = load(Path("ptask/data_1_23_40_66.dat"))

モデルの作成と求解と結果表示

モデルを作成して、求解して、結果を表示します。結果は、スタッフと割り当てられたジョブです。元記事と同じ結果ではありませんが、スタッフ数は20なので最適解になっています。

m = Model()
df_staff["VarS"] = m.add_var_tensor((len(df_staff),), "VarS", var_type="B")
df["Var"] = m.add_var_tensor((len(df),), "Var", var_type="B")
df = df.merge(df_staff)
m.objective = minimize(xsum(df_staff.VarS))  # 目的関数
for _, gr in df.groupby("Job"):
    m += xsum(gr.Var) == 1  # 1つ目の制約条件
for _, gr in df.groupby("Staff"):
    for clique in cliques:
        sub = gr[gr.Job.isin(clique)]
        if not sub.empty:
            m += xsum(sub.Var) <= sub.VarS.iloc[0]  # 2つ目の制約条件
m.verbose = 0
m.optimize()
if m.status.value == 0:
    df["Val"] = df.Var.astype(float)
    print(f"スタッフの人数 = {m.objective_value}")
    print(df[df.Val> 0.5].groupby("Staff").Job.apply(list))

実行結果

スタッフの人数 = 20.0
Staff
0       [1, 4]
1     [14, 16]
2     [30, 36]
3      [7, 25]
4     [22, 32]
5      [37, 3]
6     [15, 31]
7      [8, 13]
9     [19, 28]
10     [34, 0]
11    [20, 35]
12     [5, 23]
13     [9, 27]
14    [21, 29]
17     [2, 33]
18    [12, 38]
19    [10, 17]
20    [18, 26]
21     [6, 39]
22    [11, 24]
Name: Job, dtype: object

解説

df.groupby("Job")とすることで、ジョブごとに繰り返しています。
同様に、df.groupby("Staff")とすることで、スタッフごとに繰り返しています。
また、gr[gr.Job.isin(clique)]とすることで、ジョブがクリークに含まれる行だけ抜き出しています。

このように変数表を使うことで、モデルが列名で表現されるので理解しやすくなります。
また、グループ化（groupby()）や絞り込み（isin()）のように、pandasの機能を使って、シンプルに記述できます。

以上