307 lines
11 KiB
Python
307 lines
11 KiB
Python
from datetime import datetime, timedelta
|
|
from typing import Self
|
|
|
|
from django.conf import settings
|
|
from django.db import models
|
|
from django.db.models import OuterRef, Q, Subquery
|
|
from django.utils import timezone
|
|
from django.utils.functional import cached_property
|
|
|
|
from membershipworks.models import EventExt, Member, MemberQuerySet
|
|
from membershipworks.models import Flag as MembershipWorksFlag
|
|
|
|
from .hid.Credential import Credential, InvalidHexCode
|
|
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",
|
|
)
|
|
|
|
class Meta:
|
|
permissions = [("assign_nfc_card", "Assign NFC cards to members.")]
|
|
|
|
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):
|
|
name = models.CharField(
|
|
help_text="Used for creating/matching Unifi Access user groups. Do not change after creation without also changing name in Access.",
|
|
max_length=255,
|
|
unique=True,
|
|
)
|
|
schedule = models.ForeignKey(Schedule, on_delete=models.CASCADE)
|
|
doors = models.ManyToManyField(Door)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def get_matching_members(self) -> MemberQuerySet:
|
|
raise NotImplementedError
|
|
|
|
|
|
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}]"
|
|
|
|
def get_matching_members(self) -> MemberQuerySet:
|
|
return self.flag.members.all()
|
|
|
|
|
|
class ActiveEventInstructorRule(AbstractScheduleRule):
|
|
def __str__(self) -> str:
|
|
return "{self.schedule} [Active Instructor]"
|
|
|
|
# grant instructors access for ~1 hour around their class times
|
|
def get_matching_members(self) -> MemberQuerySet:
|
|
now = timezone.now()
|
|
margin = timedelta(hours=1)
|
|
active_event_instructors = EventExt.objects.filter(
|
|
occurred=True,
|
|
meeting_times__start__lt=now + margin,
|
|
meeting_times__end__gt=now - margin,
|
|
).values_list("instructor", flat=True)
|
|
return Member.objects.filter(eventinstructor__in=active_event_instructors)
|
|
|
|
|
|
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}]"
|
|
|
|
def get_matching_members(self) -> MemberQuerySet:
|
|
return Member.objects.filter(**{self.access_field: True})
|
|
|
|
|
|
class HIDEventQuerySet(models.QuerySet):
|
|
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.TextField(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=True,
|
|
)
|
|
|
|
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(attr):
|
|
if attr is None:
|
|
return None
|
|
else:
|
|
return attr == "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:
|
|
if self.raw_card_number is None:
|
|
return None
|
|
try:
|
|
cred = Credential.from_26bit_hex(self.raw_card_number)
|
|
return f"{cred.facility_code} - {cred.card_number}"
|
|
except InvalidHexCode as e:
|
|
return f"Invalid: {e}"
|