はじめに
下記の記事のシフトスケジューリング問題を、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の機能を使って、シンプルに記述できます。
以上