(1)やってみよう
以下のplotyを使うと一気に作ってくれる。
plotly.com
環境としては、plotlyのデフォルトのバージョンだと動かないので、以下のバージョンを指定して動かす。
!pip install plotly==4.9.0
以下は、上記のURLの通り。そのまま張り付けるとガントチャートが表示される。
import plotly.express as px
import pandas as pd
df = pd.DataFrame([
dict(Task="Job A", Start='2009-01-01', Finish='2009-02-28'),
dict(Task="Job B", Start='2009-03-05', Finish='2009-04-15'),
dict(Task="Job C", Start='2009-02-20', Finish='2009-05-30')
])
fig = px.timeline(df, x_start="Start", x_end="Finish", y="Task")
fig.update_yaxes(autorange="reversed")
fig.show()

(2)タスクリストをもとにしたガントチャート
"""
プログラムの概要:タスクの前後関係を考慮しながら、最短作業時間を計算する
- タスクリスト(tasks)に、タスクおよび作業時間や前後関係、担当者が割り当てられている
- 1分ごとに、タスクリストの情報から、作業する内容を決めて、while でループさせる。
- 複数のタスクがある場合は上から順に実行する → クリティカルパスを選べない場合がある
- 各担当者が作業済かどうかのフラグを用意する。
- ループごとに、それぞれのタスクが実行可能か判断する。実行不可能な条件は以下。
- 完了済みである。(残り時間が0)
- 前工程が残っている。
- 担当者がすでに作業済み。
- 実行可能なタスクなら、次のふたつの処理を行う。
- タスクの残り時間を-1。
- タスクを担当したリソースの「作業したフラグ」を True にする。
- 実行可能なタスクが無ければ終了。
- そしてガントチャートを表示する。
"""
!pip install plotly==4.9.0
!pip list | grep plot
import plotly.express as px
import pandas as pd
tasks = [
['A',0,[],['CSIRT']],
['B',60,['A'],['CSIRT']],
['C',60,['B'],['CSIRT']],
['D',120,['B'],['CSIRT']],
['E',300,['B'],['CSIRT']],
['F',60,['C'],['Sales']],
['G',60,['F'],['Sales']],
['H',360,['D'],['Vender']],
['I',60,['E'],['CSIRT']],
['J',120, ['D', 'H'],['CSIRT']],
['K',60,['G'],['Sales']],
['L',120,['E', 'I', 'J', 'G', 'K'],['CSIRT','Sales']],
['M',120,['L'],['CSIRT']],
['N',60,['M', 'I'],['Exective']],
['O',120,['E', 'J', 'N'],['CSIRT']],
]
total_time = 0
resources = []
for task in tasks:
for resource in task[3]:
resources.append(resource)
resources = list(set(resources))
for task in tasks:
if task[1] == 0:
for _task in tasks:
if task[0] in _task[2]:
_task[2].remove(task[0])
gantt_source = []
i = 0
while 1:
worked_flag = {}
for resource in resources:
worked_flag[resource] = False
for task in tasks:
if task[1] == 0:
continue
if task[2]:
continue
someone_is_busy = False
for resource in resources:
if resource in task[3] and worked_flag[resource] is True:
someone_is_busy = True
break
if someone_is_busy:
continue
task[1] -= 1
gantt_source.append(
dict(
Task=task[0],
Start=total_time,
Finish=total_time + 1,
Resource=' & '.join(task[3])),
)
for resource in resources:
if resource in task[3]:
worked_flag[resource] = True
for task in tasks:
if task[1] == 0:
for _task in tasks:
if task[0] in _task[2]:
_task[2].remove(task[0])
for resource in resources:
if worked_flag[resource]:
total_time += 1
break
else:
break
if i == 10000:
break
i += 1
df = pd.DataFrame(gantt_source)
df['delta'] = df['Finish'] - df['Start']
fig = px.timeline(df, x_start='Start', x_end='Finish', y='Task')
fig.layout.xaxis.type = 'linear'
fig.data[0].x = df.delta.tolist()
fig.update_yaxes(autorange='reversed')
fig.show()
(3)生産性なども考慮したガントチャート
!pip install plotly==4.9.0
!pip list | grep plot
from google.colab import files
uploaded = files.upload()
import csv
csvfile='tasklist.csv'
with open(csvfile, encoding='shift_jis') as file1:
tasks = list(csv.reader(file1))
print(tasks)
for _task in range(len(tasks)):
tasks[_task][1] = int(tasks[_task][1])
tasks[_task][2] = tasks[_task][2].split(',')
tasks[_task][3] = tasks[_task][3].split(',')
print(tasks)
"""
プログラムの概要:タスクの前後関係を考慮しながら、最短作業時間を計算する
- タスクリスト(tasks)に、タスクおよび作業時間や前後関係、担当者が割り当てられている
- 1分ごとに、タスクリストの情報から、作業する内容を決めて、while でループさせる。
- 複数のタスクがある場合は上から順に実行する → クリティカルパスを選べない場合がある
- 各担当者が作業済かどうかのフラグを用意する。
- ループごとに、それぞれのタスクが実行可能か判断する。実行不可能な条件は以下。
- 完了済みである。(残り時間が0)
- 前工程が残っている。
- 担当者がすでに作業済み。
- 実行可能なタスクなら、次のふたつの処理を行う。
- タスクの残り時間を-1。
- タスクを担当したリソースの「作業したフラグ」を True にする。
- 実行可能なタスクが無ければ終了。
- そしてガントチャートを表示する。
"""
import datetime
import pandas as pd
import plotly.express as px
tasks = [
['A',0,[],['CSIRT']],
['B',60,['A'],['CSIRT']],
['C',60,['B'],['CSIRT']],
['D',120,['B'],['CSIRT']],
['E',300,['B'],['CSIRT']],
['F',60,['C'],['Sales']],
['G',60,['F'],['Sales']],
['H',360,['D'],['Vender']],
['I',60,['E'],['CSIRT']],
['J',120, ['D', 'H'],['CSIRT']],
['K',60,['G'],['Sales']],
['L',120,['E', 'I', 'J', 'G', 'K'],['CSIRT','Sales']],
['M',120,['L'],['CSIRT']],
['N',60,['M', 'I'],['Exective']],
['O',120,['E', 'J', 'N'],['CSIRT']],
]
"""
tasks = [
['A', 10, [], ['CSIRT']], # noqa: E241 <- flake8 を無視する設定です。
['B', 5, [], ['Sales']], # noqa: E241
['C', 15, ['A', 'B'], ['CSIRT', 'Sales', 'Management']],
['D', 10, [], ['Sales']], # noqa: E241
['E', 20, ['D'], ['Vender']], # noqa: E241
]
"""
total_time = 0
resources = []
for task in tasks:
for resource in task[3]:
resources.append(resource)
resources = list(set(resources))
productivity = {
'CSIRT': 2,
}
for resource in resources:
productivity.setdefault(resource, 1)
for task in tasks:
if task[1] == 0:
for _task in tasks:
if task[0] in _task[2]:
_task[2].remove(task[0])
gantt_source = []
i = 0
while 1:
worked_flag = {}
for resource in resources:
worked_flag[resource] = [False] * productivity[resource]
for task in tasks:
if task[1] == 0:
continue
if task[2]:
continue
someone_is_busy = False
for resource in resources:
if resource in task[3] and all(worked_flag[resource]):
someone_is_busy = True
break
if someone_is_busy:
continue
remaining_power = 1
NOTE
if len(task[3]) == 1:
resources_for_this_task = task[3]
resource_name_for_this_task = resources_for_this_task[0]
remaining_power = worked_flag[resource_name_for_this_task].count(False)
for _ in range(remaining_power):
if task[1] > 0:
task[1] -= 1
else:
break
for resource in task[3]:
worked_flag[resource][worked_flag[resource].index(False)] = True
gantt_source.append(
dict(
Task=task[0],
Start=total_time,
Finish=total_time + 1,
Resource=' & '.join(task[3])),
)
for task in tasks:
if task[1] == 0:
for _task in tasks:
if task[0] in _task[2]:
_task[2].remove(task[0])
for resource in resources:
if any(worked_flag[resource]):
total_time += 1
break
else:
break
if i == 10000:
break
i += 1
gantt_source_2 = []
for achievement in gantt_source:
connected = False
for _achievement in gantt_source_2:
if _achievement['Task'] != achievement['Task']:
continue
if _achievement['Finish'] == achievement['Start']:
_achievement['Finish'] = achievement['Finish']
connected = True
break
if not connected:
gantt_source_2.append(achievement)
print(gantt_source_2)
for row in gantt_source_2:
row['Delta'] = row['Finish'] - row['Start']
base_date = datetime.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
for row in gantt_source_2:
row['Start'] = base_date + datetime.timedelta(minutes=row['Start'])
row['Finish'] = base_date + datetime.timedelta(minutes=row['Finish'])
row['Start'] = row['Start'].strftime('%Y-%m-%d %H:%M')
row['Finish'] = row['Finish'].strftime('%Y-%m-%d %H:%M')
df = pd.DataFrame(gantt_source_2)
print(df)
fig = px.timeline(
df,
x_start='Start',
x_end='Finish',
y='Task',
color='Resource',
text='Delta',
)
fig.update_yaxes(autorange='reversed')
fig.show()
print(total_time)