import re from typing import List, Optional, Sequence from .dates_add_days import add_days from .fleet_drivers_hours import DriverActivity, HoursBreach from .fleet_tachograph_rest_data import REST_LIMITS, REST_LIMITS_HISTORY 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, )