import logging from datetime import date, datetime, timedelta from time import sleep import lgpio 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: 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("Dor 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): 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() for auth in card.group.accessauths: logger.info(f"checking auth: {auth.name}") 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")