import re from typing import Dict, List, Sequence from .dates_add_days import add_days from .dates_day_of_week import day_of_week from .fleet_drivers_hours_data import DRIVING_LIMITS, DRIVING_LIMITS_HISTORY 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)