fleet.tachograph-rest Unreviewed

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

import re
from typing import List, Optional, Sequence

from .dates_add_days import add_days  ← from dates.add-days ^1.0.0 · built alongside by fune
from .fleet_drivers_hours import DriverActivity, HoursBreach  ← from fleet.drivers-hours ^1.0.0 · built alongside by fune
from .fleet_tachograph_rest_data import REST_LIMITS, REST_LIMITS_HISTORY  ← this capability’s own data, compiled from data/rest-limits.json into the same file by fune build
from .fleet_tachograph_rest_types import RestCheck

ISO_DATE = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}")
HH_MM = re.compile(r"([01][0-9]|2[0-3]):[0-5][0-9]")
KINDS = ("driving", "work", "availability", "rest")


def _limit_on(rule: str, on_date: str) -> int:
    best = None
    earliest = None
    for row in REST_LIMITS:
        if row.rule != rule:
            continue
        if earliest is None or row.valid_from < earliest:
            earliest = row.valid_from
        if on_date < row.valid_from:
            continue
        if row.valid_to is not None and on_date > row.valid_to:
            continue
        if best is None or row.valid_from > best.valid_from:
            best = row
    if best is not None:
        return best.value
    if REST_LIMITS_HISTORY != "full" and earliest is not None and on_date < earliest:
        raise ValueError(
            'no rest limit "%s" on %s: this build was installed with history=%s, '
            "so it only carries limits from %s. Reinstall with history=full for older records."
            % (rule, on_date, REST_LIMITS_HISTORY, earliest)
        )
    raise ValueError('no rest limit "%s" on %s' % (rule, on_date))


def _local_time(start_date: str, minute: int) -> str:
    day = minute // 1440
    in_day = minute - day * 1440
    return "%sT%02d:%02d" % (add_days(start_date, day), in_day // 60, in_day % 60)


def check_tachograph_rest(start_date: str, start_time: str, activities: Sequence[DriverActivity]) -> RestCheck:
    """Check daily and weekly rest against Regulation (EC) 561/2006, Article 8,
    as it applies in Great Britain.

    The records must start straight after a weekly rest, so the first 24-hour
    and six-day windows open at start_date/start_time. A daily rest is judged
    on the part of it inside the 24-hour window (Art. 8(2)): 11 hours there is
    regular, 9 to 11 is reduced, and 3 hours then 9 is a split regular rest.
    """
    if not isinstance(start_date, str) or not ISO_DATE.fullmatch(start_date):
        raise ValueError('startDate must be an ISO date (YYYY-MM-DD), received "%s"' % (start_date,))
    if not isinstance(start_time, str) or not HH_MM.fullmatch(start_time):
        raise ValueError('startTime must be HH:MM, 24-hour, received "%s"' % (start_time,))
    window = _limit_on("daily-rest-window", start_date)
    regular_daily = _limit_on("regular-daily-rest", start_date)
    reduced_daily = _limit_on("reduced-daily-rest", start_date)
    split_first = _limit_on("split-rest-first", start_date)
    split_second = _limit_on("split-rest-second", start_date)
    reduced_allowed = _limit_on("reduced-daily-rests-allowed", start_date)
    regular_weekly = _limit_on("regular-weekly-rest", start_date)
    reduced_weekly = _limit_on("reduced-weekly-rest", start_date)
    weekly_window = _limit_on("weekly-rest-window", start_date)

    # Rests back to back are one uninterrupted rest: [start, end] pairs.
    rests: List[List[int]] = []
    total = 0
    last_kind: Optional[str] = None
    for activity in activities:
        if activity.kind not in KINDS:
            raise ValueError('unknown activity kind "%s"' % (activity.kind,))
        minutes = activity.minutes
        if isinstance(minutes, bool) or not isinstance(minutes, int) or minutes < 1:
            raise ValueError("minutes must be a whole number of at least 1, received %s" % (minutes,))
        if activity.kind == "rest":
            if last_kind == "rest":
                rests[-1][1] += minutes
            else:
                rests.append([total, total + minutes])
        last_kind = activity.kind
        total += minutes

    breaches: List[tuple] = []  # (at_minute, rule, actual, limit)
    events: List[tuple] = []  # (start, end, weekly, reduced)

    # Daily rest: one within each 24 hours after the previous rest ended.
    window_start = 0
    while True:
        window_end = window_start + window
        found = None
        three_hours_taken = False
        longest = 0
        for rest in rests:
            if rest[0] < window_start or rest[0] >= window_end:
                continue
            portion = min(rest[1], window_end) - rest[0]
            if portion > longest:
                longest = portion
            if portion >= reduced_daily:
                found = (rest, portion >= regular_daily or (three_hours_taken and portion >= split_second))
                break
            if portion >= split_first:
                three_hours_taken = True
        if found is not None:
            rest, regular = found
            if rest[1] - rest[0] < reduced_weekly:
                events.append((rest[0], rest[1], False, not regular))
            window_start = rest[1]
            continue
        if window_end > total:
            break  # the records stop before the window closes
        breaches.append((window_end, "daily-rest", longest, reduced_daily))
        # Pick the count up again after the next rest long enough to be one.
        following = [r for r in rests if r[0] >= window_start and r[1] - r[0] >= reduced_daily]
        if not following:
            break
        window_start = following[0][1]

    for rest in rests:
        length = rest[1] - rest[0]
        if length >= reduced_weekly:
            events.append((rest[0], rest[1], True, length < regular_weekly))
    events.sort(key=lambda e: e[0])

    daily_rests = 0
    reduced_daily_rests = 0
    weekly_rests = 0
    reduced_weekly_rests = 0
    previous_weekly_end = 0
    previous_reduced = False
    reduced_since_weekly = 0
    for start, end, weekly, reduced in events:
        if not weekly:
            daily_rests += 1
            if reduced:
                reduced_daily_rests += 1
                reduced_since_weekly += 1
                if reduced_since_weekly > reduced_allowed:
                    breaches.append((start, "reduced-daily-rest", reduced_since_weekly, reduced_allowed))
            continue
        weekly_rests += 1
        deadline = previous_weekly_end + weekly_window
        if start > deadline:
            breaches.append((deadline, "weekly-rest-late", start - previous_weekly_end, weekly_window))
        if reduced:
            reduced_weekly_rests += 1
            if previous_reduced:
                breaches.append((start, "reduced-weekly-rest", end - start, regular_weekly))
        previous_weekly_end = end
        previous_reduced = reduced
        reduced_since_weekly = 0
    # No weekly rest yet, and the records run past the deadline without ending
    # in a rest that began in time (and may yet become one).
    deadline = previous_weekly_end + weekly_window
    ends_in_rest = bool(rests) and rests[-1][1] == total and rests[-1][0] <= deadline
    if total > deadline and not ends_in_rest:
        breaches.append((deadline, "weekly-rest-late", total - previous_weekly_end, weekly_window))

    start_minute = int(start_time[0:2]) * 60 + int(start_time[3:5])
    # sorted() is stable, so breaches at the same minute keep the order found.
    ordered = [
        HoursBreach(rule=rule, at=_local_time(start_date, start_minute + at), actual=actual, limit=limit)
        for at, rule, actual, limit in sorted(breaches, key=lambda b: b[0])
    ]
    return RestCheck(
        compliant=len(ordered) == 0,
        daily_rests=daily_rests,
        reduced_daily_rests=reduced_daily_rests,
        weekly_rests=weekly_rests,
        reduced_weekly_rests=reduced_weekly_rests,
        breaches=ordered,
    )