6

この人たちと一緒に私を助けてくれることを願っています。それは仕事を助けるものではありません。非常に勤勉なボランティアの慈善団体のためのものであり、彼らは現在持っているものよりも混乱や煩わしさの少ないタイムテーブルシステムを実際に使用することができます.

これを(確かに)自動化する優れたサードパーティ製アプリを誰かが知っていれば、それはほぼ同じです. ただ...教室を予約するためのようなランダムな時間割を提案しないでください.彼らはこれを行うことができないと思います.

読んでいただきありがとうございます。私はそれが大きな投稿であることを知っています。ただし、これを明確に文書化し、自分で努力したことを示すために最善を尽くしています.

問題

次の基準を満たすワーカーのシフトを生成するワーカー/タイムスロット スケジューリング アルゴリズムが必要です。

入力データ

import datetime.datetime as dt

class DateRange:
    def __init__(self, start, end):
        self.start = start
        self.end   = end

class Shift:
    def __init__(self, range, min, max):
        self.range = range
        self.min_workers = min
        self.max_workers = max

tue_9th_10pm = dt(2009, 1, 9,   22, 0)
wed_10th_4am = dt(2009, 1, 10,   4, 0)
wed_10th_10am = dt(2009, 1, 10, 10, 0)

shift_1_times = Range(tue_9th_10pm, wed_10th_4am)
shift_2_times = Range(wed_10th_4am, wed_10th_10am)
shift_3_times = Range(wed_10th_10am, wed_10th_2pm)

shift_1 = Shift(shift_1_times, 2,3)  # allows 3, requires 2, but only 2 available
shift_2 = Shift(shift_2_times, 2,2)  # allows 2
shift_3 = Shift(shift_3_times, 2,3)  # allows 3, requires 2, 3 available

shifts = ( shift_1, shift_2, shift_3 )

joe_avail = [ shift_1, shift_2 ]
bob_avail = [ shift_1, shift_3 ]
sam_avail = [ shift_2 ]
amy_avail = [ shift_2 ]
ned_avail = [ shift_2, shift_3 ]
max_avail = [ shift_3 ]
jim_avail = [ shift_3 ]

joe = Worker('joe', joe_avail)
bob = Worker('bob', bob_avail)
sam = Worker('sam', sam_avail)
ned = Worker('ned', ned_avail)
max = Worker('max', max_avail)
amy = Worker('amy', amy_avail)
jim = Worker('jim', jim_avail)

workers = ( joe, bob, sam, ned, max, amy, jim )

処理

上記から、シフト労働者は、処理する 2 つの主要な入力変数です。

各シフトには、必要な労働者の最小数と最大数があります。シフトの最小要件を満たすことは成功に不可欠ですが、他のすべてが失敗した場合、手動で埋めるギャップのある勤務表は「エラー」よりも優れています:) 主なアルゴリズムの問​​題は、十分な場合に不必要なギャップがあってはならないということです労働者が利用可能です。

理想的には、シフトの最大数の労働者が満たされることですが、これは他の制約に比べて優先度が最も低いため、何かを与える必要がある場合は、これにする必要があります。

柔軟な制約

これらは少し柔軟性があり、「完璧な」ソリューションが見つからない場合は、境界を少し広げることができます。ただし、この柔軟性は、ランダムに悪用されるのではなく、最後の手段であるべきです。理想的には、柔軟性は「fudge_factor」変数などで構成可能です。

  • 2 つのシフトの間に最小時間間隔があります。したがって、たとえば、同じ日に 2 つのシフトをスケジュールするべきではありません。
  • 労働者が特定の期間 (たとえば、1 か月) に行うことができるシフトの最大数があります。
  • 1 か月に行うことができる特定のシフトの最大数があります (夜勤など)。

あると便利ですが、必須ではありません

上記を実行し、これらの一部またはすべてを含むアルゴリズムを考え出すことができれば、私は非常に感銘を受け、感謝します. これらのビットを個別に実行するアドオン スクリプトも素晴らしいでしょう。

  • 重複するシフト。たとえば、「フロントデスク」のシフトと「バックオフィス」のシフトが同時に発生するように指定できるとよいでしょう。これは、異なるシフト データを使用してプログラムを個別に呼び出すことで実行できますが、特定の期間内に複数のシフトに人をスケジュールすることに関する制約が失われる場合があります。

  • (グローバルではなく) 作業者ごとに指定可能な、作業者の最小再スケジュール期間。たとえば、Joe が過労を感じているか、個人的な問題に対処している場合、または初心者であり、他の作業者よりも少ない頻度でスケジュールすることをお勧めします。

  • 利用可能な労働者が適合しない場合に、最小シフト数を満たすためにスタッフを選択する自動/ランダム/公正な方法。

  • 突然のキャンセルに対処し、他のシフトを再調整せずにギャップを埋める方法。

出力テスト

おそらく、アルゴリズムはできるだけ多くの一致するソリューションを生成する必要があります。各ソリューションは次のようになります。

class Solution:
    def __init__(self, shifts_workers):
        """shifts_workers -- a dictionary of shift objects as keys, and a
        a lists of workers filling the shift as values."""

        assert isinstance(dict, shifts_workers)
        self.shifts_workers = shifts_workers

上記のデータが与えられた場合の個々のソリューションのテスト関数を次に示します。これは正しいと思いますが、それについてのピアレビューもいただければ幸いです。

def check_solution(solution):
  assert isinstance(Solution, solution)

  def shift_check(shift, workers, workers_allowed):
      assert isinstance(Shift, shift):
      assert isinstance(list, workers):
      assert isinstance(list, workers_allowed)

      num_workers = len(workers)
      assert num_workers >= shift.min_workers
      assert num_workers <= shift.max_workers

      for w in workers_allowed:
          assert w in workers

  shifts_workers = solution.shifts_workers

  # all shifts should be covered
  assert len(shifts_workers.keys()) == 3
  assert shift1 in shifts_workers.keys()
  assert shift2 in shifts_workers.keys()
  assert shift3 in shifts_workers.keys()

  # shift_1 should be covered by 2 people - joe, and bob
  shift_check(shift_1, shifts_workers[shift_1], (joe, bob))

  # shift_2 should be covered by 2 people - sam and amy
  shift_check(shift_2, shifts_workers[shift_2], (sam, amy))

  # shift_3 should be covered by 3 people - ned, max, and jim
  shift_check(shift_3, shifts_workers[shift_3], (ned,max,jim))

試み

これを遺伝的アルゴリズムで実装しようとしましたが、完全に調整できないようです。そのため、基本原理は単一シフトで機能するように見えますが、いくつかのシフトといくつかの簡単なケースでさえ解決できません。労働者。

私の最新の試みは、可能なすべての順列を解決策として生成し、次に制約を満たさない順列を絞り込むことです。これははるかに迅速に機能するようで、さらに理解を深めましたが、順列の生成を支援するために python 2.6 の itertools.product() を使用していますが、うまくいきません。正直なところ、問題は私の頭にうまく収まらないので、多くのバグがあっても驚かないでしょう:)

現在、このコードは、models.py と rota.py の 2 つのファイルにあります。models.py は次のようになります。

# -*- coding: utf-8 -*-
class Shift:
    def __init__(self, start_datetime, end_datetime, min_coverage, max_coverage):
        self.start = start_datetime
        self.end = end_datetime
        self.duration = self.end - self.start
        self.min_coverage = min_coverage
        self.max_coverage = max_coverage

    def __repr__(self):
        return "<Shift %s--%s (%r<x<%r)" % (self.start, self.end, self.min_coverage, self.max_coverage)

class Duty:
    def __init__(self, worker, shift, slot):
        self.worker = worker
        self.shift = shift
        self.slot = slot

    def __repr__(self):
        return "<Duty worker=%r shift=%r slot=%d>" % (self.worker, self.shift, self.slot)

    def dump(self,  indent=4,  depth=1):
        ind = " " * (indent * depth)
        print ind + "<Duty shift=%s slot=%s" % (self.shift, self.slot)
        self.worker.dump(indent=indent, depth=depth+1)
        print ind + ">"

class Avail:
    def __init__(self, start_time, end_time):
        self.start = start_time
        self.end = end_time

    def __repr__(self):
        return "<%s to %s>" % (self.start, self.end)

class Worker:
    def __init__(self, name, availabilities):
        self.name = name
        self.availabilities = availabilities

    def __repr__(self):
        return "<Worker %s Avail=%r>" % (self.name, self.availabilities)

    def dump(self,  indent=4,  depth=1):
        ind = " " * (indent * depth)
        print ind + "<Worker %s" % self.name
        for avail in self.availabilities:
            print ind + " " * indent + repr(avail)
        print ind + ">"

    def available_for_shift(self,  shift):
        for a in self.availabilities:
            if shift.start >= a.start and shift.end <= a.end:
                return True
        print "Worker %s not available for %r (Availability: %r)" % (self.name,  shift, self.availabilities)
        return False

class Solution:
    def __init__(self, shifts):
        self._shifts = list(shifts)

    def __repr__(self):
        return "<Solution: shifts=%r>" % self._shifts

    def duties(self):
        d = []
        for s in self._shifts:
            for x in s:
                yield x

    def shifts(self):
        return list(set([ d.shift for d in self.duties() ]))

    def dump_shift(self, s, indent=4, depth=1):
        ind = " " * (indent * depth)
        print ind + "<ShiftList"
        for duty in s:
            duty.dump(indent=indent, depth=depth+1)
        print ind + ">"

    def dump(self, indent=4, depth=1):
        ind = " " * (indent * depth)
        print ind + "<Solution"
        for s in self._shifts:
            self.dump_shift(s, indent=indent, depth=depth+1)
        print ind + ">"

class Env:
    def __init__(self, shifts, workers):
        self.shifts = shifts
        self.workers = workers
        self.fittest = None
        self.generation = 0

class DisplayContext:
    def __init__(self,  env):
        self.env = env

    def status(self, msg, *args):
        raise NotImplementedError()

    def cleanup(self):
        pass

    def update(self):
        pass

と rota.py は次のようになります。

#!/usr/bin/env python2.6
# -*- coding: utf-8 -*-

from datetime import datetime as dt
am2 = dt(2009,  10,  1,  2, 0)
am8 = dt(2009,  10,  1,  8, 0)
pm12 = dt(2009,  10,  1,  12, 0)

def duties_for_all_workers(shifts, workers):
    from models import Duty

    duties = []

    # for all shifts
    for shift in shifts:
        # for all slots
        for cov in range(shift.min_coverage, shift.max_coverage):
            for slot in range(cov):
                # for all workers
                for worker in workers:
                    # generate a duty
                    duty = Duty(worker, shift, slot+1)
                    duties.append(duty)

    return duties

def filter_duties_for_shift(duties,  shift):
    matching_duties = [ d for d in duties if d.shift == shift ]
    for m in matching_duties:
        yield m

def duty_permutations(shifts,  duties):
    from itertools import product

    # build a list of shifts
    shift_perms = []
    for shift in shifts:
        shift_duty_perms = []
        for slot in range(shift.max_coverage):
            slot_duties = [ d for d in duties if d.shift == shift and d.slot == (slot+1) ]
            shift_duty_perms.append(slot_duties)
        shift_perms.append(shift_duty_perms)

    all_perms = ( shift_perms,  shift_duty_perms )

    # generate all possible duties for all shifts
    perms = list(product(*shift_perms))
    return perms

def solutions_for_duty_permutations(permutations):
    from models import Solution
    res = []
    for duties in permutations:
        sol = Solution(duties)
        res.append(sol)
    return res

def find_clashing_duties(duty, duties):
    """Find duties for the same worker that are too close together"""
    from datetime import timedelta
    one_day = timedelta(days=1)
    one_day_before = duty.shift.start - one_day
    one_day_after = duty.shift.end + one_day
    for d in [ ds for ds in duties if ds.worker == duty.worker ]:
        # skip the duty we're considering, as it can't clash with itself
        if duty == d:
            continue

        clashes = False

        # check if dates are too close to another shift
        if d.shift.start >= one_day_before and d.shift.start <= one_day_after:
            clashes = True

        # check if slots collide with another shift
        if d.slot == duty.slot:
            clashes = True

        if clashes:
            yield d

def filter_unwanted_shifts(solutions):
    from models import Solution

    print "possibly unwanted:",  solutions
    new_solutions = []
    new_duties = []

    for sol in solutions:
        for duty in sol.duties():
            duty_ok = True

            if not duty.worker.available_for_shift(duty.shift):
                duty_ok = False

            if duty_ok:
                print "duty OK:"
                duty.dump(depth=1)
                new_duties.append(duty)
            else:
                print "duty **NOT** OK:"
                duty.dump(depth=1)

        shifts = set([ d.shift for d in new_duties ])
        shift_lists = []
        for s in shifts:
            shift_duties = [ d for d in new_duties if d.shift == s ]
            shift_lists.append(shift_duties)

        new_solutions.append(Solution(shift_lists))

    return new_solutions

def filter_clashing_duties(solutions):
    new_solutions = []

    for sol in solutions:
        solution_ok = True

        for duty in sol.duties():
            num_clashing_duties = len(set(find_clashing_duties(duty, sol.duties())))

            # check if many duties collide with this one (and thus we should delete this one
            if num_clashing_duties > 0:
                solution_ok = False
                break

        if solution_ok:
            new_solutions.append(sol)

    return new_solutions

def filter_incomplete_shifts(solutions):
    new_solutions = []

    shift_duty_count = {}

    for sol in solutions:
        solution_ok = True

        for shift in set([ duty.shift for duty in sol.duties() ]):
            shift_duties = [ d for d in sol.duties() if d.shift == shift ]
            num_workers = len(set([ d.worker for d in shift_duties ]))

            if num_workers < shift.min_coverage:
                solution_ok = False

        if solution_ok:
            new_solutions.append(sol)

    return new_solutions

def filter_solutions(solutions,  workers):
    # filter permutations ############################
    # for each solution
    solutions = filter_unwanted_shifts(solutions)
    solutions = filter_clashing_duties(solutions)
    solutions = filter_incomplete_shifts(solutions)
    return solutions

def prioritise_solutions(solutions):
    # TODO: not implemented!
    return solutions

    # prioritise solutions ############################
    # for all solutions
        # score according to number of staff on a duty
        # score according to male/female staff
        # score according to skill/background diversity
        # score according to when staff last on shift

    # sort all solutions by score

def solve_duties(shifts, duties,  workers):
    # ramify all possible duties #########################
    perms = duty_permutations(shifts,  duties)
    solutions = solutions_for_duty_permutations(perms)
    solutions = filter_solutions(solutions,  workers)
    solutions = prioritise_solutions(solutions)
    return solutions

def load_shifts():
    from models import Shift
    shifts = [
        Shift(am2, am8, 2, 3),
        Shift(am8, pm12, 2, 3),
    ]
    return shifts

def load_workers():
    from models import Avail, Worker
    joe_avail = ( Avail(am2,  am8), )
    sam_avail = ( Avail(am2,  am8), )
    ned_avail = ( Avail(am2,  am8), )
    bob_avail = ( Avail(am8,  pm12), )
    max_avail = ( Avail(am8,  pm12), )
    joe = Worker("joe", joe_avail)
    sam = Worker("sam", sam_avail)
    ned = Worker("ned", sam_avail)
    bob = Worker("bob", bob_avail)
    max = Worker("max", max_avail)
    return (joe, sam, ned, bob, max)

def main():
    import sys

    shifts = load_shifts()
    workers = load_workers()
    duties = duties_for_all_workers(shifts, workers)
    solutions = solve_duties(shifts, duties, workers)

    if len(solutions) == 0:
        print "Sorry, can't solve this.  Perhaps you need more staff available, or"
        print "simpler duty constraints?"
        sys.exit(20)
    else:
        print "Solved.  Solutions found:"
        for sol in solutions:
            sol.dump()

if __name__ == "__main__":
    main()

結果の前にデバッグ出力を切り取ると、現在次のようになります。

Solved.  Solutions found:
    <Solution
        <ShiftList
            <Duty shift=<Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2<x<3) slot=1
                <Worker joe
                    <2009-10-01 02:00:00 to 2009-10-01 08:00:00>
                >
            >
            <Duty shift=<Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2<x<3) slot=1
                <Worker sam
                    <2009-10-01 02:00:00 to 2009-10-01 08:00:00>
                >
            >
            <Duty shift=<Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2<x<3) slot=1
                <Worker ned
                    <2009-10-01 02:00:00 to 2009-10-01 08:00:00>
                >
            >
        >
        <ShiftList
            <Duty shift=<Shift 2009-10-01 08:00:00--2009-10-01 12:00:00 (2<x<3) slot=1
                <Worker bob
                    <2009-10-01 08:00:00 to 2009-10-01 12:00:00>
                >
            >
            <Duty shift=<Shift 2009-10-01 08:00:00--2009-10-01 12:00:00 (2<x<3) slot=1
                <Worker max
                    <2009-10-01 08:00:00 to 2009-10-01 12:00:00>
                >
            >
        >
    >
4

3 に答える 3

3

遺伝的アルゴリズムでこれを実装しようとしましたが、正しく調整できていないようです。そのため、基本原理は1シフトで機能するように見えますが、数シフトと数シフトの簡単なケースでも解決できません。労働者。

要するに、しないでください!遺伝的アルゴリズムの経験が豊富でない限り、これを正しく行うことはできません。

  • これらは、実行可能なソリューションへの収束を保証するものではない近似的な方法です。
  • これらは、現在のソリューションの品質を合理的に十分に確立できる場合(つまり、基準の数が満たされていない場合)にのみ機能します。
  • それらの品質は、以前のソリューションを新しいソリューションに結合/変更するために使用する演算子の品質に大きく依存します。

GAの経験がほとんどない場合、小さなPythonプログラムを正しく理解するのは難しいことです。少人数のグループの場合、徹底的な検索はそれほど悪い選択肢ではありません。n問題は、それが人々にとって正しく機能する可能性があり、n+1人々にとっては遅く、耐えられないほど遅くなるn+2可能性があることnです。

あなたはNP完全問題に取り組んでおり、簡単な勝利の解決策はありません。選択した空想の時刻表スケジューリング問題が十分に機能しない場合は、Pythonスクリプトでより良いものが得られる可能性はほとんどありません。

独自のコードを介してこれを行うことを主張する場合は、min-maxまたはシミュレーテッドアニーリングでいくつかの結果を得るのがはるかに簡単です。

于 2009-10-12T14:05:22.080 に答える
1

アルゴリズムの選択肢はありませんが、いくつかの実用的な考慮事項を関連付けることはできます。

アルゴリズムはキャンセルを処理するため、スケジューリングの例外が発生するたびに実行して、全員を再スケジュールする必要があります。

一部のアルゴリズムはあまり直線的ではなく、その時点からすべての人を根本的に再スケジュールする可能性があることを考慮してください。あなたはおそらくそれを避けたいと思うでしょう.人々は自分のスケジュールを前もって知りたいのです.

次に利用可能な人または 2 人を事前にスケジュールできるため、アルゴリズムを再実行せずに一部のキャンセルに対処できます。

常に最適なソリューションを生成することは不可能であり、望ましくない場合もありますが、ワーカーごとに「最適ではない」イベントの実行中のカウントを維持し、別の "下手な選択"。それが人々が一般的に気にかけていることです (いくつかの「悪い」スケジューリングの決定が頻繁に/不当に行われています)。

于 2009-10-12T15:23:55.310 に答える
1

わかりました、特定のアルゴリズムについてはわかりませんが、ここで考慮すべきことを示します。

評価

方法が何であれ、解がどれだけ制約を満たしているかを評価する関数が必要です。「比較」アプローチ (グローバル スコアではなく、2 つのソリューションを比較する方法) を使用することもできますが、評価をお勧めします。

より短い期間、たとえば毎日のスコアを取得できれば、本当に良いことです。部分的なソリューションから最終的なスコアの範囲を「予測」できれば、アルゴリズムで非常に役立ちます (たとえば、最初の 3 つだけ)。 7日のうち)。このようにして、この部分解が既に低すぎて期待に応えられない場合、この部分解に基づいて計算を中断できます。

対称

これらの 200 人の中で、類似したプロファイルを持っている可能性があります。つまり、同じ特性 (可用性、経験、意欲など) を共有している人々です。同じプロファイルを持つ 2 人の人物を取り上げると、交換可能になります。

  • 解決策 1: (シフト 1: ジョー)(シフト 2: ジム) ...他の従業員のみ...
  • 解決策 2: (シフト 1: ジム)(シフト 2: ジョー) ...他の従業員のみ...

実際には、あなたの観点からは同じソリューションです。

良いことは、通常、人よりもプロファイルが少ないことです。これにより、計算に費やす時間が大幅に短縮されます。

たとえば、ソリューション 1 に基づいてすべてのソリューションを生成した場合、ソリューション 2 に基づいて何も計算する必要がないと想像してください。

反復

スケジュール全体を一度に生成する代わりに、段階的に (一度に 1 週​​間など) 生成することを検討できます。正味の利益は、1 週間の複雑さが軽減されることです (可能性が少なくなります)。

次に、今週を取得したら、もちろん制約のために最初の週を考慮して最初の週を慎重に考慮しながら、2番目の週を計算します。

利点は、すでに使用されているソリューションを考慮してアルゴリズムを明示的に設計することです。このようにして、次のスケジュール生成では、人が 24 時間連続で働かないようにすることができます。

シリアル化

ソリューション オブジェクトのシリアル化を検討する必要があります (選択を選択してください。Python には pickle が適しています)。新しいスケジュールを生成するときは以前のスケジュールが必要になりますが、200 人の場合は手動で入力したくないでしょう。

網羅的

さて、結局のところ、対称性と評価を使用すると可能性がそれほど多くない可能性があるため、実際には徹底的な検索を好みます (問題は NP 完全のままですが、特効薬はありません)。

Backtracking Algorithmを試してみてください。

また、同様の問題を扱っている次のリンクも参照してください。

どちらも実装中に発生した問題について説明しているため、それらをチェックアウトすると役立つはずです。

于 2009-10-12T15:58:43.570 に答える