ahtlon
1d37be5317
Make the sleep interruptable Add debug endpoints for caldav - gonna make a new endpoint for them i think...
134 lines
3.7 KiB
Python
134 lines
3.7 KiB
Python
import logging
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import exc
|
|
from sqlmodel import select
|
|
|
|
from app.model.models import Card, OneShotAccess
|
|
from app.services.caldav import get_caldav_handler
|
|
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,
|
|
mqtt_host: str = "localhost",
|
|
mqtt_port: int = 1883,
|
|
mock_factory: bool = False,
|
|
):
|
|
self._is_open: bool = False
|
|
self._mqtt = None
|
|
self._mqtt_topic = "nukihub/lock/action"
|
|
self._mock = mock_factory
|
|
|
|
if not mock_factory:
|
|
import paho.mqtt.client as mqtt
|
|
|
|
self._mqtt = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
|
|
self._mqtt.connect(
|
|
mqtt_host, mqtt_port
|
|
) # TODO: add login with username+pw, tls
|
|
self._mqtt.loop_start()
|
|
logger.info("Mqtt client connect to %s:%s", mqtt_host, mqtt_port)
|
|
|
|
logger.info(
|
|
"DoorController started. topic=%s mock=%s",
|
|
self._mqtt_topic,
|
|
mock_factory,
|
|
)
|
|
|
|
def open(self):
|
|
if self._mock:
|
|
self._is_open = True
|
|
logger.info("Door unlocked.[MOCK]")
|
|
return
|
|
|
|
self._mqtt.publish(self._mqtt_topic, "unlock")
|
|
self._is_open = True
|
|
logger.info("Door unlocked!")
|
|
|
|
def close(self):
|
|
if self._mock:
|
|
self._is_open = False
|
|
logger.info("Door locked.[MOCK]")
|
|
return
|
|
|
|
self._mqtt.publish(self._mqtt_topic, "lock")
|
|
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:
|
|
card = db.exec(select(Card).where(Card.key == key)).one()
|
|
if not card.enabled:
|
|
logger.info("Card Inactive!")
|
|
return False
|
|
|
|
caldav_handler = get_caldav_handler()
|
|
if caldav_handler and caldav_handler.is_group_active(card.group.name):
|
|
logger.info("Access Valid!")
|
|
return True
|
|
|
|
current_time = datetime.now().astimezone()
|
|
for auth in card.group.accessauths:
|
|
logger.info(f"checking auth: {auth.name}")
|
|
if not auth.is_active:
|
|
logger.info("AA inactive!")
|
|
continue
|
|
if auth.type == "oneshot":
|
|
logger.info(f" oneshot auth found: {auth.oneshot}")
|
|
ends_at = auth.oneshot.ends_at
|
|
if ends_at.tzinfo is None:
|
|
ends_at = ends_at.astimezone()
|
|
if current_time < ends_at and auth.oneshot.uses > 0:
|
|
decrementOneshot(db, auth.oneshot)
|
|
return True
|
|
logger.info("No more auths found")
|
|
return False
|
|
except exc.NoResultFound as err:
|
|
raise Exception(
|
|
"No Access with that key found, this might be a db error"
|
|
) from err
|