feat: add timecard extraction

Signed-off-by: Konstantin Fickel <mail@konstantinfickel.de>

src/streamer/localize/preconfigured_configurations.py

@@ -41,55 +41,3 @@ TaskConfiguration = RepositoryConfiguration(
         ),
     },
 )
-
-BasicTimesheetConfiguration = RepositoryConfiguration(
-    dimensions={
-        "timesheet": Dimension(
-            display_name="Timesheet",
-            comment="Used by Timesheet-Subcommand to create Timecards",
-            propagate=False,
-        )
-    },
-    markers={
-        "VacationDay": Marker(
-            display_name="Vacation Day",
-            placements=[
-                MarkerPlacement(
-                    if_with={"Timesheet"},
-                    dimension="timesheet",
-                    value="day_off_sick_leave",
-                )
-            ],
-        ),
-        "Holiday": Marker(
-            display_name="Offical Holiday",
-            placements=[
-                MarkerPlacement(
-                    if_with={"Timesheet"},
-                    dimension="timesheet",
-                    value="day_off_holiday",
-                )
-            ],
-        ),
-        "SickLeave": Marker(
-            display_name="Sick Leave",
-            placements=[
-                MarkerPlacement(
-                    if_with={"Timesheet"},
-                    dimension="timesheet",
-                    value="day_off_sick_leave",
-                )
-            ],
-        ),
-        "UndertimeDay": Marker(
-            display_name="Undertime Leave",
-            placements=[
-                MarkerPlacement(
-                    if_with={"Timesheet"},
-                    dimension="timesheet",
-                    value="day_off_undertime",
-                )
-            ],
-        ),
-    },
-)

src/streamer/query/find.py

@@ -26,4 +26,10 @@ def find_shard_by_position(
     )
 
 
-__all__ = ["find_shard_by_position", "find_shard"]
+def find_shard_by_set_dimension(
+    shards: list[LocalizedShard], dimension: str
+) -> list[LocalizedShard]:
+    return find_shard(shards, lambda shard: dimension in shard.location)
+
+
+__all__ = ["find_shard_by_position", "find_shard", "find_shard_by_set_dimension"]

src/streamer/timesheet/configuration.py

@@ -0,0 +1,115 @@
+from enum import StrEnum
+
+from streamer.localize import RepositoryConfiguration
+from streamer.localize.repository_configuration import (
+    Dimension,
+    Marker,
+    MarkerPlacement,
+)
+
+TIMESHEET_TAG = "Timesheet"
+TIMESHEET_DIMENSION_NAME = "timesheet"
+
+
+class TimesheetPointType(StrEnum):
+    Card = "CARD"
+    SickLeave = "SICK_LEAVE"
+    Vacation = "VACATION"
+    Undertime = "UNDERTIME"
+    Holiday = "HOLIDAY"
+    Break = "BREAK"
+
+
+BasicTimesheetConfiguration = RepositoryConfiguration(
+    dimensions={
+        TIMESHEET_DIMENSION_NAME: Dimension(
+            display_name="Timesheet",
+            comment="Used by Timesheet-Subcommand to create Timecards",
+            propagate=False,
+        )
+    },
+    markers={
+        TIMESHEET_TAG: Marker(
+            display_name="A default time card",
+            placements=[
+                MarkerPlacement(
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.Vacation.value,
+                    overwrites=False,
+                )
+            ],
+        ),
+        "VacationDay": Marker(
+            display_name="Vacation Day",
+            placements=[
+                MarkerPlacement(
+                    if_with={TIMESHEET_TAG},
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.Vacation.value,
+                )
+            ],
+        ),
+        "Break": Marker(
+            display_name="Break",
+            placements=[
+                MarkerPlacement(
+                    if_with={TIMESHEET_TAG},
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.Break.value,
+                )
+            ],
+        ),
+        "LunchBreak": Marker(
+            display_name="Break",
+            placements=[
+                MarkerPlacement(
+                    if_with={TIMESHEET_TAG},
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.Break.value,
+                )
+            ],
+        ),
+        "Feierabend": Marker(
+            display_name="Break",
+            placements=[
+                MarkerPlacement(
+                    if_with={TIMESHEET_TAG},
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.Break.value,
+                )
+            ],
+        ),
+        "Holiday": Marker(
+            display_name="Offical Holiday",
+            placements=[
+                MarkerPlacement(
+                    if_with={TIMESHEET_TAG},
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.Holiday.value,
+                )
+            ],
+        ),
+        "SickLeave": Marker(
+            display_name="Sick Leave",
+            placements=[
+                MarkerPlacement(
+                    if_with={TIMESHEET_TAG},
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.SickLeave.value,
+                )
+            ],
+        ),
+        "UndertimeDay": Marker(
+            display_name="Undertime Leave",
+            placements=[
+                MarkerPlacement(
+                    if_with={TIMESHEET_TAG},
+                    dimension=TIMESHEET_DIMENSION_NAME,
+                    value=TimesheetPointType.Undertime.value,
+                )
+            ],
+        ),
+    },
+)
+
+__all__ = ["BasicTimesheetConfiguration", "TIMESHEET_TAG", "TIMESHEET_DIMENSION_NAME"]

src/streamer/timesheet/extract.py

@@ -0,0 +1,114 @@
+from datetime import datetime
+from itertools import groupby
+
+from pydantic import BaseModel
+
+from streamer.localize import LocalizedShard
+from streamer.query.find import find_shard_by_set_dimension
+
+from .configuration import TIMESHEET_DIMENSION_NAME, TimesheetPointType
+from .timecard import SpecialDayType, Timecard, Timesheet
+
+
+class TimesheetPoint(BaseModel):
+    moment: datetime
+    type: TimesheetPointType
+
+
+def shard_to_timesheet_point(shard: LocalizedShard) -> TimesheetPoint:
+    return TimesheetPoint(
+        moment=shard.moment,
+        type=TimesheetPointType(shard.location[TIMESHEET_DIMENSION_NAME]),
+    )
+
+
+def shards_to_timesheet_points(shards: list[LocalizedShard]) -> list[TimesheetPoint]:
+    return list(
+        map(
+            shard_to_timesheet_point,
+            find_shard_by_set_dimension(shards, TIMESHEET_DIMENSION_NAME),
+        )
+    )
+
+
+def aggregate_timecard_day(points: list[TimesheetPoint]) -> Timesheet | None:
+    sorted_points = sorted(points, key=lambda point: point.moment)
+
+    is_sick_leave = False
+    special_day_type = None
+
+    card_date = sorted_points[0].moment.date()
+
+    # We expect timesheet points to alternate between "Card" (start work) and
+    # "Break" (end work). Starting in "break" means we are not currently in a
+    # work block until we see the first Card.
+    last_is_break = True
+    last_time = sorted_points[0].moment.time()
+
+    timecards: list[Timecard] = []
+    for point in sorted_points:
+        if point.moment.date() != card_date:
+            raise ValueError("Dates of all given timesheet days should be consistent")
+
+        point_time = point.moment.time()
+
+        match point.type:
+            case TimesheetPointType.Holiday:
+                if special_day_type is not None:
+                    raise ValueError(
+                        f"{card_date} is both {point.type} and {special_day_type}"
+                    )
+                special_day_type = SpecialDayType.Holiday
+            case TimesheetPointType.Vacation:
+                if special_day_type is not None:
+                    raise ValueError(
+                        f"{card_date} is both {point.type} and {special_day_type}"
+                    )
+                special_day_type = SpecialDayType.Vacation
+            case TimesheetPointType.Undertime:
+                if special_day_type is not None:
+                    raise ValueError(
+                        f"{card_date} is both {point.type} and {special_day_type}"
+                    )
+                special_day_type = SpecialDayType.Undertime
+            case TimesheetPointType.SickLeave:
+                is_sick_leave = True
+            case TimesheetPointType.Break:
+                if not last_is_break:
+                    timecards.append(Timecard(from_time=last_time, to_time=point_time))
+                    last_is_break = True
+                    last_time = point_time
+            case TimesheetPointType.Card:
+                if last_is_break:
+                    last_is_break = False
+                    last_time = point_time
+
+    if not last_is_break:
+        raise ValueError(f"Last Timecard of {card_date} is not a break!")
+
+    if len(timecards) == 0 and not is_sick_leave and special_day_type is None:
+        return None
+
+    return Timesheet(
+        date=card_date,
+        is_sick_leave=is_sick_leave,
+        special_day_type=special_day_type,
+        timecards=timecards,
+    )
+
+
+def aggregate_timecards(points: list[TimesheetPoint]) -> list[Timesheet]:
+    day_timecards = [
+        aggregate_timecard_day(list(timecard))
+        for _date, timecard in groupby(points, key=lambda point: point.moment.date())
+    ]
+
+    return [timecard for timecard in day_timecards if timecard is not None]
+
+
+def extract_timesheets(shards: list[LocalizedShard]) -> list[Timesheet]:
+    points = shards_to_timesheet_points(shards)
+    return aggregate_timecards(points)
+
+
+__all__ = ["extract_timesheets"]

src/streamer/timesheet/timecard.py

@@ -0,0 +1,23 @@
+from datetime import date, time
+from enum import StrEnum
+
+from pydantic import BaseModel
+
+
+class SpecialDayType(StrEnum):
+    Vacation = "VACATION"
+    Undertime = "UNDERTIME"
+    Holiday = "HOLIDAY"
+    Weekend = "WEEKEND"
+
+
+class Timecard(BaseModel):
+    from_time: time
+    to_time: time
+
+
+class Timesheet(BaseModel):
+    date: date
+    is_sick_leave: bool = False
+    special_day_type: SpecialDayType | None = None
+    timecards: list[Timecard]