Explainable-Workforce-Scheduling/generator.py at master · ML-KULeuven/Explainable-Workforce-Scheduling · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

import random
import math
import pandas as pd

class DisruptionScenario:

    FEW_DISRUPTIONS = "few_disruptions"
    LONG_DISRUPTIONS = "long_disruptions"
    MANY_DISRUPTIONS = "many_disruptions"
    MANY_TEAMS_AFFECTED = "many_teams_affected"
    ALL_TEAMS_AFFECTED = "all_teams_affected"
    ONE_TEAM_DROPPED = "one_team_dropped"
    TWO_TEAMS_DROPPED = "two_teams_dropped"
    DELAY = "delay"


def generate_from_scenario(category, horizon, teams, seed=0):

    if category == DisruptionScenario.FEW_DISRUPTIONS:
        params = dict(
            num_disruptions=list(range(1,3)),
            num_teams_affected = (1, math.ceil(len(teams)/2)),
            duration = [15,30,60,120]
        )

    elif category == DisruptionScenario.LONG_DISRUPTIONS:
        params = dict(
            num_disruptions=list(range(1, 4)),
            num_teams_affected = (1,1),
            duration = (120, 240, 360)
        )

    elif category == DisruptionScenario.MANY_DISRUPTIONS:
        params = dict(
            num_disruptions=list(range(5,11)),
            num_teams_affected = (1, math.ceil(len(teams))),
            duration = (15,30,60,120)
        )

    elif category == DisruptionScenario.MANY_TEAMS_AFFECTED:
        params = dict(
            num_disruptions=(1,1),
            num_teams_affected = (math.ceil(len(teams) / 2), len(teams)),
            duration = (15,30,60,120)
        )

    elif category == DisruptionScenario.ALL_TEAMS_AFFECTED:
        params = dict(
            num_disruptions=(1,1),
            num_teams_affected = len(teams),
            duration = (15,30,60,120)
        )

    elif category == DisruptionScenario.ONE_TEAM_DROPPED:
        params = dict(
            num_disruptions=list(range(1, 4)),
            num_teams_affected = (1,1),
            duration = horizon
        )

    elif category == DisruptionScenario.TWO_TEAMS_DROPPED:
        params = dict(
            num_disruptions=range(1, 4),
            num_teams_affected = (2,2),
            duration = horizon
        )

    # TODO: add delay scenario

    else:
        raise ValueError(f"Invalid disruption category: {category}")

    return generate_disruption(**params,horizon=horizon,teams=teams,seed=seed)


def generate_disruption(num_disruptions, num_teams_affected, duration, horizon, teams, seed=0):

    random.seed(seed)

    disruptions = []
    for _ in range(random.randint(*num_disruptions)):
        dur = random.choice(duration)
        start = random.randint(0, horizon - dur)
        end = start + dur

        teams_affected = random.sample(teams, random.randint(*num_teams_affected))

        for team in teams_affected:
            disruptions.append(dict(
                team_id = team,
                start_unavailable = start,
                end_unavailable = end
            ))

    return pd.DataFrame(disruptions)