50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
|
|
import paho.mqtt.client as mqttClient
|
|
import paho.mqtt.publish as publish
|
|
|
|
from sqlmodel import select
|
|
from fastapi import Depends, HTTPException, status
|
|
from sqlalchemy.orm import selectinload
|
|
import sqlalchemy.exc as exc
|
|
import datetime
|
|
|
|
from app.services.database import Session, get_session
|
|
from app.model.models import *
|
|
|
|
doorIsOpen = True
|
|
client = mqttClient.Client(client_id="", userdata=None, protocol=mqttClient.MQTTv5)
|
|
client.tls_set(tls_version=mqttClient.ssl.PROTOCOL_TLS)
|
|
client.username_pw_set("username", "passwort")
|
|
#client.connect("host", port=8883)
|
|
# I think this could also be gpio controlled
|
|
#See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional
|
|
|
|
def openDoor():
|
|
doorIsOpen = True
|
|
publish.single(topic="/lock/action", payload="unlock")
|
|
pass
|
|
|
|
def closeDoor():
|
|
doorIsOpen = False
|
|
publish.single(topic="/lock/action", payload="lock")
|
|
pass
|
|
|
|
def isDoorOpen():
|
|
return doorIsOpen
|
|
|
|
def checkAccess(uuid: str, db: Session = Depends(get_session)):
|
|
try:
|
|
current_weekday = datetime.datetime.weekday(datetime.date.today())
|
|
print(f"current day is {current_weekday}")
|
|
card = db.exec(select(Card).where(Card.uuid == uuid)).one()
|
|
for auth in card.group.accessauths:
|
|
print(f"checking auth: {auth.name}")
|
|
for timetable in auth.timetables:
|
|
print(f"timetable date is {timetable.weekday - 1}")
|
|
if current_weekday == timetable.weekday - 1:
|
|
return True
|
|
return False
|
|
except exc.NoResultFound:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
|
|
|