fleet.drivers-hours

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 Dict, List, Sequence

from .dates_add_days import add_days  ← from dates.add-days ^1.0.0 · built alongside by fune
from .dates_day_of_week import day_of_week  ← from dates.day-of-week ^1.0.0 · built alongside by fune
from .fleet_drivers_hours_data import DRIVING_LIMITS, DRIVING_LIMITS_HISTORY  ← this capability’s own data, compiled from data/limits.json into the same file by fune build
from .fleet_drivers_hours_types import DriverActivity, DriversHoursCheck, HoursBreach

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")
WEEK = 10080


def _limit_on(rule: str, on_date: str) -> int:
    best = None
    earliest = None
    for row in DRIVING_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 DRIVING_LIMITS_HISTORY != "full" and earliest is not None and on_date < earliest:
        raise ValueError(
            'no drivers\' hours 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, DRIVING_LIMITS_HISTORY, earliest)
        )
    raise ValueError('no drivers\' hours limit "%s" on %s' % (rule, on_date))


class _Pending:
    __slots__ = ("rule", "at_minute", "actual", "limit")

    def __init__(self, rule: str, at_minute: int, limit: int) -> None:
        self.rule = rule
        self.at_minute = at_minute
        self.actual = 0
        self.limit = limit


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_drivers_hours(start_date: str, start_time: str, activities: Sequence[DriverActivity]) -> DriversHoursCheck:
    """Check a run of tachograph activities against the GB/EU drivers' hours
    rules on driving time and breaks (Regulation (EC) 561/2006, Articles 6 and 7).

    Activities are back to back from start_date/start_time, in local time. A
    rest of 45 minutes, or 15 then 30, ends a driving period; a rest of 9 hours
    or more is a daily rest and ends the daily driving time. Weeks are fixed
    weeks, Monday 00:00 to Sunday 24:00, so driving across Sunday midnight is
    split.
    """
    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,))
    driving_period = _limit_on("driving-period", start_date)
    full_break = _limit_on("break", start_date)
    split_first = _limit_on("split-break-first", start_date)
    split_second = _limit_on("split-break-second", start_date)
    daily = _limit_on("daily-driving", start_date)
    daily_extended = _limit_on("daily-driving-extended", start_date)
    extensions_allowed = _limit_on("extended-days-per-week", start_date)
    weekly_limit = _limit_on("weekly-driving", start_date)
    fortnight_limit = _limit_on("fortnightly-driving", start_date)
    daily_rest = _limit_on("daily-rest-minimum", start_date)

    # Merge neighbours of the same kind: two rests back to back are one
    # uninterrupted rest, whatever the recording split them into.
    segments: List[List] = []
    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 segments and segments[-1][0] == activity.kind:
            segments[-1][1] += minutes
        else:
            segments.append([activity.kind, minutes])

    start_minute = int(start_time[0:2]) * 60 + int(start_time[3:5])
    week_base = (day_of_week(start_date) - 1) * 1440 + start_minute

    breaches: List[_Pending] = []
    state = {
        "accumulated": 0,
        "first_part": False,
        "break_breach": None,
        "day_driving": 0,
        "day_passed_daily": None,
        "day_week": None,
        "day_breach": None,
    }
    extensions: Dict[int, int] = {}
    weekly: Dict[int, int] = {}
    weekly_breaches: Dict[int, _Pending] = {}
    fortnight_breaches: Dict[int, _Pending] = {}

    def end_driving_period() -> None:
        if state["break_breach"] is not None:
            state["break_breach"].actual = state["accumulated"]
        state["accumulated"] = 0
        state["first_part"] = False
        state["break_breach"] = None

    def close_day() -> None:
        day_driving = state["day_driving"]
        if day_driving > daily and state["day_week"] is not None:
            used = extensions.get(state["day_week"], 0) + 1
            extensions[state["day_week"]] = used
            if used > extensions_allowed and day_driving <= daily_extended and state["day_passed_daily"] is not None:
                pending = _Pending("daily-driving", state["day_passed_daily"], daily)
                pending.actual = day_driving
                breaches.append(pending)
        if state["day_breach"] is not None:
            state["day_breach"].actual = day_driving
        state["day_driving"] = 0
        state["day_passed_daily"] = None
        state["day_week"] = None
        state["day_breach"] = None

    total_driving = 0
    offset = 0
    for kind, minutes in segments:
        if kind == "driving":
            total_driving += minutes
            start = offset
            end = offset + minutes
            while start < end:
                week = (week_base + start) // WEEK
                chunk_end = min(end, (week + 1) * WEEK - week_base)
                m = chunk_end - start
                if state["break_breach"] is None and state["accumulated"] + m > driving_period:
                    pending = _Pending("break", start + (driving_period - state["accumulated"]), driving_period)
                    state["break_breach"] = pending
                    breaches.append(pending)
                state["accumulated"] += m
                if state["day_week"] is None:
                    state["day_week"] = week
                day_driving = state["day_driving"]
                if state["day_passed_daily"] is None and day_driving + m > daily:
                    state["day_passed_daily"] = start + (daily - day_driving)
                if state["day_breach"] is None and day_driving + m > daily_extended:
                    pending = _Pending("daily-driving", start + (daily_extended - day_driving), daily_extended)
                    state["day_breach"] = pending
                    breaches.append(pending)
                state["day_driving"] = day_driving + m
                before = weekly.get(week, 0)
                if week not in weekly_breaches and before + m > weekly_limit:
                    pending = _Pending("weekly-driving", start + (weekly_limit - before), weekly_limit)
                    weekly_breaches[week] = pending
                    breaches.append(pending)
                pair_before = weekly.get(week - 1, 0) + before
                if week not in fortnight_breaches and pair_before + m > fortnight_limit:
                    pending = _Pending("fortnightly-driving", start + (fortnight_limit - pair_before), fortnight_limit)
                    fortnight_breaches[week] = pending
                    breaches.append(pending)
                weekly[week] = before + m
                start = chunk_end
        elif kind == "rest":
            if minutes >= full_break or (state["first_part"] and minutes >= split_second):
                end_driving_period()
            elif minutes >= split_first and state["accumulated"] > 0:
                state["first_part"] = True
            if minutes >= daily_rest:
                close_day()
        offset += minutes
    end_driving_period()
    close_day()
    for week, pending in weekly_breaches.items():
        pending.actual = weekly.get(week, 0)
    for week, pending in fortnight_breaches.items():
        pending.actual = weekly.get(week - 1, 0) + weekly.get(week, 0)

    # sorted() is stable, so breaches at the same minute keep the order found.
    ordered = [
        HoursBreach(rule=b.rule, at=_local_time(start_date, start_minute + b.at_minute), actual=b.actual, limit=b.limit)
        for b in sorted(breaches, key=lambda b: b.at_minute)
    ]
    return DriversHoursCheck(compliant=len(ordered) == 0, driving_minutes=total_driving, breaches=ordered)