Skip to content
Snippets Groups Projects
Commit 61922712 authored by Felix Blanke's avatar Felix Blanke
Browse files

Detach merge_blocks from Event object

parent fd6adb0e
No related branches found
No related tags found
1 merge request!5Refactor main
Pipeline #266716 passed
...@@ -48,6 +48,69 @@ class OptimizerTimeslot: ...@@ -48,6 +48,69 @@ class OptimizerTimeslot:
TimeslotBlock = list[OptimizerTimeslot] TimeslotBlock = list[OptimizerTimeslot]
def merge_blocks(
blocks: Iterable[TimeslotBlock]
) -> Iterable[TimeslotBlock]:
"""Merge iterable of blocks together.
The timeslots of all blocks are grouped into maximal blocks.
Timeslots with the same start and end are identified with each other
and merged (cf `OptimizerTimeslot.merge`).
Throws a ValueError if any timeslots are overlapping but do not
share the same start and end, i.e. partial overlap is not allowed.
:param blocks: iterable of blocks to merge.
:return: iterable of merged blocks.
:rtype: iterable over lists of OptimizerTimeslot objects
"""
if not blocks:
return []
# flatten timeslot iterables to single chain
timeslot_chain = itertools.chain.from_iterable(blocks)
# sort timeslots according to start
timeslots = sorted(
timeslot_chain,
key=lambda slot: slot.avail.start
)
if not timeslots:
return []
all_blocks = []
current_block = [timeslots[0]]
timeslots = timeslots[1:]
for slot in timeslots:
if current_block and slot.avail.overlaps(current_block[-1].avail, strict=True):
if (
slot.avail.start == current_block[-1].avail.start
and slot.avail.end == current_block[-1].avail.end
):
# the same timeslot -> merge
current_block[-1] = current_block[-1].merge(slot)
else:
# partial overlap of interiors -> not supported
# TODO: Show comprehensive message in production
raise ValueError(
"Partially overlapping timeslots are not supported!"
f" ({current_block[-1].avail.simplified}, {slot.avail.simplified})"
)
elif not current_block or slot.avail.overlaps(current_block[-1].avail, strict=False):
# only endpoints in intersection -> same block
current_block.append(slot)
else:
# no overlap at all -> new block
all_blocks.append(current_block)
current_block = [slot]
if current_block:
all_blocks.append(current_block)
return all_blocks
class Event(models.Model): class Event(models.Model):
""" """
An event supplies the frame for all Aks. An event supplies the frame for all Aks.
...@@ -319,68 +382,6 @@ class Event(models.Model): ...@@ -319,68 +382,6 @@ class Event(models.Model):
constraints=category_constraints, constraints=category_constraints,
) )
def merge_blocks(
self, blocks: Iterable[TimeslotBlock]
) -> Iterable[TimeslotBlock]:
"""Merge iterable of blocks together.
The timeslots of all blocks are grouped into maximal blocks.
Timeslots with the same start and end are identified with each other
and merged (cf `OptimizerTimeslot.merge`).
Throws a ValueError if any timeslots are overlapping but do not
share the same start and end, i.e. partial overlap is not allowed.
:param blocks: iterable of blocks to merge.
:return: iterable of merged blocks.
:rtype: iterable over lists of OptimizerTimeslot objects
"""
if not blocks:
return []
# flatten timeslot iterables to single chain
timeslot_chain = itertools.chain.from_iterable(blocks)
# sort timeslots according to start
timeslots = sorted(
timeslot_chain,
key=lambda slot: slot.avail.start
)
if not timeslots:
return []
all_blocks = []
current_block = [timeslots[0]]
timeslots = timeslots[1:]
for slot in timeslots:
if current_block and slot.avail.overlaps(current_block[-1].avail, strict=True):
if (
slot.avail.start == current_block[-1].avail.start
and slot.avail.end == current_block[-1].avail.end
):
# the same timeslot -> merge
current_block[-1] = current_block[-1].merge(slot)
else:
# partial overlap of interiors -> not supported
# TODO: Show comprehensive message in production
raise ValueError(
"Partially overlapping timeslots are not supported!"
f" ({current_block[-1].avail.simplified}, {slot.avail.simplified})"
)
elif not current_block or slot.avail.overlaps(current_block[-1].avail, strict=False):
# only endpoints in intersection -> same block
current_block.append(slot)
else:
# no overlap at all -> new block
all_blocks.append(current_block)
current_block = [slot]
if current_block:
all_blocks.append(current_block)
return all_blocks
def schedule_from_json(self, schedule: str) -> None: def schedule_from_json(self, schedule: str) -> None:
"""Load AK schedule from a json string. """Load AK schedule from a json string.
...@@ -395,7 +396,7 @@ class Event(models.Model): ...@@ -395,7 +396,7 @@ class Event(models.Model):
timeslot_dict = { timeslot_dict = {
timeslot.idx: timeslot timeslot.idx: timeslot
for block in self.merge_blocks(self.default_time_slots(slots_in_an_hour=slots_in_an_hour)) for block in merge_blocks(self.default_time_slots(slots_in_an_hour=slots_in_an_hour))
for timeslot in block for timeslot in block
} }
......
...@@ -9,7 +9,7 @@ from django.views.generic import ListView, DetailView ...@@ -9,7 +9,7 @@ from django.views.generic import ListView, DetailView
from AKModel.availability.models import Availability from AKModel.availability.models import Availability
from AKModel.metaviews.admin import AdminViewMixin, FilterByEventSlugMixin, EventSlugMixin, IntermediateAdminView, \ from AKModel.metaviews.admin import AdminViewMixin, FilterByEventSlugMixin, EventSlugMixin, IntermediateAdminView, \
IntermediateAdminActionView IntermediateAdminActionView
from AKModel.models import AKRequirement, AKSlot, Event, AKOrgaMessage, AK, Room, AKOwner from AKModel.models import AKRequirement, AKSlot, Event, AKOrgaMessage, AK, Room, AKOwner, merge_blocks
class AKRequirementOverview(AdminViewMixin, FilterByEventSlugMixin, ListView): class AKRequirementOverview(AdminViewMixin, FilterByEventSlugMixin, ListView):
...@@ -110,7 +110,7 @@ class AKJSONExportView(AdminViewMixin, FilterByEventSlugMixin, ListView): ...@@ -110,7 +110,7 @@ class AKJSONExportView(AdminViewMixin, FilterByEventSlugMixin, ListView):
if (values := AKSlot.objects.select_related().filter(ak__pk=ak_id, fixed=True)).exists() if (values := AKSlot.objects.select_related().filter(ak__pk=ak_id, fixed=True)).exists()
} }
for block in self.event.merge_blocks(self.event.default_time_slots(slots_in_an_hour=SLOTS_IN_AN_HOUR)): for block in merge_blocks(self.event.default_time_slots(slots_in_an_hour=SLOTS_IN_AN_HOUR)):
current_block = [] current_block = []
for timeslot in block: for timeslot in block:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment