pingu_98/cara
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Two improved algorithms for present_interval. About to remove the more complex one.

Browse source
This commit is contained in:
Phil Elson 2020-12-17 20:49:30 +01:00
parent 4532fdce59
commit cf535c954a
2 changed files with 191 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

201
cara/apps/calculator/model_generator.py
View file

@ -1,5 +1,6 @@
from dataclasses import dataclass
import html
import logging
import typing
import numpy as np
@ -8,6 +9,9 @@ from cara import models
from cara import data
LOG = logging.getLogger(__name__)
@dataclass
class FormData:
# Number of minutes after 00:00
@ -307,6 +311,12 @@ class FormData:
break_times.append((begin, end))
return tuple(break_times)
def lunch_break_times(self) -> typing.Tuple[typing.Tuple[int, int]]:
result = []
if self.lunch_option:
result.append((self.lunch_start, self.lunch_finish))
return tuple(result)
def coffee_break_times(self) -> typing.Tuple[typing.Tuple[int, int]]:
if not self.coffee_breaks:
return ()
@ -327,16 +337,179 @@ class FormData:
breaks = self._compute_breaks_in_interval(self.activity_start, self.activity_finish, self.coffee_breaks)
return breaks
def present_interval(self, start, finish) -> models.Interval:
leave_times = []
enter_times = []
if self.lunch_option:
leave_times.append(self.lunch_start)
enter_times.append(self.lunch_finish)
def present_interval(
self,
start: int,
finish: int,
breaks: typing.Tuple[typing.Tuple[int, int], ...] = None,
) -> models.Interval:
"""
Calculate the presence interval given the start and end times (in minutes), and
a number of monotonic, non-overlapping, but potentially unsorted, breaks (also in minutes).
"""
if not breaks:
breaks = []
if self.lunch_option:
breaks.append((self.lunch_start, self.lunch_finish))
for coffee_start, coffee_end in self.coffee_break_times():
breaks.append((coffee_start, coffee_end))
if not breaks:
# If there are no breaks, the interval is the start and end.
return models.SpecificInterval(((start, finish),))
# Order the breaks by their start-time, and ensure that they are monotonic
# and that the start of one break happens after the end of another.
breaks = sorted(breaks, key=lambda break_pair: break_pair[0])
for break_start, break_end in breaks:
if break_start >= break_end:
raise ValueError("Break ends before it begins.")
prev_break_end = breaks[0][1]
for break_start, break_end in breaks[1:]:
if prev_break_end >= break_start:
raise ValueError(f"A break starts before another ends ({break_start}, {break_end}, {prev_break_end}).")
prev_break_end = break_end
present_intervals = []
def hours2time(hours: float):
# Convert times like 14.5 to strings, like "14:30"
return f"{int(np.floor(hours)):02d}:{int(np.round((hours % 1) * 60)):02d}"
current_time = start
LOG.debug(f"starting time march at {hours2time(current_time/60)} to {hours2time(finish/60)}")
# As we step through the breaks. For each break there are 6 important cases
# we must cover. Let S=start; E=end; Bs=Break start; Be=Break end:
# 1. The interval is entirely before the break. S < E <= Bs < Be
# 2. The interval straddles the start of the break. S < Bs < E <= Be
# 3. The break is entirely inside the interval. S < Bs < Be < E
# 4. The interval is entirely inside the break. Bs <= S < E <= Be
# 5. The interval straddles the end of the break. Bs < S <= Be <= E
# 6. The interval is entirely after the break. Bs < Be <= S < E
for current_break in breaks:
LOG.debug(f"handling break {hours2time(current_break[0]/60)}-{hours2time(current_break[1]/60)} "
f" (current time: {hours2time(current_time/60)})")
if current_time == current_break[0]:
# For the special case of the marching time being exactly at
# the start of the next break, skip this break and move the
# current time to the end of the break.
current_time = current_break[1]
LOG.debug(f"skipping break: it starts at the same time as the current time")
continue
if current_time >= finish:
break
break_s, break_e = current_break[0], current_break[1]
case1 = finish <= break_s
case2 = current_time < break_s < finish <= break_e
case3 = current_time <= break_s < break_e <= finish
case4 = break_s <= current_time < finish <= break_e
case5 = break_s < current_time <= break_e <= finish
case6 = break_e <= current_time
# LOG.debug(current_time <= break_s)
# LOG.debug(break_e <= finish)
# LOG.debug(", ".join(str(i) for i in [case1, case2, case3, case4, case5, case6]))
assert any([case1, case2, case3, case4, case5, case6])
if case1:
LOG.debug(f"case 1: interval entirely before break")
present_intervals.append((current_time / 60, finish / 60))
current_time = finish
elif case2:
LOG.debug(f"case 2: interval straddles start of break")
present_intervals.append((current_time / 60, break_s / 60))
current_time = break_e
elif case3:
LOG.debug(f"case 3: break entirely inside interval")
# We add the bit before the break, but not the bit afterwards,
# as it may hit another break.
present_intervals.append((current_time / 60, break_s / 60))
current_time = break_e
elif case4:
LOG.debug(f"case 4: interval entirely inside break")
current_time = finish
elif case5:
LOG.debug(f"case 4: interval straddles end of break")
current_time = break_e
elif case6:
LOG.debug(f"case 4: interval entirely after the break")
continue
current_time_before_break = current_time < current_break[0]
current_time_inside_break = current_break[0] <= current_time < current_break[1]
finish_inside_break = current_break[0] <= finish < current_break[1]
break_ends_after_finish = finish < current_break[1]
break_starts_before_finish = current_break[0] < finish
if current_time_inside_break:
# If the current time is in the break, we mustn't add it.
# If the break ends after the finish time then we move the current
# time to the finish, and no more breaks will be added.
# Otherwise we move the current time forwards to the end of the
# break.
if break_ends_after_finish:
LOG.debug(f"skipping break: start is in the break and the finish occurs before the end of the break")
current_time = finish
else:
LOG.debug(f"skipping break: finishes before the current time")
current_time = current_break[1]
continue
elif current_time_before_break:
presence_start = current_time
presence_end = current_break[0]
if break_starts_before_finish:
#
current_time = current_break[1]
else:
# If we are supposed to finish before the end of this break
# then move the finish forwards to the start of the break.
if not finish_inside_break:
LOG.debug(f"break covers the finish time: moving the finish to the beginning of the break")
presence_end = finish
current_time = finish
if presence_start < presence_end:
# If the break starts within the current time and the end time, then we include it.
LOG.debug(f"break added {hours2time(presence_start/60)} {hours2time(presence_end/60)}")
present_intervals.append((presence_start / 60, presence_end / 60))
else:
LOG.debug("skipping break: it starts after the current time")
if current_time < finish:
LOG.debug("end added")
present_intervals.append((current_time / 60, finish / 60))
return models.SpecificInterval(tuple(present_intervals))
# If the start happens before the first break, then insert a break
# just before the start.
first_break = breaks[0]
if start < first_break[0]:
breaks.insert(0, (start, start))
# If the end happens after the last break, then insert a break
# just after the finish.
last_break = breaks[-1]
if finish > last_break[1]:
breaks.append((finish, finish))
# We now know that there are breaks before and after our desired interval, so now step through each break,
# and determine the periods that we have presence.
prev_break = breaks[0]
for break_start, break_end in breaks[1:]:
if prev_break[0] >= start and break_end <= finish:
present_intervals.append((prev_break[0] / 60, break_end / 60))
return models.SpecificInterval(tuple(present_intervals))
for coffee_start, coffee_end in self.coffee_break_times():
leave_times.append(coffee_start)
enter_times.append(coffee_end)
# These lists represent the times where the infected person leaves or enters the room, respectively, sorted in
# reverse order. Note that these lists allows the person to "leave" when they should not even be present in the
@ -390,10 +563,16 @@ class FormData:
return models.SpecificInterval(tuple(present_intervals))
def infected_present_interval(self) -> models.Interval:
return self.present_interval(self.infected_start, self.infected_finish)
return self.present_interval(
self.infected_start, self.infected_finish,
breaks=self.lunch_break_times() + self.coffee_break_times(),
)
def exposed_present_interval(self) -> models.Interval:
return self.present_interval(self.activity_start, self.activity_finish)
return self.present_interval(
self.activity_start, self.activity_finish,
breaks=self.lunch_break_times() + self.coffee_break_times(),
)
def build_expiration(expiration_definition) -> models.Expiration:

2
cara/tests/apps/calculator/test_model_generator.py
View file

@ -164,7 +164,7 @@ def test_exposed_present_intervals_starting_with_lunch(baseline_form):
baseline_form.activity_start = baseline_form.lunch_start = 13 * 60
baseline_form.activity_finish = 18 * 60
baseline_form.lunch_finish = 14 * 60
correct = ((14.0, 18.0),)
correct = ((14.0, 18.0), )
assert baseline_form.exposed_present_interval().present_times == correct