roster_planner.py

from ics import Calendar  # pip3 install ics
import json
from datetime import date, datetime, timedelta
import math
import random
import os
import itertools


def read_calendar(calendar_object, calendar):
    # Loop through events
    # sample:
    # Event: des medical appt
    # Start: 2025-09-11T00:00:00+00:00
    # End: 2025-09-12T00:00:00+00:00
    for event in calendar.events:
        print("Event:", event.name)
        print("Start:", event.begin)
        print("End:", event.end)
        print("-" * 40)


# If true it means the event is after today's date
def check_date(current_date, datetime_str):
    # format of datetime_str = 2025-09-11T00:00:00+00:00
    # Convert string to datetime object
    datetime_str = str(datetime_str).split("T")[0]
    datetime_object = datetime.strptime(str(datetime_str), "%Y-%m-%d")
    # Convert datetime to date
    date_only = datetime_object.date()
    if date_only < current_date:
        return False
    else:
        return True


def return_dates(start_date, end_date):
    start_str = str(start_date).split("T")[0]
    start = datetime.strptime(start_str, "%Y-%m-%d")
    end_str = str(end_date).split("T")[0]
    end = datetime.strptime(end_str, "%Y-%m-%d")
    dates = [start + timedelta(days=i) for i in range((end - start).days)]
    dates_str = []
    for date_entry in dates:
        date_value_str = date_entry.strftime("%Y-%m-%d")
        date_list = date_value_str.split("-")
        dates_str.append([int(date_list[0]), int(date_list[1]), int(date_list[2])])
    return dates_str


def create_year_month_entry(ooo_dict, person):
    year = date.today().year
    if person not in ooo_dict:
        ooo_dict[person] = {}
    for y in [year, year + 1]:
        if str(y) not in ooo_dict[person]:
            ooo_dict[person][str(y)] = {str(m): {} for m in range(1, 13)}
    return ooo_dict


# dates should be a nested list [[year], [month], [days]]
def update_ooo_dict(dates, person, ooo_dict):
    ooo_dict = create_year_month_entry(ooo_dict, person)
    for year, month, day in dates:
        year, month, day = str(year), str(month), str(day)
        if day not in ooo_dict[person][year][month]:
            ooo_dict[person][year][month][day] = 0
    return ooo_dict


def build_calendar_dict(people_dict, ooo_dict, calendar):
    now = date.today()
    for event in calendar.events:
        relevant_event = check_date(now, event.begin)
        if relevant_event:
            name_found = False
            for names in people_dict:
                if names in str(event.name).lower():
                    dates = return_dates(event.begin, event.end)
                    # calculate dates to put in ooo_dict
                    ooo_dict = update_ooo_dict(dates, people_dict[names], ooo_dict)
                    name_found = True
                    break
            if not name_found:
                print("no person indicated in: " + event.name)
    return ooo_dict


# TODO this algo needs to change
# this function runs when someone in their planned shift is unavailable
def roster_shuffle(order, people_to_swap, dates_list, people_ooo):
    while not people_to_swap:
        random.shuffle(order)
        people_to_swap = check_their_availability(order, dates_list, people_ooo)
    return order


# dates will be a list of [ [year, month, day], [year, month, day], ...]
def is_available(person, people_ooo, date):
    available = True
    # need to convert to string bc the json saves the values as a str
    year = str(date[0])
    month = str(date[1])
    day = str(date[2])
    if person in people_ooo:
        if year in people_ooo[person]:
            if month in people_ooo[person][year]:
                if day in people_ooo[person][year][month]:
                    available = False
    return available


# example usage:
# print(swap_with_nearest(items, 2, is_available))
# swaps item at index 2 ith nearest available (index 3)
def swap_with_nearest(order, target_index, people_ooo, dates):
    def check_whether_they_ooo(person):
        for date1 in dates:
            on_leave = is_available(person, people_ooo, date1)
            # print("person: " + str(person) + " " + str(on_leave))
            if not on_leave:  # if false means they are ooo on that date
                break
        return on_leave

    n = len(order)
    print("original order:" + str(order))
    # Check outward from the target index
    # you want to check the right most person first, then left most, then repeat with the 2nd right most, etc
    for offset in range(1, n):
        right = target_index + offset
        if right >= n:  # to prevent index error when the count exceeds the max number of items
            right -= n - 1
        # print("checking order[right]: " + str(order[right]))
        not_ooo_right = check_whether_they_ooo(order[right])
        if not_ooo_right:
            order[right], order[target_index] = order[target_index], order[right]
            break
        else:
            left = target_index - offset
            if left >= 0:
                not_ooo_left = check_whether_they_ooo(order[left])
                if not_ooo_left:
                    order[left], order[target_index] = order[target_index], order[left]
                    break

    print("new order: " + str(order))
    # No swap found
    return order


def reorder_subset(big_list, small_list):
    # Create a mapping from item -> its new index in small_list
    order_map = {item: i for i, item in enumerate(small_list)}

    # Extract items that are in small_list and sort them by the new order
    reordered = iter(sorted(
        (item for item in big_list if item in small_list),
        key=lambda x: order_map[x]
    ))

    # Rebuild big_list: replace small_list items with the next from reordered
    return [next(reordered) if item in small_list else item for item in big_list]


# people to swap should be a list
def swap_among_themselves(order, people_to_swap, people_ooo, dates_dict):
    # Generate all permutations
    perms = list(itertools.permutations(people_to_swap))
    # Exclude the original
    perms_to_try = [p for p in perms if list(p) != people_to_swap]
    all_dates = []
    new_order = []
    # generate the order of dates to try:
    for people in people_to_swap:
        all_dates.append[people_ooo[people]]
    for perms in perms_to_try:
        people_count = 0
        available = True
        for people_count in range(len(perms)):
            for dates in all_dates[people_count]:
                # TODO adjust this if there are optional and mandatory dates instead of all mandatory
                if str(dates[2]) in people_ooo[perms[people_count]][str(dates[0])][str(dates[1])]:
                    available = False
                    break
        if available:
            new_order = reorder_subset(order, perms)
    return new_order


# rostered_dict e.g. : {'Ysabella': {0: [[2025, 9, 8]] }, 'Melodee': {0: [[2025, 9, 15]]},
def roster_adjust(rostered_dict, order, people_ooo, people_to_swap):
    full_order = []
    for cycles in people_to_swap:
        if people_to_swap[cycles]:
            number_of_swaps = len(people_to_swap[cycles])
            print("people_to_swap: " + str(people_to_swap))
            list_of_people_to_swap = list(people_to_swap[cycles].keys())
            new_order = order  # this line shouldn't be necessary, only added it to get rid of the warning
            if number_of_swaps == 1:
                target_index = order.index(list_of_people_to_swap[0])
                dates = next(iter(people_to_swap[cycles].values()))[1]
                new_order = swap_with_nearest(order, target_index, people_ooo, dates)
            elif number_of_swaps >= 2:
                new_order = swap_among_themselves(order, list_of_people_to_swap, people_ooo, rostered_dict[cycles])
                if not new_order:
                    print("No solution found for swapping among themselves. Resolve this LOL")
                    exit()
            full_order.append(new_order)
        else:
            full_order.append(order)
    print(full_order)
    return full_order


def assign_dates_to_people(orders, dates_list, repeats):
    assigned_roster = {}
    for people in orders:
        assigned_roster[people] = {}
    for count in range(0, repeats):
        counter = count * len(orders)
        people_count = 0
        for people in orders:
            date_count = counter + people_count
            assigned_roster[people][count] = dates_list[date_count]
            people_count += 1
    return assigned_roster


# TODO can be optimised further, this is slow if there is one person who cannot make it
def check_their_availability_previous(order, dates_list, people_ooo):
    count = 0
    people_who_need_to_swap = []
    for people in order:
        if people not in people_ooo:
            # TODO investigate this, see if there any errors
            print(people + " is not in people_ooo")
        else:
            for dates in dates_list[count]:
                if str(dates[2]) in people_ooo[people][str(dates[0])][str(dates[1])]:
                    people_who_need_to_swap.append(people)
        count += 1
    return people_who_need_to_swap


def check_their_availability(rostered_dict, people_ooo, repeats):
    count = 0
    unavailable_people = {}
    # print("Rostered_dict in check_their_availability:" + str(rostered_dict))
    for count in range(repeats):
        unavailable_people[count] = {}

    for people in rostered_dict:
        if people not in people_ooo:
            # TODO investigate this, see if there any errors
            print(people + " is not in people_ooo")
        else:
            for cycles in rostered_dict[people]:
                dates_unavailable = []
                for dates in rostered_dict[people][cycles]:
                    if str(dates[2]) in people_ooo[people][str(dates[0])][str(dates[1])]:
                        dates_unavailable.append(dates)
                # TODO including info on what day of the week it is would help speed up runtime later on.
                if dates_unavailable:
                    # you want to include the dates that they are unavailable, and the full range of dates you checked
                    unavailable_people[count][people] = [dates_unavailable, rostered_dict[people][cycles]]
    return unavailable_people


def build_dates_list(start_date, days_of_the_week, repeats, order):
    all_days = []
    total_days = repeats * len(order)
    # print(days_of_the_week)
    for times in range(total_days):
        days_this_week = []
        for days in days_of_the_week:
            new_date = start_date + timedelta(days=int(days) - 1)
            days_this_week.append([new_date.year, new_date.month, new_date.day])
        all_days.append(days_this_week)
        start_date += timedelta(days=int(7))
    # print(all_days)
    return all_days


# duration needs to be measured in terms of weeks
def compute_repeats(people_involved, duration):
    count = len(people_involved)
    if count >= duration:
        repeats = 1
    else:
        repeats = math.ceil(duration / count)
    return repeats


# TODO this function needs to be fixed, code is a bit messy
def create_schedule(date_list, order_list, roster_type):
    roster_solution = {}
    combined_list = [item for sublist in order_list for item in sublist]
    for count in range(0, len(date_list)):
        for date in date_list[count]:
            date_list_str = str(date[2]) + "-" + str(date[1]) + "-" + str(date[0])
            roster_solution[date_list_str] = combined_list[count]
    print("roster_solution: " + str(roster_solution))
    return roster_solution


# TODO right now the code doesn't account for the fact that 1 ooo in the week isn't a big deal.
# the roster should also be your fg1_rosters["team"]. It's specific to the team
# start date should be [year,month,day]
def calculate_rosters(roster, ooo_dict, start_date, roster_type, change_freezes):
    date_to_check = datetime(start_date[0], start_date[1], start_date[2])
    if date_to_check.weekday() != 0:  # mon = 0 , sunday = 6
        print("Start date should be a monday!")
        exit()
    else:
        start_dt_str = str(start_date[0]) + "-" + str(start_date[1]) + "-" + str(start_date[2])
        start = datetime.strptime(start_dt_str, "%Y-%m-%d")
        # duration for now will be toggled to 3 months or 3 full cycles, whichever is longer
        # if it's longer than 3 months will round up to completion of next cycle.
        duration = 12  # number of weeks, can be adjusted
        repeats = compute_repeats(roster["order"], duration)
        all_days = build_dates_list(start, roster["applicable days"], repeats, roster["order"])
        print("all_days: " + str(all_days))
        roster_to_check = assign_dates_to_people(roster["order"], all_days, repeats)
        # print("roster_to_check: " + str(roster_to_check))
        people_to_swap = check_their_availability(roster_to_check, ooo_dict, repeats)
        # print("people_to_swap: " + str(people_to_swap))
        new_order = roster_adjust(roster_to_check, roster["order"], ooo_dict, people_to_swap)
        # roster_solution = create_schedule(new_order, all_days, repeats)
        roster_solution = create_schedule(all_days, new_order, roster_type)
        for items in roster_solution:
            roster["dates"][items] = roster_solution[items]
        print("roster: " + str(roster))
    return roster


def main_test(start_date, test_file):
    with open(test_file, "r", encoding="utf-8") as f:
        calendar = Calendar(f.read())
    # update all shorthands in the fg1_names.json also
    people_dict = json.load(open("./fg1_names.json"))
    ooo_dict = json.load(open("./people_ooo.json"))
    rosters = json.load(open("./fg1_rosters.json"))
    ooo_dict = build_calendar_dict(people_dict["all"], ooo_dict, calendar)
    # start_date = [2025, 9, 8]
    rosters["pnw"] = calculate_rosters(rosters["pnw"], ooo_dict, start_date, "pnw")
    with open("./people_ooo.json", "w") as file:
        json.dump(ooo_dict, file, indent=4)
    with open("./fg1_rosters.json", "w") as file:
        json.dump(rosters, file, indent=4)
    # ics file is deleted so that you don't have to keep changing the name to the latest copy.


def main():
    # Adjust this accordingly based on computer file path
    desktop_file_path = "C:/Users/ace-j/Downloads/"
    laptop_file_path = "C:/Users/User/Downloads/"
    # TODO shift this elsewhere
    # Need to test one where the change freeze is for a future cycle
    change_freezes = [[[2025, 12, 19], [2026, 1, 2]]]  # start and end dates
    # need to test one where the change freeze is for an already computed cycle
    # TODO for this, add the people affected by change freezes to after the cycle
    change_freezes1 = [[2026, 1, 5], [2026, 1, 6]]
    # check which file path exists so that you don't need to keep switching when working on desktop or laptop
    if os.path.exists(desktop_file_path):
        file_path = desktop_file_path + "FG1 OOO Calendar.ics"
    elif os.path.exists(laptop_file_path):
        file_path = laptop_file_path + "FG1 OOO Calendar.ics"
    with open(file_path, "r", encoding="utf-8") as f:
        calendar = Calendar(f.read())
    # update all shorthands in the fg1_names.json also
    people_dict = json.load(open("./fg1_names.json"))
    ooo_dict = json.load(open("./people_ooo.json"))
    rosters = json.load(open("./fg1_rosters.json"))
    ooo_dict = build_calendar_dict(people_dict["all"], ooo_dict, calendar)
    # TODO what if this date is not a monday
    start_date = [2025, 9, 8]  # for now, ensure starting date is a monday
    rosters["pnw"] = calculate_rosters(rosters["pnw"], ooo_dict, start_date, "pnw", change_freezes)
    exit()
    with open("./people_ooo.json", "w") as file:
        json.dump(ooo_dict, file, indent=4)
    with open("./fg1_rosters.json", "w") as file:
        json.dump(rosters, file, indent=4)
    # ics file is deleted so that you don't have to keep changing the name to the latest copy.


if __name__ is main():
    main()