src/make_queue/models/machine.py
from abc import abstractmethod
from datetime import datetime, timedelta
from django.contrib.auth.models import AnonymousUser
from django.db import models
from django.db.models import F, Prefetch, Q
from django.db.models.functions import Lower
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from django_hosts import reverse
from simple_history.models import HistoricalRecords
from users.models import User
from util.validators import lowercase_slug_validator
from web.modelfields import URLTextField, UnlimitedCharField
from web.multilingual.modelfields import MultiLingualRichTextUploadingField, MultiLingualTextField
from .course import Printer3DCourse
class MachineTypeQuerySet(models.QuerySet):
def default_order_by(self) -> 'MachineTypeQuerySet[MachineType]':
return self.order_by('priority')
def prefetch_machines(self, *, machine_queryset=None, machines_attr_name: str) -> 'MachineTypeQuerySet[MachineType]':
"""
Returns a ``QuerySet`` where all the machine types' machines have been prefetched
and can be accessed through the attribute with the same name as ``machines_attr_name``.
"""
if machine_queryset is None:
machine_queryset = Machine.objects.all()
return self.prefetch_related(
Prefetch('machines', queryset=machine_queryset, to_attr=machines_attr_name)
)
class MachineType(models.Model):
class UsageRequirement(models.TextChoices):
IS_AUTHENTICATED = 'AUTH', _("Only has to be logged in")
TAKEN_3D_PRINTER_COURSE = '3DPR', _("Taken the 3D printer course")
TAKEN_RAISE3D_PRINTERS_COURSE = "R3DP", _("Taken the course on Raise3D printers")
TAKEN_SLA_3D_PRINTER_COURSE = "SLAP", _("Taken the SLA 3D printer course")
name = MultiLingualTextField(unique=True)
cannot_use_text = MultiLingualTextField(blank=True)
usage_requirement = models.CharField(
choices=UsageRequirement.choices,
max_length=4,
default=UsageRequirement.IS_AUTHENTICATED,
verbose_name=_("usage requirement"),
)
has_stream = models.BooleanField(default=False)
priority = models.IntegerField(
verbose_name=_("priority"),
help_text=_("The machine types are sorted ascending by this value."),
)
objects = MachineTypeQuerySet.as_manager()
class Meta:
ordering = ('priority',)
def __str__(self):
return str(self.name)
def can_user_use(self, user: User):
match self.usage_requirement:
case self.UsageRequirement.IS_AUTHENTICATED:
return user.is_authenticated
case self.UsageRequirement.TAKEN_3D_PRINTER_COURSE:
return self.can_use_3d_printer(user)
case self.UsageRequirement.TAKEN_RAISE3D_PRINTERS_COURSE:
return self.can_use_raise3d_printer(user)
case self.UsageRequirement.TAKEN_SLA_3D_PRINTER_COURSE:
return self.can_use_sla_printer(user)
case _:
return False
@staticmethod
def can_use_3d_printer(user: User | AnonymousUser):
if not user.is_authenticated:
return False
if hasattr(user, 'printer_3d_course'):
return True
if Printer3DCourse.objects.filter(username=user.username).exists():
course_registration = Printer3DCourse.objects.get(username=user.username)
course_registration.user = user
course_registration.save()
return True
return user.has_perm('make_queue.add_reservation') # This will typically only be the case for superusers
@staticmethod
def can_use_raise3d_printer(user: User | AnonymousUser):
if not user.is_authenticated:
return False
if Printer3DCourse.objects.filter(user=user).exists():
course_registration = Printer3DCourse.objects.get(user=user)
return course_registration.raise3d_course
if Printer3DCourse.objects.filter(username=user.username).exists():
course_registration = Printer3DCourse.objects.get(username=user.username)
course_registration.user = user
course_registration.save()
return course_registration.raise3d_course
return False
# TODO: reduce code duplication between this and the two methods above
@staticmethod
def can_use_sla_printer(user: User | AnonymousUser):
if not user.is_authenticated:
return False
if Printer3DCourse.objects.filter(user=user).exists():
course_registration = Printer3DCourse.objects.get(user=user)
return course_registration.sla_course
if Printer3DCourse.objects.filter(username=user.username).exists():
course_registration = Printer3DCourse.objects.get(username=user.username)
course_registration.user = user
course_registration.save()
return course_registration.sla_course
return False
class MachineQuerySet(models.QuerySet):
def visible_to(self, user: User) -> 'MachineQuerySet[Machine]':
if user.has_perm('internal.is_internal'):
return self.all()
exclude_query = Q(internal=True)
# Machines that require the SLA course should not be visible to non-internal users who have not taken the SLA course
if not hasattr(user, 'printer_3d_course') or not user.printer_3d_course.sla_course:
exclude_query |= Q(machine_type__usage_requirement=MachineType.UsageRequirement.TAKEN_SLA_3D_PRINTER_COURSE)
return self.exclude(exclude_query)
def default_order_by(self) -> 'MachineQuerySet[Machine]':
return self.order_by(
'machine_type__priority',
F('priority').asc(nulls_last=True),
Lower('name'),
)
class Machine(models.Model):
class Status(models.TextChoices):
RESERVED = 'R', _("Reserved")
AVAILABLE = 'F', _("Available")
IN_USE = 'I', _("In use")
OUT_OF_ORDER = 'O', _("Out of order")
MAINTENANCE = 'M', _("Maintenance")
STATUS_CHOICES_DICT = dict(Status.choices)
name = UnlimitedCharField(unique=True, verbose_name=_("name"))
stream_name = models.CharField(
blank=True,
max_length=50,
default="",
validators=[lowercase_slug_validator],
verbose_name=_("stream name"),
help_text=_("Used for connecting to the machine's stream."),
)
machine_model = UnlimitedCharField(verbose_name=_("machine model"))
machine_type = models.ForeignKey(
to=MachineType,
on_delete=models.PROTECT,
related_name='machines',
verbose_name=_("machine type"),
)
location = UnlimitedCharField(verbose_name=_("location"))
location_url = URLTextField(verbose_name=_("location URL"))
internal = models.BooleanField(default=False, verbose_name=_("internal"),
help_text=_("If selected, the machine will only be visible to and reservable by MAKE members."))
status = models.CharField(choices=Status.choices, max_length=2, default=Status.AVAILABLE, verbose_name=_("status"))
info_message = models.TextField(blank=True, verbose_name=_("info message"), help_text=_(
"Information that's useful to know before using the machine, e.g. the filament that the 3D printer uses,"
" the needle that's currently inserted in the sewing machine, or just the machine's current state/“mood” (emojis are allowed 🤠)."
))
info_message_date = models.DateTimeField(blank=True, default=timezone.localtime, verbose_name=_("time the info message was changed"))
priority = models.IntegerField(
null=True,
blank=True,
verbose_name=_("priority"),
help_text=_("If specified, the machines are sorted ascending by this value."),
)
notes = models.TextField(blank=True, verbose_name=_("notes"), help_text=_("This is only for internal use and is not displayed anywhere."))
last_modified = models.DateTimeField(auto_now=True, verbose_name=_("last modified"))
objects = MachineQuerySet.as_manager()
history = HistoricalRecords(excluded_fields=['status', 'info_message_date', 'priority', 'last_modified'])
def __str__(self):
return f"{self.name} - {self.machine_model}"
def get_absolute_url(self):
return reverse('machine_detail', args=[self.pk])
def get_next_reservation(self):
return self.reservations.filter(start_time__gt=timezone.now()).order_by('start_time').first()
@abstractmethod
def can_user_use(self, user):
return self.machine_type.can_user_use(user)
def reservations_in_period(self, start_time: datetime, end_time: datetime):
return (self.reservations.filter(start_time__lte=start_time, end_time__gt=start_time)
| self.reservations.filter(start_time__gte=start_time, end_time__lte=end_time)
| self.reservations.filter(start_time__lt=end_time, start_time__gt=start_time, end_time__gte=end_time))
def get_status(self):
if self.status in (self.Status.OUT_OF_ORDER, self.Status.MAINTENANCE):
return self.status
if self.reservations_in_period(timezone.now(), timezone.now() + timedelta(seconds=1)):
return self.Status.RESERVED
else:
return self.Status.AVAILABLE
def _get_FIELD_display(self, field):
if field.attname == 'status':
return self.STATUS_CHOICES_DICT[self.get_status()]
return super()._get_FIELD_display(field)
def reservable_status_display_tuple(self) -> tuple[bool, str]:
return (
self.get_status() in {self.Status.AVAILABLE, self.Status.RESERVED, self.Status.IN_USE},
self.get_status_display(),
)
class MachineUsageRule(models.Model):
"""
Allows for specification of rules for each type of machine.
"""
machine_type = models.OneToOneField(
to=MachineType,
on_delete=models.CASCADE,
related_name='usage_rule',
)
content = MultiLingualRichTextUploadingField(verbose_name=_("content"))
last_modified = models.DateTimeField(auto_now=True, verbose_name=_("last modified"))
history = HistoricalRecords(excluded_fields=['last_modified'])
def __str__(self):
return _("Usage rules for {machine_type}").format(machine_type=self.machine_type)
def get_absolute_url(self):
return reverse('machine_usage_rule_detail', args=[self.machine_type.pk])