Coverage for arbejdstimer/arbejdstimer.py: 95.62%
238 statements
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-04 15:36:40 +00:00
1"""Working hours (Danish arbejdstimer) or not? API."""
3import copy
4import datetime as dti
5import json
6import os
7import pathlib
8import sys
9from typing import Tuple, Union, no_type_check
11from pydantic import ValidationError
13import arbejdstimer.api as api
15DEBUG_VAR = 'ARBEJDSTIMER_DEBUG'
16DEBUG = os.getenv(DEBUG_VAR)
18ENCODING = 'utf-8'
19ENCODING_ERRORS_POLICY = 'ignore'
21DEFAULT_CONFIG_NAME = '.arbejdstimer.json'
22CfgType = dict[str, Union[dict[str, str], list[dict[str, Union[str, list[str]]]]]]
23WorkingHoursType = Union[tuple[int, int], tuple[None, None]]
24CmdType = Tuple[str, str, str, bool]
25DATE_FMT = '%Y-%m-%d'
26YEAR_MONTH_FORMAT = '%Y-%m'
27DEFAULT_WORK_HOURS_MARKER = (None, None)
28DEFAULT_WORK_HOURS_CLOSED_INTERVAL = (7, 16)
31@no_type_check
32def year_month_me(date) -> str:
33 """DRY."""
34 return date.strftime(YEAR_MONTH_FORMAT)
37@no_type_check
38def load_config(path_or_str):
39 """DRY."""
40 with open(path_or_str, 'rt', encoding=ENCODING) as handle:
41 return json.load(handle)
44def weekday(date: dti.date) -> int:
45 """Return current weekday."""
46 return date.isoweekday()
49def no_weekend(day_number: int) -> bool:
50 """Return if day number is weekend."""
51 return day_number < 6
54def the_hour() -> int:
55 """Return the hour of day as integer within [0, 23]."""
56 return dti.datetime.now().hour
59@no_type_check
60def days_of_year(day=None) -> list[dti.date]:
61 """Return all days of the year that contains the day."""
62 year = dti.date.today().year if day is None else day.year
63 d_start = dti.date(year, 1, 1)
64 d_end = dti.date(year, 12, 31)
65 return [d_start + dti.timedelta(days=x) for x in range((d_end - d_start).days + 1)]
68@no_type_check
69def workdays(off_days: list[dti.date], days=None) -> list[dti.date]:
70 """Return all workdays of the year that contains the day."""
71 if days is None:
72 days = days_of_year(None)
73 return [cand for cand in days if cand not in off_days and no_weekend(weekday(cand))]
76@no_type_check
77def workdays_of_month(work_days, month) -> list[dti.date]:
78 """Return all workdays of the month."""
79 return [work_day for work_day in work_days if year_month_me(work_day) == month]
82@no_type_check
83def workdays_count_per_month(work_days) -> dict[str, int]:
84 """Return the workday count per month of the year that contains the day."""
85 per = {}
86 for work_day in work_days:
87 month = year_month_me(work_day)
88 if month not in per:
89 per[month] = 0
90 per[month] += 1
91 return per
94@no_type_check
95def cumulative_workdays_count_per_month(work_days) -> dict[str, int]:
96 """Return the cumulative workday count per month of the year that contains the day."""
97 per = workdays_count_per_month(work_days)
98 months = list(per)
99 cum = copy.deepcopy(per)
100 for slot, month in enumerate(months, start=1):
101 cum[month] = sum(per[m] for m in months[:slot])
102 return cum
105@no_type_check
106def workdays_count_of_month_in_between(work_days, month, day, first_month, last_month) -> int:
107 """Return the workday count of month to date for day (incl.) given first and last month."""
108 per = workdays_count_per_month(work_days)
109 if month not in per or month < first_month or month > last_month:
110 return 0
112 wds = workdays_of_month(work_days, month)
113 return sum(1 for d in wds if d.day <= day)
116@no_type_check
117def closed_interval_months(work_days) -> tuple[str, str]:
118 """DRY."""
119 return year_month_me(work_days[0]), year_month_me(work_days[-1])
122@no_type_check
123def workdays_count_of_year_in_between(work_days, month, day, first_month=None, last_month=None) -> int:
124 """Return the workday count of year to date for day (incl.) given first and last month."""
125 per = workdays_count_per_month(work_days)
126 if any(
127 (
128 month not in per,
129 first_month and first_month not in per,
130 last_month and last_month not in per,
131 )
132 ):
133 return 0
135 initial, final = closed_interval_months(work_days)
136 if first_month is None:
137 first_month = initial
138 if last_month is None:
139 last_month = final
141 count = 0
142 for d in work_days:
143 ds = year_month_me(d)
144 if first_month <= ds <= last_month:
145 if ds < month or ds == month and d.day <= day:
146 count += 1
148 return count
151@no_type_check
152def remaining_workdays_count_of_year_in_between(work_days, month, day, first_month=None, last_month=None) -> int:
153 """Return the workday count of year from date for day (incl.) given first and last month."""
154 per = workdays_count_per_month(work_days)
155 if any(
156 (
157 month not in per,
158 first_month and first_month not in per,
159 last_month and last_month not in per,
160 )
161 ):
162 return 0
164 initial, final = closed_interval_months(work_days)
165 if first_month is None:
166 first_month = initial
167 if last_month is None:
168 last_month = final
170 count = 0
171 for d in work_days:
172 ds = year_month_me(d)
173 if first_month <= ds <= last_month:
174 if ds == month and d.day > day or ds > month:
175 count += 1
177 return count
180@no_type_check
181def workday(off_days: list[dti.date], cmd: str, date: str = '', strict: bool = False) -> Tuple[int, str]:
182 """Apply the effective rules to the given date (default today)."""
183 day = dti.datetime.strptime(date, DATE_FMT).date() if date else dti.date.today()
184 if strict:
185 if not off_days:
186 return 2, '- empty date range of configuration'
187 if not (off_days[0].year <= day.year < off_days[-1].year):
188 return 2, '- Day is not within year range of configuration'
189 else:
190 if cmd.startswith('explain'): 190 ↛ 193line 190 didn't jump to line 193, because the condition on line 190 was never false
191 print(f'- Day ({day}) is within date range of configuration')
193 if day not in off_days: 193 ↛ 197line 193 didn't jump to line 197, because the condition on line 193 was never false
194 if cmd.startswith('explain'):
195 print(f'- Day ({day}) is not a holiday')
196 else:
197 return 1, '- Day is a holiday.'
199 work_day = no_weekend(weekday(day))
200 if work_day:
201 if cmd.startswith('explain'):
202 print(f'- Day ({day}) is not a weekend')
203 else:
204 return 1, '- Day is weekend.'
206 return 0, ''
209@no_type_check
210def apply(
211 off_days: list[dti.date], working_hours: WorkingHoursType, cmd: str, day: str, strict: bool
212) -> Tuple[int, str]:
213 """Apply the effective rules to the current date and time."""
214 working_hours = working_hours if working_hours != (None, None) else DEFAULT_WORK_HOURS_CLOSED_INTERVAL
215 code, message = workday(off_days, cmd, date=str(day), strict=strict)
216 if code:
217 return code, message
218 hour = the_hour()
219 if working_hours[0] <= hour <= working_hours[1]: 219 ↛ 223line 219 didn't jump to line 223, because the condition on line 219 was never false
220 if cmd.startswith('explain'):
221 print(f'- At this hour ({hour}) is work time')
222 else:
223 return 1, f'- No worktime at hour({hour}).'
225 return 0, ''
228@no_type_check
229def load(cfg: CfgType) -> Tuple[int, str, list[dti.date], WorkingHoursType]:
230 """Load the configuration and return error, message and holidays as well as working hours list.
232 The holidays as well as non-default working hours will be ordered.
233 """
234 if not cfg:
235 return 0, 'empty configuration, using default', [], DEFAULT_WORK_HOURS_MARKER
237 try:
238 model = api.Arbejdstimer(**cfg)
239 except ValidationError as err:
240 return 2, str(err), [], (None, None)
242 holidays_date_list = []
243 if model.holidays:
244 holidays = model.model_dump()['holidays']
245 for nth, holiday in enumerate(holidays, start=1):
246 dates = holiday['at']
247 if len(dates) == 1:
248 holidays_date_list.append(dates[0])
249 elif len(dates) == 2:
250 data = sorted(dates)
251 start, end = [data[n] for n in (0, 1)]
252 current = start
253 holidays_date_list.append(current)
254 while current < end:
255 current += dti.timedelta(days=1)
256 holidays_date_list.append(current)
257 else:
258 for a_date in dates:
259 holidays_date_list.append(a_date)
261 working_hours = DEFAULT_WORK_HOURS_MARKER
262 if model.working_hours:
263 working_hours = tuple(sorted(model.working_hours.model_dump()))
264 return 0, '', sorted(holidays_date_list), working_hours
267@no_type_check
268def workdays_from_config(cfg: CfgType, day=None) -> list[dti.date]:
269 """Ja, ja, ja."""
270 error, _, holidays, _ = load(cfg)
271 return [] if error else workdays(holidays, days=days_of_year(day))
274def verify_request(argv: Union[CmdType, None]) -> Tuple[int, str, CmdType]:
275 """Fail with grace."""
276 err = ('', '', '', False)
277 if not argv or len(argv) != 4:
278 return 2, 'received wrong number of arguments', err
280 command, date, config, strict = argv
282 if command not in ('explain', 'explain_verbatim', 'now'):
283 return 2, 'received unknown command', err
285 if not config:
286 return 2, 'configuration missing', err
288 config_path = pathlib.Path(str(config))
289 if not config_path.is_file():
290 return 1, f'config ({config_path}) is no file', err
291 if not ''.join(config_path.suffixes).lower().endswith('.json'):
292 return 1, 'config has not .json extension', err
294 return 0, '', argv
297def main(argv: Union[CmdType, None] = None) -> int:
298 """Drive the lookup."""
299 error, message, strings = verify_request(argv)
300 if error:
301 print(message, file=sys.stderr)
302 return error
304 command, date, config, strict = strings
306 configuration = load_config(config)
307 error, message, holidays, working_hours = load(configuration)
308 if error:
309 if command.startswith('explain'):
310 print('Configuration file failed to parse (INVALID)')
311 print(message, file=sys.stderr)
312 return error
314 if command.startswith('explain'):
315 print(f'read valid configuration from ({config})')
316 if command == 'explain_verbatim':
317 lines = json.dumps(configuration, indent=2).split('\n')
318 line_count = len(lines)
319 counter_width = len(str(line_count))
320 print(f'configuration has {line_count} line{"" if line_count == 1 else "s"} of (indented) JSON content:')
321 for line, content in enumerate(lines, start=1):
322 print(f' {line:>{counter_width + 1}} | {content}')
323 if command == 'explain_verbatim':
324 if strict:
325 print('detected strict mode (queries outside of year frame from config will fail)')
326 else:
327 print('detected non-strict mode (no constraints on year frame from config)')
329 if command == 'explain':
330 print(f'consider {len(holidays)} holidays:')
331 elif command == 'explain_verbatim':
332 print('effective configuration:')
333 if holidays:
334 print(f'- given {len(holidays)} holidays within [{holidays[0]}, {holidays[-1]}]:')
335 for holiday in holidays:
336 print(f' + {holiday}')
337 else:
338 print(f'- no holidays defined in ({config}):')
339 print('- working hours:')
340 if working_hours != DEFAULT_WORK_HOURS_MARKER:
341 print(f' + [{working_hours[0]}, {working_hours[1]}] (from configuration)')
342 else:
343 effective_range = DEFAULT_WORK_HOURS_CLOSED_INTERVAL
344 print(f' + [{effective_range[0]}, {effective_range[1]}] (application default)')
345 print('evaluation:')
347 if strict:
348 print('detected strict mode (queries outside of year frame from config will fail)')
350 error, message = apply(holidays, working_hours, command, date, strict)
351 if error:
352 if command.startswith('explain'):
353 print(message, file=sys.stdout)
354 return error
356 return 0