cmsmanage/doorcontrol/models.py

from datetime import datetime
from typing import Self

from django.conf import settings
from django.db import models
from django.db.models import F, Func, OuterRef, Q, Subquery
from django.db.models.functions import Mod
from django.utils import timezone
from django.utils.functional import cached_property

from membershipworks.models import Flag as MembershipWorksFlag
from membershipworks.models import Member

from .hid.DoorController import DoorController


class Door(models.Model):
    name = models.CharField(max_length=64, unique=True)
    ip = models.GenericIPAddressField(protocol="IPv4")
    access_field = models.TextField(
        max_length=128,
        help_text="Membershipworks field that grants members access to this door",
    )

    def __str__(self):
        return self.name

    @property
    def controller(self) -> DoorController:
        return DoorController(
            self.ip,
            settings.HID_DOOR_USERNAME,
            settings.HID_DOOR_PASSWORD,
        )

    @cached_property
    def card_formats(self):
        return self.controller.get_cardFormats()

    @cached_property
    def schedules_map(self):
        return self.controller.get_scheduleMap()


class DoorCardholderMember(models.Model):
    door = models.ForeignKey(Door, on_delete=models.CASCADE)
    cardholder_id = models.IntegerField()
    member = models.ForeignKey(Member, on_delete=models.CASCADE, db_constraint=False)

    class Meta:
        constraints = (
            models.UniqueConstraint(
                fields=("door", "cardholder_id"), name="unique_door_cardholder_id"
            ),
            models.UniqueConstraint(
                fields=("door", "member"), name="unique_door_member"
            ),
        )

    def __str__(self):
        return f"{self.door} [{self.cardholder_id}]: {self.member}"


class Schedule(models.Model):
    name = models.CharField(max_length=255, unique=True)

    def __str__(self) -> str:
        return f"{self.name}"


class AbstractScheduleRule(models.Model):
    schedule = models.ForeignKey(Schedule, on_delete=models.CASCADE)
    doors = models.ManyToManyField(Door)

    class Meta:
        abstract = True


class FlagScheduleRule(AbstractScheduleRule):
    flag = models.ForeignKey(
        MembershipWorksFlag,
        null=True,
        blank=True,
        on_delete=models.PROTECT,
        db_constraint=False,
    )

    def __str__(self) -> str:
        return f"{self.schedule} [flag: {self.flag}]"


class AttributeScheduleRule(AbstractScheduleRule):
    access_field = models.CharField(
        max_length=128,
        help_text=(
            "Membershipworks field that grants members access to this door using this schedule."
        ),
    )

    def __str__(self) -> str:
        return f"{self.schedule} [attribute: {self.access_field}]"


class HIDEventQuerySet(models.QuerySet):
    def with_decoded_card_number(self):
        # TODO: CONV and BIT_COUNT are MySQL/MariaDB specific
        class Conv(Func):
            function = "CONV"
            arity = 3
            # This is technically not true, but fine for my purposes
            output_field = models.IntegerField()

        class BitCount(Func):
            function = "BIT_COUNT"
            arity = 1

        return (
            self.alias(card_number=Conv(F("raw_card_number"), 16, 10))
            .alias(more_than_26_bits=F("card_number").bitrightshift(26))
            .annotate(card_is_26_bit=Q(more_than_26_bits=0))
            .alias(
                parity_a=Mod(
                    BitCount(F("card_number").bitrightshift(1).bitand(0xFFF)), 2
                ),
                parity_b=Mod(
                    BitCount(F("card_number").bitrightshift(13).bitand(0xFFF)), 2
                ),
            )
            .annotate(
                card_is_valid_26_bit=~Q(parity_a=F("card_number").bitand(1))
                & Q(parity_b=F("card_number").bitrightshift(25).bitand(1))
            )
            .annotate(
                card_number_26_bit=F("card_number").bitrightshift(1).bitand(0xFFFF),
                card_facility_code_26_bit=F("card_number")
                .bitrightshift(17)
                .bitand(0xFF),
            )
        )

    def with_member_id(self):
        return self.annotate(
            member_id=Subquery(
                DoorCardholderMember.objects.filter(
                    door=OuterRef("door"), cardholder_id=OuterRef("cardholder_id")
                ).values("member_id"),
            )
        )


class HIDEvent(models.Model):
    class EventType(models.IntegerChoices):
        DENIED_ACCESS_CARD_NOT_FOUND = 1022, "Denied Access: Card Not Found"
        DENIED_ACCESS_ACCESS_PIN_NOT_FOUND = 1023, "Denied Access Access: PIN Not Found"
        GRANTED_ACCESS = 2020, "Granted Access"
        GRANTED_ACCESS_EXTENDED_TIME = 2021, "Granted Access: Extended Time"
        DENIED_ACCESS_SCHEDULE = 2024, "Denied Access: Schedule"
        DENIED_ACCESS_WRONG_PIN = 2029, "Denied Access: Wrong PIN"
        DENIED_ACCESS_CARD_EXPIRED = 2036, "Denied Access: Card Expired"
        DENIED_ACCESS_PIN_LOCKOUT = 2042, "Denied Access: PIN Lockout"
        DENIED_ACCESS_UNASSIGNED_CARD = 2043, "Denied Access: Unassigned Card"
        DENIED_ACCESS_UNASSIGNED_ACCESS_PIN = (
            2044,
            "Denied Access: Unassigned Access PIN",
        )
        DENIED_ACCESS_PIN_EXPIRED = 2046, "Denied Access: PIN Expired"
        ALARM_ACKNOWLEDGED = 4034, "Alarm Acknowledged"
        DOOR_LOCKED_SCHEDULED = 4035, "Door Locked: Scheduled"
        DOOR_UNLOCKED_SCHEDULED = 4036, "Door Unlocked: Scheduled"
        DOOR_FORCED_ALARM = 4041, "Door Forced Alarm"
        DOOR_HELD_ALARM = 4042, "Door Held Alarm"
        TAMPER_SWITCH_ALARM = 4043, "Tamper Switch Alarm"
        AC_FAILURE = 4044, "AC Failure"
        BATTERY_FAILURE = 4045, "Battery Failure"
        REX_SWITCH_ALARM = 4051, "REX Switch Alarm"
        TIME_SET_TO = 7020, "Time Set To"
        GRANTED_ACCESS_MANUAL = 12031, "Granted Access: Manual"
        DOOR_UNLOCKED = 12032, "Door Unlocked"
        DOOR_LOCKED = 12033, "Door Locked"

        @classmethod
        def any_granted_access(cls) -> list[Self]:
            return [t for t in cls if t.name.startswith("GRANTED_ACCESS")]

    door = models.ForeignKey(Door, on_delete=models.CASCADE)
    timestamp = models.DateTimeField()
    event_type = models.IntegerField(db_column="eventType", choices=EventType.choices)
    reader_address = models.IntegerField(db_column="readerAddress")
    cardholder_id = models.IntegerField(blank=True, null=True, db_column="cardholderID")
    command_status = models.BooleanField(
        blank=True, null=True, db_column="commandStatus"
    )
    forename = models.TextField(blank=True, null=True)
    surname = models.TextField(blank=True, null=True)
    io_state = models.BooleanField(blank=True, null=True, db_column="ioState")
    new_time = models.DateTimeField(blank=True, null=True, db_column="newTime")
    old_time = models.DateTimeField(blank=True, null=True, db_column="oldTime")
    raw_card_number = models.CharField(
        max_length=8, blank=True, null=True, db_column="rawCardNumber"
    )

    # Based on `function isRedEvent` from /html/hid-global.js on a HID EDGE EVO Solo
    is_red = models.GeneratedField(
        expression=Q(
            event_type__in=[
                EventType.DENIED_ACCESS_CARD_NOT_FOUND,
                EventType.DENIED_ACCESS_ACCESS_PIN_NOT_FOUND,
                EventType.DENIED_ACCESS_SCHEDULE,
                EventType.DENIED_ACCESS_WRONG_PIN,
                EventType.DENIED_ACCESS_CARD_EXPIRED,
                EventType.DENIED_ACCESS_PIN_LOCKOUT,
                EventType.DENIED_ACCESS_UNASSIGNED_CARD,
                EventType.DENIED_ACCESS_PIN_EXPIRED,
                EventType.DOOR_FORCED_ALARM,
                EventType.DOOR_HELD_ALARM,
                EventType.TAMPER_SWITCH_ALARM,
                EventType.AC_FAILURE,
                EventType.BATTERY_FAILURE,
            ]
        ),
        output_field=models.BooleanField(),
        db_persist=False,
    )

    objects = HIDEventQuerySet.as_manager()

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["door", "timestamp", "event_type"], name="unique_hidevent"
            )
        ]
        db_table = "hidevent"
        ordering = ("-timestamp",)

    def __str__(self):
        return f"{self.door.name} {self.timestamp} - {self.description}"

    @classmethod
    def from_xml_attributes(cls, door: Door, attrib: dict[str, str]):
        field_lookup = {
            field.column: field.attname for field in HIDEvent._meta.get_fields()
        }

        def attr_to_bool(str):
            if str is None:
                return None
            else:
                return str == "true"

        return cls(
            **{
                **{field_lookup[k]: v for k, v in attrib.items()},
                "door": door,
                # fixups for specific attributes
                "timestamp": timezone.make_aware(
                    datetime.fromisoformat(attrib["timestamp"])
                ),
                "io_state": attr_to_bool(attrib.get("ioState")),
                "command_status": attr_to_bool(attrib.get("commandStatus")),
            }
        )

    @property
    def description(self):
        """
        Based on `Global.localeStrings` from /html/en_EN/en_EN.js
        and `function eventDataHandler` from /html/modules/hid-dashboard.js
        on a HID EDGE EVO Solo
        """

        name = f"{self.forename} {self.surname}"
        direction = "IN" if self.reader_address == 0 else "OUT"

        event_types = {
            1022: f"Denied Access, {direction} Card Not Found {self.raw_card_number}",
            1023: f"Denied Access, {direction} Access PIN Not Found {self.raw_card_number}",
            2020: f"Granted Access, {direction} {name}",
            2021: f"Granted Access, {direction} Extended Time {name}",
            2024: f"Denied Access, {direction} Schedule {name}",
            2029: f"Denied Access, {direction} Wrong PIN {name}",
            2036: f"Denied Access, {direction} Card Expired {name}",
            2042: f"Denied Access, {direction} PIN Lockout {name}",
            2043: f"Denied Access, {direction} Unassigned Card {self.raw_card_number}",
            2044: f"Denied Access, {direction} Unassigned Access PIN {self.raw_card_number}",
            2046: f"Denied Access - PIN Expired {name}",
            4034: "Alarm Acknowledged",
            4035: "Door Locked-Scheduled",
            4036: "Door Unlocked-Scheduled",
            4041: "Door Forced Alarm",
            4042: "Door Held Alarm",
            4043: "Tamper Switch Alarm",
            4044: "AC Failure",
            4045: "Battery Failure",
            4051: "REX Switch Alarm",
            7020: f"Time Set to: {self.new_time}",
            12031: f"Granted Access, {direction} Manual",
            12032: "Door Unlocked",
            12033: "Door Locked",
        }

        return event_types.get(self.event_type, f"Unknown Event Type {self.event_type}")

    def decoded_card_number(self) -> str | None:
        """Requires annotations from `with_decoded_card_number`"""
        if self.raw_card_number is None:
            return None
        elif self.card_is_26_bit:
            if self.card_is_valid_26_bit:
                return f"{self.card_facility_code_26_bit} - {self.card_number_26_bit}"
            else:
                return "Invalid"
        else:
            return "Not 26 bit card"