「クラスタ平均法を組み込んだ遺伝的アルゴリズムによるジョブショップスケジューリング問題の解法」をPython DEAPで実装した 2/5

はじめに

平野廣美氏による1995年の論文クラスタ平均法を組み込んだ遺伝的アルゴリズムによるジョブショップスケジューリング問題の解法に記載の方法をPythonのDEAPモジュールを使用して実装した。

今回実装を5章に分けて紹介する。

今回実装を5章に分けて紹介する。この記事は第2章である。

1. ジョブショップスケジューリング問題について
2. ジョブショップスケジューリング問題の解表現（←この記事）
3. ジョブショップスケジューリング問題向きの遺伝的操作
4. DEAPライブラリ
5. メイン処理、実行結果とまとめ

なおソースコード全体は以下に置いてある。

なお、論文補足のため同氏による2000年初版の書籍 遺伝的アルゴリズムと遺伝的プログラミング オブジェクト指向フレームワークによる構成と応用 での記述および付属ソースコードを参考にしている。ただしクラス構成などはオリジナルとは異なる。オリジナルは洗練されており参照してほしい。

ジョブショップスケジューリング問題の解表現

機械単位でのガントチャート

ジョブショップスケジューリング問題の場合最終的に欲しいのはガントチャートである。ガントチャートは、ジョブ単位で記述する場合や加工機械単位で記述する場合がある。

ジョブ単位のガントチャートは縦軸をジョブ番号、横軸を時間、セル値を機械番号とする表であり、機械単位でのガントチャートはは縦軸を機械番号、横軸を時間、セル値をジョブ番号とする表である。どちらも内容は同じであるが、ここでは機械単位でのガントチャートを念頭に置く。

-	0時	1時	2時	3時	4時
ジョブ1	機械1	機械1	機械2	機械2	機械3
ジョブ2				機械3
ジョブ3	機械3	機械3	機械3		機械1

Table: ジョブ単位でのガントチャートの例

-	0時	1時	2時	3時	4時
機械1	ジョブ1	ジョブ1			ジョブ3
機械2			ジョブ1	ジョブ1
機械3	ジョブ3	ジョブ3	ジョブ3	ジョブ2	ジョブ1

Table: 機械単位でのガントチャートの例

作業順序に着目した解の表現

平野氏は、乱数での改変に耐えられるよう、機械単位のガントチャートをジョブごとの作業順序に着目して [ 1, 3, 1, ... ] のような１次元の配列だけで表現することにした。そして実際の評価の際に処理時間も含めたガントチャートに展開する方法を取った。

例えば上述した3ジョブ4機械の場合について考えてみる。ジョブと加工機械、処理時間の関係を以下に再掲する。

-	工程1	工程2	工程3	工程4
ジョブ1	( 1, 3 )	( 2, 2 )	( 3, 4 )	( 4, 1 )
ジョブ2	( 2, 2 )	( 4, 2 )	( 3, 1 )	( 1, 3 )
ジョブ3	( 1, 2 )	( 3, 2 )	( 2, 1 )	( 4, 2 )

Table: 3ジョブ4機械の問題（再掲）

そして解の候補を[ 1, 3, 1, ..., 1 ]という1次元の配列で表現したとする。このとき配列の各要素はジョブ番号を表している。ジョブ番号はその工程数だけ配列内に重複して出現する。この例ではどのジョブも4工程あるのでジョブ番号は4つずつ出現することになる。1回目の出現は工程1の作業、2回目の出現は工程2の作業を表している。次の表にも記載した。

解候補のindex	解候補の値	説明
0	1	ジョブ1の最初の工程
1	3	ジョブ3の最初の工程
2	1	ジョブ1の2番目の工程
...	...	....
11	1	ジョブ1の最後の工程

Table: 解候補の意味

今回の問題はどのジョブも必要な工程数が同じであるので、これよりこの解候補の配列長は、常にジョブ数 x 工程数となる。

たとえばランダムな解候補は次のように生成できる。

import random

# listへの積は要素の繰り返しとなる。[ 0, 1, 2 ] * 2 = [ 0, 1, 2, 0, 1, 2 ]
individual = list ( range ( ジョブ数 ) ) * 工程数
random.shuffle ( individual )

なお解候補のことを個体（individual）と呼ぶ。

解表現からガントチャート（機械単位）への展開

前節で紹介した解候補はガントチャートに展開してはじめて評価できる。今回はすべてのジョブが完了する時刻、つまり最大完了時刻（メイクスパン）を評価対象とする。メイクスパンが小さければ小さいほどよい解となる。

解候補（個体）からgetGantt()のような手順でガントチャート（機械単位）を取得する。
この関数の引数jmTableは前章で説明したJobMachineTableBaseを継承した問題オブジェクトであり、もう一つの引数individualは解候補でジョブ番号のリストである。

getGantt()

def getGantt ( jmTable, individual ) :
	""" 個体からガントチャートを取得する """

この関数では、ganttリストにガントチャート（機械単位）を構築していく。ganttリストは機械ごとの作業スケジュールを格納する。このスケジュールには、次のように【開始時刻、終了時刻、ジョブ番号】の3つ組で表す作業のリストを格納する。
なおスケジュール内の作業は開始時刻の昇順に並んでいる。

-	0	1	2	...
機械1	[ 0, 0, None ]	最初の作業の[ 開始時刻, 終了時刻, ジョブ番号 ]	次の作業の [ 開始時刻, 終了時刻, ジョブ番号 ]	...
機械2	[ 0, 0, None ]	最初の作業の[ 開始時刻, 終了時刻, ジョブ番号 ]	...	...
...	[ 0, 0, None ]	最初の作業の[ 開始時刻, 終了時刻, ジョブ番号 ]	...	...

Table: ganttリストデータ構造

なお、ガントチャート作成の処理が簡単になるよう初期値としてダミーの作業[ 0, 0, None]および[ sys.max, sys.max, None]をセットしておく。このことで追加する作業は必ずダミーの作業の隙間に収まるこがわかる。

getGantt()

	# 機械数を問題オブジェクトから取得
	MAX_MACHINES = jmTable.getMachinesCount()
	# gantt [ MACHINE NUMBER ] = [ [0,0,None], [start, end, job_num], ...]
	# startの昇順に並ぶ、初期値にダミーの作業をセットしておく
	# 非稼働日があるならここでセットする
	gantt = [ [[0, 0, None], [sys.maxsize, sys.maxsize, None] ] for _ in range ( MAX_MACHINES ) ]

そして、個体individualの値（ジョブ番号）を順次とりだしてガントチャートを作成する。

getGantt()

	jmChild = jmTable.getChild()
	for job_num in individual :
		# ganttの適切な場所に作業を追加する

なおjmTable.getChild()は次の節で説明する。

ジョブの状態管理用クラスの追加

個体から取り出したジョブ番号をガントチャートに展開する際、そのジョブ番号が何番目であるかを管理する必要がある。
この状態管理用のオブジェクトをJobMachineTableオブジェクトからJobMachineChildオブジェクトとして生成する。

こうすることで（Pythonではほぼ関係ないが）マルチスレッド処理を行う際のオブジェクト管理が楽になることを期待している。

JobMachineTableBaseクラスにジョブの状態管理用のオブジェクトを生成するメソッドを追加する。

class JobMachineTableBase

class JobMachineTableBase :
	# ... 中略 ...
	def getChild ( self ) :
		return JobMachineChild ( self )

また、ジョブの状態管理用のクラスJobMachineChildを以下のように定義する。
インスタンス生成時にジョブごとに次の工程番号を格納する配列と、次工程を開始できる時刻の配列を作成する。

class JobMachineChild

class JobMachineChild :
	def __init__ ( self, jmParent ) :
		self._jmParent = jmParent
		# ジョブごとの次工程番号を格納する配列
		self._next_process_indexes = [ 0 ] * jmParent.getJobsCount()
		# ジョブごとの次工程が開始できる時刻の配列
		self._next_process_starts = [ 0 ] * jmParent.getJobsCount()

ジョブごとの次工程番号や次工程が開始できる時刻のゲッターやセッターも定義しまとめると次のようになる。

JobMachineTable.py

class JobMachineChild :
	""" ジョブの状態を管理するクラス """
	def __init__ ( self, jmParent ) :
		self._jmParent = jmParent
		# ジョブごとの次工程番号を格納する配列
		self._next_process_indexes = [ 0 ] * jmParent.getJobsCount()
		# ジョブごとの次工程が開始できる時刻の配列
		self._next_process_starts = [ 0 ] * jmParent.getJobsCount()
	def getMachine ( self, job_num ) :
		""" job_numのジョブの次工程の機械番号を取得 """
		process_index = self._next_process_indexes [ job_num ]
		return self._jmParent.getMachine ( job_num, process_index )
	def getProcessTime( self, job_num ) :
		""" job_numのジョブの次工程の処理時間を取得 """
		process_index = self._next_process_indexes [ job_num ]
		return self._jmParent.getProcessTime ( job_num, process_index )
	def getEarliest ( self, job_num ) :
		""" job_numジョブの最も早い次工程の開始時刻を取得; この時間より後に開始できない"""
		return self._next_process_starts [ job_num ]
	def setNextEarliest ( self, job_num, job_end ) :
		""" job_numジョブを次工程に進める。次工程は現工程の終了時刻以降に開始できる """
		self._next_process_starts [ job_num ] = job_end
		self._next_process_indexes [ job_num ] += 1

このJobMachineTableBase.getChild()で取得したJobMachineChildオブジェクトを使ってガントチャートを作成する。
JobMachineTableモジュールの全体は以下にある。

スケジュールに工程をセット

前節の前のコード（以下再掲）の続きを説明する。

getGantt()

	jmChild = jmTable.getChild()
	for job_num in individual :
		# ganttの適切な場所に作業を追加する

まずindividualから取り出した今回のジョブ番号job_numの作業内容（machine, process_time, job_earliest）を前節で導入したJobMachineChildオブジェクトを使って取得する。

getGantt()

		# job_numジョブのこの工程の(Machine番号, 処理時間)を取得
		machine = jmChild.getMachine ( job_num )
		process_time = jmChild.getProcessTime ( job_num ) 
		# このジョブの最も早い開始時刻を取得
		job_earliest = jmChild.getEarliest ( job_num ) 

そしてmachine番号の機械のスケジュールを確認する。このジョブの前の工程の終了時刻以降（job_earliest）以降でprocess_timeだけの連続した隙間を確認する。ダミージョブを初期値に設定しているので必ず隙間がみつかる。

gantt [ machine ]がこの機械のスケジュールとなる。知りたいのはスケジュールの隙間なので1つ目の作業の終了時刻と2つ目の作業の開始時刻とを取得する。
gantt [ machine ][ : -1 ]が1つ目の作業のリストで最初から最後-1までの作業を表している。gantt [ machine ][ 1 : ]が2つ目の作業リストで2番目から最後までの作業を表している。
これらをzip ( gantt [ machine ][:-1], gantt [ machine ][1:] )のようにzipでまとめて（1番目の作業, 2番目の作業）, (2番目の作業, 3番目の作業)というように、順次取得する。

取得した作業には開始時刻、終了時刻、ジョブ番号を格納している。これらをst0, ed0, _（ジョブ番号はここでは使わない）およびst1, ed1, _の各変数で受ける。作業1の終了時刻ed0が隙間の開始時刻gap_stであり、作業2の開始時刻st1が隙間の終了時刻gap_edである。

getGantt()

		# 左シフトで挿入できる隙間をさがす
		for idx, ( ( st0, ed0, _ ), ( st1, ed1, _ ) ) \
			in enumerate ( zip ( gantt [ machine ][:-1], gantt [ machine ][1:] ) ) :
			gap_st, gap_ed = ed0, st1

あとは隙間の大きさと今回の工程の処理時間、最早開始時刻を考慮して左シフト可能ならばこの作業を挿入する。

getGantt()

			# 隙間終了時刻でも最早時刻に満たない スキップ
			if gap_ed <= job_earliest : continue
			# 最早時刻が隙間の途中にあるとき 隙間の開始時刻を最早時刻にする
			gap_st = job_earliest if gap_st < job_earliest else ed0 
			# 隙間にこの処理が入らない スキップ
			if ( gap_ed - gap_st ) < process_time : continue
			# 隙間にこの処理が入る; スケジュールにこの工程を挿入
			job_end = gap_st + process_time
			gantt [ machine ].insert ( idx + 1, [ gap_st, job_end, job_num ] )
			break

最後にジョブの状態を更新して、次のジョブ番号の処理に移る

getGantt()

		jmChild.setNextEarliest ( job_num, job_end )

これらの処理を最後まで繰り返してガントチャートを得ることができる。
念のためガントチャートからダミーの作業を削除しておく。

getGantt()

	# 最初と最後のダミー作業を削除
	gantt = [ row[1:-1] for row in gantt ]

また、機械別のスケジュールの末尾の作業終了時刻を確認すればメイクスパンを取得できる。
なお、メイクスパンは小さいほど良い個体であるが、大きいほど良い個体とするためにその逆数を適応度としている。

eval()

def eval ( jmTable, individual ) :
	""" individualの適応度を取得する """
	# individualからガントチャートを取得する
	gantt = getGantt ( jmTable, individual )
	# 最後の作業終了時刻からメイクスパンを取得する
	makespan = 0
	for row in gantt :
		# rowは [ start, end, job_num ]のリスト
		makespan = max ( makespan, row [ -1 ][ 1 ] )
	return 1/makespan,

個体からガントチャートを取得する関数のソース

ここでの関数を以下にまとめる。

schedule.py

def getGantt ( jmTable, individual ) :
	"""
	個体からガントチャートを取得する
	@param	jmTable	job x machine num, process time table
	@param	individual
	"""
	MAX_MACHINES = jmTable.getMachinesCount()
	# gantt [ MACHINE NUMBER ] = [ [0,0,None], [start, end, job_num], ...]
	# startの昇順に並ぶ、初期値にダミーの作業をセットしておく
	# 非稼働日があるならここでセットする
	gantt = [ [[0, 0, None], [sys.maxsize, sys.maxsize, None]] for _ in range ( MAX_MACHINES ) ]
	jmChild = jmTable.getChild()
	for job_num in individual :
		# job_numジョブのこの工程の(Machine番号, 処理時間)を取得
		machine = jmChild.getMachine ( job_num )
		process_time = jmChild.getProcessTime ( job_num )
		# このジョブの最も早い開始時刻を取得
		job_earliest = jmChild.getEarliest ( job_num )
		#print ( job_num, machine, process_time )
		# 左シフトで挿入できる隙間をさがす
		for idx, ( ( st0, ed0, _ ), ( st1, ed1, _ ) ) \
			in enumerate ( zip ( gantt [ machine ][:-1], gantt [ machine ][1:] ) ) :
			gap_st, gap_ed = ed0, st1
			# 隙間終了時刻でも最早時刻に満たない スキップ
			if gap_ed <= job_earliest : continue
			# 最早時刻が隙間の途中にあるとき 隙間の開始時刻を最早時刻にする
			gap_st = job_earliest if gap_st < job_earliest else ed0
			# 隙間にこの処理が入らない スキップ
			if ( gap_ed - gap_st ) < process_time : continue
			# 隙間にこの処理が入る; スケジュールにこの工程を挿入
			job_end = gap_st + process_time
			gantt [ machine ].insert ( idx + 1, [ gap_st, job_end, job_num ] )
			break
		jmChild.setNextEarliest ( job_num, job_end )
	# 最初と最後のダミー作業を削除
	gantt = [ row[1:-1] for row in gantt ]
	return gantt

def eval ( jmTable, individual ) :
	""" individualの適応度を取得する """
	# individualからガントチャートを取得する
	gantt = getGantt ( jmTable, individual )
	# 最後の作業終了時刻からメイクスパンを取得する
	makespan = 0
	for row in gantt :
		# rowは [ start, end, job_num ]のリスト
		makespan = max ( makespan, row [ -1 ][ 1 ] )
	return 1/makespan,

なお次章の内容も含まれるが以下にこのコードを含むscheduleモジュールがある。

例

例えば、解候補が [2, 3, 1, 2, 1, 2, 3, 2, 1, 1, 3, 3] である場合の加工機械の作業スケジュールは以下のようになる。
ただし、縦軸が機械1, 2, 3, 4で、横軸が時間、セルの数字をジョブ番号とする。

-	0時	1時	2時	3時	4時	5時	6時	7時	8時	9時	10時	11時
機械1	3	3	1	1	1	2	2	2
機械2	2	2			3	1	1
機械3			3	3	2			1	1	1	1
機械4			2	2		3	3					1

1. 解候補の最初の値は2である。ジョブ2の工程1は機械2に2時間であるので、機械2の最初に2時間分スケジュールする。
2. 次の解候補の値は3である。ジョブ3の工程1は機械1に2時間であるので、機械1の最初に2時間分スケジュールする。
3. 次の解候補の値は1である。ジョブ1の工程1は機械1に3時間である。機械1の最初2時間はジョブ3で使用済みのため、その次に3時間分スケジュールする。
4. 以降同様に処理しスケジュールを完成する。
5. この場合の最大完了時刻は12時である。

（つづく）

---

次の章「ジョブショップスケジューリング問題向きの遺伝的操作」はこちら

前の章「ジョブショップスケジューリング問題について」はこちら