import logging from datetime import date, datetime, timedelta from time import sleep from sqlalchemy import exc from sqlmodel import select from app.model.models import Card, OneShotAccess from app.services.database import Session, add_and_refresh logger = logging.getLogger(__name__) # See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional # TODO: add sensor pin class DoorController: def __init__( self, lock_pin: int = 17, # connected to 20 on the esp unlock_pin: int = 18, # connected to 21 on the esp mock_factory: bool = False, ): self._is_open: bool = False self._lock_pin = lock_pin self._unlock_pin = unlock_pin self._mock = mock_factory self._chip = None if not mock_factory: import lgpio self._chip = lgpio.gpiochip_open(0) lgpio.gpio_claim_output(self._chip, unlock_pin, 1) lgpio.gpio_claim_output(self._chip, lock_pin, 1) logger.info( "DoorController started. lock=%s unlock=%s mock=%s", lock_pin, unlock_pin, mock_factory, ) def open(self): if self._mock: self._is_open = True logger.info("Door unlocked.[MOCK]") return lgpio.gpio_write(self._chip, self._unlock_pin, 0) sleep(0.4) lgpio.gpio_write(self._chip, self._unlock_pin, 1) self._is_open = True logger.info("Door unlocked!") def close(self): if self._mock: self._is_open = False logger.info("Door locked.[MOCK]") return lgpio.gpio_write(self._chip, self._lock_pin, 0) sleep(0.4) lgpio.gpio_write(self._chip, self._lock_pin, 1) self._is_open = False logger.info("Door locked!") def is_open(self): return self._is_open _contoller: DoorController | None = None def init_controller(ctrl: DoorController): global _contoller _contoller = ctrl def get_controller(): if _contoller is None: raise RuntimeError("DoorController not initialized.") return _contoller def openDoor(): get_controller().open() def closeDoor(): get_controller().close() def isDoorOpen(): return get_controller().is_open() def decrementOneshot(db: Session, oneshot: OneShotAccess): data = oneshot.model_dump() if data["uses"] > 0: data["uses"] = data["uses"] - 1 oneshot.sqlmodel_update(oneshot, update=data) add_and_refresh(db, oneshot) def checkAccess(key: str, db: Session): try: current_weekday = datetime.weekday(date.today()) current_time = datetime.now() card = db.exec(select(Card).where(Card.key == key)).one() if not card.enabled: return False for auth in card.group.accessauths: logger.info(f"checking auth: {auth.name}") if not auth.is_active: return False if auth.type == "timetable": for timetable in auth.timetables: logger.info(f" checking timetable {timetable.id}") logger.info( f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}" ) if current_weekday == timetable.weekday: starttime = datetime.combine(date.today(), timetable.starttime) endtime = starttime + timedelta(minutes=timetable.duration) logger.info( f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}" ) if starttime < current_time < endtime: logger.info("Access Valid!") return True if auth.type == "oneshot": logger.info(f" oneshot auth found: {auth.oneshot}") if current_time < auth.oneshot.ends_at: if auth.oneshot.uses > 0: decrementOneshot(db, auth.oneshot) return True logger.info("No more auths found") return False except exc.NoResultFound: raise Exception("No Access with that key found, this might be a db error")