Workdays-Cycle-Allocation

workdays_cycle_allocation.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
#
# Copyleft (ɔ) 2024 wildfootw <[email protected]>
#
# Distributed under terms of the MIT license.

import pandas as pd
from datetime import datetime, timedelta

# Taiwan public holidays
holidays_2024 = [
    "2024-01-01",  # New Year's Day
    "2024-02-09", "2024-02-12", "2024-02-13", "2024-02-14", "2024-02-15",  # Chinese New Year
    "2024-02-28",  # Peace Memorial Day
    "2024-04-04", "2024-04-05",  # Children's Day and Tomb Sweeping Day
    "2024-05-01",  # Labor Day
    "2024-06-10",  # Dragon Boat Festival
    "2024-09-17",  # Mid-Autumn Festival
    "2024-10-10",  # National Day
]
holidays_2025 = [ # [TODO] Feature: makeup working days
    "2025-01-01",  # New Year's Day
    "2025-01-27", "2025-01-28", "2025-01-29", "2025-01-30", "2025-01-31",  # Chinese New Year
    "2025-02-28",  # Peace Memorial Day
    "2025-04-03", "2025-04-04",  # Children's Day and Tomb Sweeping Day
    "2025-05-01",  # Labor Day
    "2025-05-30",  # Dragon Boat Festival
    "2025-10-06",  # Mid-Autumn Festival
    "2025-10-10",  # National Day
]

# ISO 8601 defines the first week of the year as the week with the first Thursday of the year.
def get_iso8601_year_start(year):
    jan4 = datetime(year, 1, 4)
    start_of_week1 = jan4 - timedelta(days=jan4.weekday())
    return start_of_week1

def get_iso8601_year_end(year):
    start_of_next_year = get_iso8601_year_start(year + 1)
    end_of_this_year = start_of_next_year - timedelta(days=1)
    return end_of_this_year

# Calculate workdays for each week
def get_workdays(week_start_date, week_end_date, holidays):
    date_range = pd.date_range(week_start_date, week_end_date)
    workdays = [d for d in date_range if d.weekday() < 5 and d.strftime("%Y-%m-%d") not in holidays]
    return len(workdays)

# Function to calculate weekly workdays for a given year
def calculate_weekly_workdays_for_year(year, holidays):
    year_start_date = get_iso8601_year_start(year)
    year_end_date = get_iso8601_year_end(year)
    weeks = pd.date_range(year_start_date, year_end_date, freq='W-MON')

    weekly_workdays = []
    for i in range(len(weeks)):
        week_start_date = weeks[i]
        week_end_date = weeks[i] + timedelta(days=6)
        workdays_count = get_workdays(week_start_date, week_end_date, set(holidays))
        weekly_workdays.append((week_start_date, week_end_date, workdays_count))

    return weekly_workdays

# Use dynamic programming (DP) and recursion to allocate workdays for each cycle
def calculate_workdays_dp(weekly_workdays, total_cycles):
    """
    Calculate the allocation of workdays for each cycle using dynamic programming and recursion.
    Args:
    - weekly_workdays: List of tuples, each containing (week_start_date, week_end_date, workdays_count)
    - total_cycles: Total number of cycles to divide the weeks into
    - max_difference: Maximum allowable difference between the cycle with the most and least workdays

    Returns:
    - A list of end week indices for each cycle
    """
    memo = {}  # Memoization dictionary to store the results of subproblems
    total_weeks = len(weekly_workdays)  # Total number of weeks in the year

    def dp(week_index, cycle_index):
        """
        Recursive function to calculate the maximum and minimum workdays for each cycle.
        Args:
        - week_index: Current week index being processed
        - cycle_index: Current cycle index being processed

        Returns:
        - A tuple containing ((max_workdays_in_a_cycle, min_workdays_in_a_cycle), week_end_index)
        where week_end_index is the index of the last week included in the current cycle.
        """
        #print(f"Processing dp(week_index={week_index}, cycle_index={cycle_index})", end="")  # Debug output

        # If remaining weeks are less than the remaining cycles, return infinity
        if total_weeks - week_index < total_cycles - cycle_index:
            result = ((float("inf"), float("-inf")), -1)
            #print(f" -> early return {result}")  # Debug output
            return result

        # If memoization contains the result, return it
        if (week_index, cycle_index) in memo:
            result = memo[(week_index, cycle_index)]
            #print(f" -> memo hit {result}")  # Debug output
            return result

        # If it's the last cycle, sum up the remaining workdays
        if total_cycles - cycle_index == 1:
            sum_of_days = sum(weekly_workdays[i][2] for i in range(week_index, total_weeks))
            result = ((sum_of_days, sum_of_days), total_weeks - 1)
            memo[(week_index, cycle_index)] = result
            #print(f" -> last cycle {result}")  # Debug output
            return result

        #print()  # To ensure the following output is on a new line
        smallest_difference = float("inf")  # Initialize the smallest difference variable
        best_result = None
        for i in range(week_index, total_weeks):
            current_cycle_workdays = sum(weekly_workdays[j][2] for j in range(week_index, i + 1))
            ((next_max, next_min), _) = dp(i + 1, cycle_index + 1)

            current_max = max(current_cycle_workdays, next_max)
            current_min = min(current_cycle_workdays, next_min)

            if current_max - current_min < smallest_difference:
                smallest_difference = current_max - current_min
                best_result = ((current_max, current_min), i)
                #print(f"Processing dp(week_index={week_index}, cycle_index={cycle_index}) -> new best {best_result}")  # Debug output

        memo[(week_index, cycle_index)] = best_result
        #print(f"Processing dp(week_index={week_index}, cycle_index={cycle_index}) -> memoizing {best_result}")  # Debug output
        return best_result

    dp(0, 0)

    end_week_indices = [memo[(0, 0)][1]]
    for i in range(0, total_cycles - 1):
        end_week_indices.append(memo[(end_week_indices[i] + 1, i + 1)][1])

    return end_week_indices

# Function to calculate cycle ranges and workdays
def calculate_cycle_ranges(weekly_workdays, end_week_indices):
    """
    Calculate the range and workdays for each cycle.
    Args:
    - weekly_workdays: List of tuples, each containing (week_start_date, week_end_date, workdays_count)
    - end_week_indices: List of end week indices for each cycle

    Returns:
    - A DataFrame containing the cycle ranges and workdays
    """
    cycle_ranges = []
    start_idx = 0
    for end_idx in end_week_indices: # [TODO] Bug: wrong year display while cross year
        start_date_week = weekly_workdays[start_idx][0].strftime("%Y-W%V")
        end_date_week = weekly_workdays[end_idx][1].strftime("%Y-W%V")
        start_date = weekly_workdays[start_idx][0].strftime("%Y-%m-%d")
        end_date = weekly_workdays[end_idx][1].strftime("%Y-%m-%d")
        total_days = sum(weekly_workdays[i][2] for i in range(start_idx, end_idx + 1))
        cycle_ranges.append((start_date_week, end_date_week, start_date, end_date, total_days))
        start_idx = end_idx + 1

    df = pd.DataFrame(cycle_ranges, columns=["Start Week", "End Week", "Start Date", "End Date", "Workdays"])
    return df

if __name__ == "__main__":
    # Calculate weekly workdays for 2024
    print("Calculating weekly workdays for 2024...")
    weekly_workdays_2024 = calculate_weekly_workdays_for_year(2024, holidays_2024)
    print("Weekly workdays for 2024:")
    for week in weekly_workdays_2024:
        print(week)

#    # Calculate weekly workdays for 2024
#    print("Calculating weekly workdays for 2025...")
#    weekly_workdays_2025 = calculate_weekly_workdays_for_year(2025, holidays_2025)
#    print("Weekly workdays for 2025:")
#    for week in weekly_workdays_2025:
#        print(week)

    weekly_workdays = weekly_workdays_2024

    # Use DP and recursion to calculate the cycles
    print("\nCalculating workdays per cycle using DP and recursion...")
    end_week_indices = calculate_workdays_dp(weekly_workdays, total_cycles=10)
    print("Workdays per cycle (end week indices):")
    print(end_week_indices)

    # Calculate the cycle ranges and workdays
    print("\nCalculating cycle ranges and workdays...")
    cycle_ranges_df = calculate_cycle_ranges(weekly_workdays, end_week_indices)
    print("Cycle ranges and workdays:")
    print(cycle_ranges_df)