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,
)