Pythonで同時に進行しているタスクの数を求めるプログラム

問題の概要

各要素が [start, end) の形式で表される区間のリスト intervals と、文字列のリスト types が与えられているとします。ある添字 i について、intervals[i] は「誰かが types[i] の種類の作業を時間帯 [start, end) に行っていた」ことを示します。なお、同じ種類の作業に対応する2つの区間が重なることや接触することはありません。

このとき、[start, end, num_types] という形式の項目からなるソート済みのマージ済みリストを作成してください。各項目は「start から end までの期間に、num_types 個のタスクが同時に進行していた」ことを表します。

たとえば、入力が次の場合を考えます。

intervals = [[0, 3], [5, 7], [0, 7]]
types = ["problem solving", "news", "game play"]

このときの出力は [[0, 3, 2], [3, 5, 1], [5, 7, 2]] となります。理由は以下の通りです。

• [0, 3) の間:「problem solving」と「game play」の2件が進行中
• [3, 5) の間:「game play」の1件のみが進行中
• [5, 7) の間:「news」と「game play」の2件が進行中

アルゴリズム(イベントソート方式)

この問題は、区間の始点と終点を「イベント」として扱い、時系列順に処理していくことで効率的に解けます。手順は以下の通りです。

1. 空のイベントリスト ev を用意します。
2. intervals 内の各区間 (s, e) について、開始イベント (s, +1) と終了イベント (e, −1) を ev に追加します。
3. ev を時刻順にソートします。
4. カウンタ cnt を 0、直前の時刻 last を −1 で初期化します。
5. 結果格納用の空リスト ans を用意します。
6. ev 内の各イベント (t, inc) を先頭から順に処理します。
   • t が last と異なり、かつ cnt が 0 でなければ、[last, t, cnt] を ans に追加します。
   • cnt に inc を加算し、last を t に更新します。
7. ans を返します。

この方法では、すべてのイベントを一度ソートして走査するだけなので、計算量は O(n log n)(n はイベント数)で抑えられます。

実装例(Python)

class Solution:
    def solve(self, intervals, jobs):
        ev = []
        for s, e in intervals:
            ev.append((s, 1))
            ev.append((e, -1))
        ev.sort()
        cnt = 0
        last = -1
        ans = []
        for t, inc in ev:
            if t != last and cnt != 0:
                ans.append([last, t, cnt])
            cnt += inc
            last = t
        return ans

ob = Solution()
intervals = [
    [0, 3],
    [5, 7],
    [0, 7]
]
types = ["problem solving", "news", "game play"]
print(ob.solve(intervals, types))

入力

[[0, 3],[5, 7],[0, 7]], ["problem solving", "news", "game play"]

出力

[[0, 3, 2], [3, 5, 1], [5, 7, 2]]

まとめ

本記事では、複数の作業区間から「各時点で何個のタスクが同時に進行していたか」を求める方法を紹介しました。開始時に +1、終了時に −1 のイベントを発生させ、時系列順に集計するシンプルな手法により、重なり合う区間の同時実行数を効率よく算出できます。スケジューリングやリソース使用状況の分析など、さまざまな場面で応用できるテクニックなので、ぜひ習得しておきましょう。