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"^\d{4}-\d{2}-\d{2}$")
HH_MM = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$")
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.match(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.match(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)