timewarrior extension to track working schedule hours

worked.py

#!/usr/bin/env python3

import calendar
import dataclasses
import json
import sys
from datetime import datetime, timedelta, UTC
from math import floor

TZ_OFFSET = timedelta(hours=1)
INTERVAL_DATE_FORMAT = "%Y%m%dT%H%M%SZ"
EXCLUSION_DATE_FORMAT = "%Y_%m_%d"
FIRST_WEEK_DAY = 0  # Monday


@dataclasses.dataclass
class TimeWarriorData:
    config: dict[str, str]
    holidays: list
    exclusions: list
    intervals: list[dict[str, datetime]]


@dataclasses.dataclass
class WorkedConfig:
    working_hours: list[str]
    working_hours_overrides: dict[datetime.date, float]
    alt_working_hours: list[str]
    alt_working_hours_dates: set[tuple]
    absenses: dict[datetime.date, float]
    reduced_hours: list[str]
    reduced_hours_dates: set[tuple]


def read_json():
    config: dict[str, str] = {}
    holidays: list = []
    exclusions: list = []
    intervals: list[tuple(datetime, datetime)] = []

    data = sys.stdin.read().splitlines()
    for line_number, line in enumerate(data):
        if line != "":
            if line.startswith("holidays."):
                holidays.append(
                    datetime.strptime(
                        line.split(".")[2].split(":")[0], EXCLUSION_DATE_FORMAT
                    ).date()
                )
            elif line.startswith("exclusions.days."):
                exclusions.append(
                    datetime.strptime(
                        line.split(".")[2].split(":")[0], EXCLUSION_DATE_FORMAT
                    ).date()
                )
            else:
                config_entry = line.split(":")
                config[config_entry[0]] = config_entry[1].strip(" ")
        else:
            break

    for interval in json.loads("".join(data[line_number:])):
        start = datetime.strptime(interval["start"], INTERVAL_DATE_FORMAT)
        if "end" in interval:
            end = datetime.strptime(interval["end"], INTERVAL_DATE_FORMAT)
        else:
            end = datetime.now(UTC).replace(tzinfo=None)

        intervals.append((start + TZ_OFFSET, end + TZ_OFFSET))

    return TimeWarriorData(config, holidays, exclusions, intervals)


def parse_worked_config(config):
    # __import__('pprint').pprint(config)

    working_hours = [float(v) for v in config["worked.working_hours"].split(",")]

    alt_working_hours = [
        float(v) for v in config["worked.working_hours.alt"].split(",")
    ]

    reduced_hours = [
        float(v) for v in config["worked.working_hours.reduced"].split(",")
    ]

    working_hours_overrides = {}
    for override in [
        k for k, v in config.items() if "worked.working_hours.overrides." in k
    ]:
        override_hours = config[override]
        working_hours_overrides[
            datetime.strptime(override.split(".")[-1], "%Y_%m_%d").date()
        ] = float(override_hours)

    alt_working_hours_dates = set()
    for alt_date_range in [
        k for k, v in config.items() if "worked.working_hours.alt.dates." in k
    ]:
        alt_start, alt_end = config[alt_date_range].split(" - ")
        alt_working_hours_dates.add(
            (
                datetime.strptime(alt_start, "%Y-%m-%d").date(),
                datetime.strptime(alt_end, "%Y-%m-%d").date(),
            )
        )

    reduced_hours_dates = set()
    for reduced_date_range in [
        k for k, v in config.items() if "worked.working_hours.reduced.dates." in k
    ]:
        reduced_start, reduced_end = config[reduced_date_range].split(" - ")
        reduced_hours_dates.add(
            (
                datetime.strptime(reduced_start, "%Y-%m-%d").date(),
                datetime.strptime(reduced_end, "%Y-%m-%d").date(),
            )
        )

    absenses = {}
    for absense in [k for k, v in config.items() if "worked.absenses" in k]:
        absenses_hours = config[absense]
        absenses[datetime.strptime(absense.split(".")[-1], "%Y_%m_%d").date()] = float(
            absenses_hours
        )

    return WorkedConfig(
        working_hours,
        working_hours_overrides,
        alt_working_hours,
        alt_working_hours_dates,
        absenses,
        reduced_hours,
        reduced_hours_dates,
    )


def to_hour_and_minutes(number, padding=3):
    return f"{floor(number):{padding}}h {round((number - floor(number)) * 60):2}m"


def print_entry(
    prefix,
    week_worked,
    week_expected,
    alt_week,
    has_holidays,
    has_overrides,
    has_absenses,
    reduced_hours,
):
    worked_hm = to_hour_and_minutes(week_worked)
    if round(week_worked, 1) > week_expected:
        worked_str = f"\033[32m{worked_hm}\033[00m"
    elif round(week_worked, 1) < week_expected:
        worked_str = f"\033[31m{worked_hm}\033[00m"
    else:
        worked_str = f"{worked_hm}"

    expected_hm = to_hour_and_minutes(week_expected)
    symbols = ""
    if has_holidays:
        symbols += "\033[32m●\033[00m "
    if alt_week:
        symbols += "\033[35m●\033[00m "
    if has_overrides:
        symbols += "\033[36m●\033[00m "
    if has_absenses:
        symbols += "\033[33m●\033[00m "
    if reduced_hours:
        symbols += "\033[34m●\033[00m"

    expected_str = f"{expected_hm} {symbols}"

    print(f"{prefix} │ {worked_str} of {expected_str}")


def main():
    timewarrior_data = read_json()
    config = parse_worked_config(timewarrior_data.config)

    exclusions = sorted(timewarrior_data.exclusions + timewarrior_data.holidays)
 
    total_expected = 0
    total_worked = 0
    week_worked = 0
    week_expected = 0
    month_worked = 0
    month_expected = 0
    previous_date = None
    previous_week = None
    previous_month = None
    alt_working_hours_week = False
    has_holidays = False
    has_overrides = False
    has_absenses = False
    reduced_hours_week = False

    print("───────────────────────────────")
    for interval in timewarrior_data.intervals:
        date = interval[0].date()
        week = interval[0].isocalendar().week
        month = date.month

        if date != previous_date:
            if (previous_week != week and previous_week is not None) or (
                previous_month != month and previous_month is not None
            ):
                for exclusion in exclusions:
                    if exclusion.isocalendar().week == previous_week:
                        if exclusion.weekday() not in (5, 6):
                            has_holidays = True
                        exclusions.remove(exclusion)

                print_entry(
                    f"{previous_date.year} W{previous_week:<2}",
                    week_worked,
                    week_expected,
                    alt_working_hours_week,
                    has_holidays,
                    has_overrides,
                    has_absenses,
                    reduced_hours_week,
                )

                total_expected += week_expected
                total_worked += week_worked

                month_expected += week_expected
                month_worked += week_worked

                alt_working_hours_week = False
                has_holidays = False
                has_overrides = False
                has_absenses = False
                reduced_hours_week = False

                week_expected = 0
                week_worked = 0

            if date not in exclusions:
                expected = 0

                for period_start, period_end in config.alt_working_hours_dates:
                    if period_start <= date <= period_end:
                        expected = config.alt_working_hours[interval[0].weekday()]
                        alt_working_hours_week = True
                        break
                for period_start, period_end in config.reduced_hours_dates:
                    if period_start <= date <= period_end:
                        expected = config.reduced_hours[interval[0].weekday()]
                        reduced_hours_week = True
                        break

                if date in config.working_hours_overrides:
                    expected = config.working_hours_overrides[date]
                    has_overrides = True
                if date in config.absenses:
                    expected -= config.absenses[date]
                    has_absenses = True

                if not (alt_working_hours_week or reduced_hours_week or has_overrides):
                    expected = config.working_hours[interval[0].weekday()]

                week_expected += expected


        if previous_month != month and previous_month is not None:
            print("┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄")
            print_entry(
                f"{previous_date.year} {calendar.month_abbr[previous_month]}",
                month_worked,
                month_expected,
                False,
                False,
                False,
                False,
                False,
            )
            print("───────────────────────────────")

            month_expected = 0
            month_worked = 0

        week_worked += (interval[1] - interval[0]).seconds / 60 / 60

        previous_week = week
        previous_date = date
        previous_month = month

    for exclusion in exclusions:
        if exclusion.isocalendar().week == week:
            if exclusion.weekday() not in (5, 6):
                has_holidays = True
            exclusions.remove(exclusion)

    print_entry(
        f"{date.year} W{week:<2}",
        week_worked,
        week_expected,
        alt_working_hours_week,
        has_holidays,
        has_overrides,
        has_absenses,
        reduced_hours_week,
    )

    month_expected += week_expected
    month_worked += week_worked

    print("┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄")
    print_entry(
        f"{date.year} {calendar.month_abbr[month]}",
        month_worked,
        month_expected,
        False,
        False,
        False,
        False,
        False,
    )

    total_expected += week_expected
    total_worked += week_worked
    diff = round(total_worked, 1) - total_expected

    total_worked_hm = to_hour_and_minutes(total_worked, 4)
    total_expected_hm = to_hour_and_minutes(total_expected, 4)
    diff_hm = to_hour_and_minutes(abs(diff), 0)

    eta = "-"
    if diff > 0:
        worked_str = f"\033[32m{total_worked_hm}\033[00m"
        diff_symbol = "+"
    elif diff < 0:
        worked_str = f"\033[31m{total_worked_hm}\033[00m"
        diff_symbol = "-"
        eta = datetime.strftime(datetime.now() + timedelta(hours=abs(diff)), "%H:%M")
    else:
        worked_str = f"{total_worked_hm}"
        diff_symbol = ""

    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
    print(f"Worked:               {worked_str}")
    print(f"Expected:             {total_expected_hm}")
    print(f"Diff:                 {diff_symbol+diff_hm:>9}")
    print("───────────────────────────────")
    print(f"ETA:                  {eta:>9}")
    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━", end="")


if __name__ == "__main__":
    main()

BUG 🐛

Si se curra un festivo en lugar de un laboral, el cálculo de horas totales falla. Esto es debido a que el intervalo de tiempo no se tiene en cuenta en el total de horas a currar (expected) al estar "excluido", y el del festivo no se cuenta.

El resultado es que se han currado más horas de las que tocarían.

Realmente no es un Bug, esto es debido a como funciona internamente Timewarrior. Pero le tengo que dar una vuelta para solucionarlo.

Workaround: reportar tiempo, unos pocos segundos, el día laboral que no se trabajó realmente, para que se tengan en cuenta las horas de ese día en el total de horas "esperadas" (expected).