Compare revisions

image: python:3.9

image: python:3.10

services:

  - mysql

        if not other.overlaps(self, strict=False):

            raise Exception('Only overlapping Availabilities can be merged.')

        return Availability(

        avail = Availability(

            start=min(self.start, other.start), end=max(self.end, other.end)

        )

        if self.event == other.event:

            avail.event = self.event

        return avail

    def __or__(self, other: 'Availability') -> 'Availability':

        """Performs the merge operation: ``availability1 | availability2``"""

        if not other.overlaps(self, False):

            raise Exception('Only overlapping Availabilities can be intersected.')

        return Availability(

        avail = Availability(

            start=max(self.start, other.start), end=min(self.end, other.end)

        )

        if self.event == other.event:

            avail.event = self.event

        return avail

    def __and__(self, other: 'Availability') -> 'Availability':

        """Performs the intersect operation: ``availability1 &

                f'{self.end.astimezone(self.event.timezone).strftime("%a %H:%M")}')

    @classmethod

    def with_event_length(cls, event, person=None, room=None, ak=None, ak_category=None):

    def with_event_length(

        cls,

        event: Event,

        person: AKOwner | None = None,

        room: Room | None = None,

        ak: AK | None = None,

        ak_category: AKCategory | None = None,

    ) -> "Availability":

        """

        Create an availability covering exactly the time between event start and event end.

        Can e.g., be used to create default availabilities.

        return Availability(start=timeframe_start, end=timeframe_end, event=event, person=person,

                                    room=room, ak=ak, ak_category=ak_category)

    @classmethod

    def is_event_covered(cls, event: Event, availabilities: List['Availability']) -> bool:

        """Check if list of availibilities cover whole event.

        :param event: event to check.

        :param availabilities: availabilities to check.

        :return: whether the availabilities cover full event.

        :rtype: bool

        """

        # NOTE: Cannot use `Availability.with_event_length` as its end is the

        #       event end + 1 day

        full_event = Availability(event=event, start=event.start, end=event.end)

        avail_union = Availability.union(availabilities)

        return any(avail.contains(full_event) for avail in avail_union)

    class Meta:

        verbose_name = _('Availability')

        verbose_name_plural = _('Availabilities')

        # Filter possible values for m2m when event is specified

        if hasattr(self.instance, "event") and self.instance.event is not None:

            self.fields["properties"].queryset = AKRequirement.objects.filter(event=self.instance.event)

class JSONScheduleImportForm(AdminIntermediateForm):

    """Form to import an AK schedule from a json file."""

    json_data = forms.CharField(

        required=True,

        widget=forms.Textarea,

        label=_("JSON data"),

        help_text=_("JSON data from the scheduling solver"),

    )

import decimal

import itertools

from datetime import timedelta

import json

import math

from dataclasses import dataclass

from datetime import datetime, timedelta

from typing import Iterable, Generator

from django.db import models

from django.apps import apps

from django.db.models import Count

from django.urls import reverse_lazy

from django.utils import timezone

from django.utils.datetime_safe import datetime

from django.utils.text import slugify

from django.utils.translation import gettext_lazy as _

from simple_history.models import HistoricalRecords

from timezone_field import TimeZoneField

@dataclass

class OptimizerTimeslot:

    """Class describing a discrete timeslot. Used to interface with an optimizer."""

    avail: "Availability"

    """The availability object corresponding to this timeslot."""

    idx: int

    """The unique index of this optimizer timeslot."""

    constraints: set[str]

    """The set of time constraints fulfilled by this object."""

    def merge(self, other: "OptimizerTimeslot") -> "OptimizerTimeslot":

        """Merge with other OptimizerTimeslot.

        Creates a new OptimizerTimeslot object.

        Its availability is constructed by merging the availabilities of self and other,

        its constraints by taking the union of both constraint sets.

        As an index, the index of self is used.

        """

        avail = self.avail.merge_with(other.avail)

        constraints = self.constraints.union(other.constraints)

        return OptimizerTimeslot(

            avail=avail, idx=self.idx, constraints=constraints

        )

    def __repr__(self) -> str:

        return f"({self.avail.simplified}, {self.idx}, {self.constraints})"

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):

    """

    An event supplies the frame for all Aks.

                .filter(availabilities__count=0, owners__count__gt=0)

                )

    def _generate_slots_from_block(

        self,

        start: datetime,

        end: datetime,

        slot_duration: timedelta,

        slot_index: int = 0,

        constraints: set[str] | None = None,

    ) -> Generator[TimeslotBlock, None, int]:

        """Discretize a time range into timeslots.

        Uses a uniform discretization into discrete slots of length `slot_duration`,

        starting at `start`. No incomplete timeslots are generated, i.e.

        if (`end` - `start`) is not a whole number multiple of `slot_duration`

        then the last incomplete timeslot is dropped.

        :param start: Start of the time range.

        :param end: Start of the time range.

        :param slot_duration: Duration of a single timeslot in the discretization.

        :param slot_index: index of the first timeslot. Defaults to 0.

        :yield: Block of optimizer timeslots as the discretization result.

        :ytype: list of OptimizerTimeslot

        :return: The first slot index after the yielded blocks, i.e.

            `slot_index` + total # generated timeslots

        :rtype: int

        """

        # local import to prevent cyclic import

        # pylint: disable=import-outside-toplevel

        from AKModel.availability.models import Availability

        current_slot_start = start

        previous_slot_start: datetime | None = None

        if constraints is None:

            constraints = set()

        current_block = []

        room_availabilities = list({

            availability

            for room in Room.objects.filter(event=self)

            for availability in room.availabilities.all()

        })

        while current_slot_start + slot_duration <= end:

            slot = Availability(

                event=self,

                start=current_slot_start,

                end=current_slot_start + slot_duration,

            )

            if any((availability.contains(slot) for availability in room_availabilities)):

                # no gap in a block

                if (

                    previous_slot_start is not None

                    and previous_slot_start + slot_duration < current_slot_start

                ):

                    yield current_block

                    current_block = []

                current_block.append(

                    OptimizerTimeslot(avail=slot, idx=slot_index, constraints=constraints)

                )

                previous_slot_start = current_slot_start

            slot_index += 1

            current_slot_start += slot_duration

        if current_block:

            yield current_block

        return slot_index

    def uniform_time_slots(self, *, slots_in_an_hour: float = 1.0) -> Iterable[TimeslotBlock]:

        """Uniformly discretize the entire event into blocks of timeslots.

        Discretizes entire event uniformly. May not necessarily result in a single block

        as slots with no room availability are dropped.

        :param slots_in_an_hour: The percentage of an hour covered by a single slot.

            Determines the discretization granularity.

        :yield: Block of optimizer timeslots as the discretization result.

        :ytype: list of OptimizerTimeslot

        """

        all_category_constraints = AKCategory.create_category_constraints(

            AKCategory.objects.filter(event=self).all()

        )

        yield from self._generate_slots_from_block(

            start=self.start,

            end=self.end,

            slot_duration=timedelta(hours=1.0 / slots_in_an_hour),

            constraints=all_category_constraints,

        )

    def default_time_slots(self, *, slots_in_an_hour: float = 1.0) -> Iterable[TimeslotBlock]:

        """Discretize all default slots into blocks of timeslots.

        In the discretization each default slot corresponds to one block.

        :param slots_in_an_hour: The percentage of an hour covered by a single slot.

            Determines the discretization granularity.

        :yield: Block of optimizer timeslots as the discretization result.

        :ytype: list of TimeslotBlock

        """

        slot_duration = timedelta(hours=1.0 / slots_in_an_hour)

        slot_index = 0

        for block_slot in DefaultSlot.objects.filter(event=self).order_by("start", "end"):

            category_constraints = AKCategory.create_category_constraints(

                block_slot.primary_categories.all()

            )

            slot_index = yield from self._generate_slots_from_block(

                start=block_slot.start,

                end=block_slot.end,

                slot_duration=slot_duration,

                slot_index=slot_index,

                constraints=category_constraints,

            )

    def discretize_timeslots(self, *, slots_in_an_hour: float = 1.0) -> Iterable[TimeslotBlock]:

        """"Choose discretization scheme.

        Uses default_time_slots if the event has any DefaultSlot, otherwise uniform_time_slots.

        :param slots_in_an_hour: The percentage of an hour covered by a single slot.

            Determines the discretization granularity.

        :yield: Block of optimizer timeslots as the discretization result.

        :ytype: list of TimeslotBlock

        """

        if DefaultSlot.objects.filter(event=self).exists():

            # discretize default slots if they exists

            yield from merge_blocks(self.default_time_slots(slots_in_an_hour=slots_in_an_hour))

        else:

            yield from self.uniform_time_slots(slots_in_an_hour=slots_in_an_hour)

    def schedule_from_json(self, schedule: str) -> None:

        """Load AK schedule from a json string.

        :param schedule: A string that can be decoded to json, describing

            the AK schedule. The json data is assumed to be constructed

            following the output specification of the KoMa conference optimizer, cf.

            https://github.com/Die-KoMa/ak-plan-optimierung/wiki/Input-&-output-format

        """

        schedule = json.loads(schedule)

        slots_in_an_hour = schedule["input"]["timeslots"]["info"]["duration"]

        timeslot_dict = {

            timeslot.idx: timeslot

            for block in self.discretize_timeslots(slots_in_an_hour=slots_in_an_hour)

            for timeslot in block

        }

        for scheduled_slot in schedule["scheduled_aks"]:

            slot = AKSlot.objects.get(id=int(scheduled_slot["ak_id"]))

            slot.room = Room.objects.get(id=int(scheduled_slot["room_id"]))

            scheduled_slot["timeslot_ids"] = list(map(int, scheduled_slot["timeslot_ids"]))

            start_timeslot = timeslot_dict[min(scheduled_slot["timeslot_ids"])].avail

            end_timeslot = timeslot_dict[max(scheduled_slot["timeslot_ids"])].avail

            slot.start = start_timeslot.start

            slot.duration = (end_timeslot.end - start_timeslot.start).total_seconds() / 3600.0

            slot.save()

class AKOwner(models.Model):

    """ An AKOwner describes the person organizing/holding an AK.

    def __str__(self):

        return self.name

    @staticmethod

    def create_category_constraints(categories: Iterable["AKCategory"]) -> set[str]:

        """Create a set of constraint strings from an AKCategory iterable.

        :param categories: The iterable of categories to derive the constraint strings from.

        :return: A set of category constraint strings, i.e. strings of the form

            'availability-cat-<cat.name>'.

        :rtype: set of strings.

        """

        return {

            f"availability-cat-{cat.name}"

            for cat in categories

        }

class AKTrack(models.Model):

    """ An AKTrack describes a set of semantically related AKs.

    def __str__(self):

        return self.title

    def as_json(self) -> str:

        """Return a json string representation of this room object.

        :return: The json string representation is constructed

            following the input specification of the KoMa conference optimizer, cf.

            https://github.com/Die-KoMa/ak-plan-optimierung/wiki/Input-&-output-format

        :rtype: str

        """

        # local import to prevent cyclic import

        # pylint: disable=import-outside-toplevel

        from AKModel.availability.models import Availability

        # check if room is available for the whole event

        # -> no time constraint needs to be introduced

        if Availability.is_event_covered(self.event, self.availabilities.all()):

            time_constraints = []

        else:

            time_constraints = [f"availability-room-{self.pk}"]

        data = {

            "id": str(self.pk),

            "info": {

                "name": self.name,

            },

            "capacity": self.capacity,

            "fulfilled_room_constraints": [constraint.name

                                           for constraint in self.properties.all()],

            "time_constraints": time_constraints

        }

        data["fulfilled_room_constraints"].append(f"fixed-room-{self.pk}")

        if not any(constr.startswith("proxy") for constr in data["fulfilled_room_constraints"]):

            data["fulfilled_room_constraints"].append("no-proxy")

        return json.dumps(data)

class AKSlot(models.Model):

    """ An AK Mapping matches an AK to a room during a certain time.

            self.duration = min(self.duration, event_duration_hours)

        super().save(force_insert, force_update, using, update_fields)

    def as_json(self) -> str:

        """Return a json string representation of the AK object of this slot.

        :return: The json string representation is constructed

            following the input specification of the KoMa conference optimizer, cf.

            https://github.com/Die-KoMa/ak-plan-optimierung/wiki/Input-&-output-format

        :rtype: str

        """

        # local import to prevent cyclic import

        # pylint: disable=import-outside-toplevel

        from AKModel.availability.models import Availability

        # check if ak resp. owner is available for the whole event

        # -> no time constraint needs to be introduced

        if self.fixed and self.start is not None:

            ak_time_constraints = [f"fixed-akslot-{self.id}"]

        elif Availability.is_event_covered(self.event, self.ak.availabilities.all()):

            ak_time_constraints = [f"availability-ak-{self.ak.pk}"]

        else:

            ak_time_constraints = []

        def _owner_time_constraints(owner: AKOwner):

            if Availability.is_event_covered(self.event, owner.availabilities.all()):

                return []

            return [f"availability-person-{owner.pk}"]

        conflict_slots = AKSlot.objects.filter(ak__in=self.ak.conflicts.all())

        dependency_slots = AKSlot.objects.filter(ak__in=self.ak.prerequisites.all())

        ceil_offet_eps = decimal.Decimal(1e-4)

        # self.slots_in_an_hour is set in AKJSONExportView

        data = {

            "id": str(self.pk),

            "duration": math.ceil(self.duration * self.slots_in_an_hour - ceil_offet_eps),

            "properties": {

                "conflicts": [str(conflict.pk) for conflict in conflict_slots.all()],

                "dependencies": [str(dep.pk) for dep in dependency_slots.all()],

            },

            "room_constraints": [constraint.name

                                 for constraint in self.ak.requirements.all()],

            "time_constraints": ["resolution"] if self.ak.reso else [],

            "info": {

                "name": self.ak.name,

                "head": ", ".join([str(owner)

                                   for owner in self.ak.owners.all()]),

                "description": self.ak.description,

                "reso": self.ak.reso,

                "duration_in_hours": float(self.duration),

                },

            }

        data["time_constraints"].extend(ak_time_constraints)

        for owner in self.ak.owners.all():

            data["time_constraints"].extend(_owner_time_constraints(owner))

        if self.ak.category:

            category_constraints = AKCategory.create_category_constraints([self.ak.category])

            data["time_constraints"].extend(category_constraints)

        if self.fixed and self.room is not None:

            data["room_constraints"].append(f"fixed-room-{self.room.pk}")

        if not any(constr.startswith("proxy") for constr in data["room_constraints"]):

            data["room_constraints"].append("no-proxy")