from datetime import datetime, UTC from ocpp.v201.datatypes import IdTokenInfoType from ocpp.v201.enums import AuthorizationStatusType from app.database import SessionLocal from app.models.id_token import IdToken from app.models.chargepoint import ChargePoint async def get_id_token_info(chargepoint_id: str, id_token: str): owner_id = None if id_token["type"] not in ["ISO14443", "ISO15693"]: return IdTokenInfoType( status=AuthorizationStatusType.invalid ), owner_id with SessionLocal() as db: db_id_token = db.query(IdToken).filter(IdToken.token == id_token["id_token"]).first() if db_id_token == None: id_token_info = IdTokenInfoType( status=AuthorizationStatusType.unknown ) db_chargepoint = db.query(ChargePoint).filter(ChargePoint.identity == chargepoint_id).first() # Learn token if requested if db_chargepoint.learn_user_id != None: if db_chargepoint.learn_until.timestamp() > datetime.now(UTC).timestamp(): db_id_token = IdToken() db_id_token.friendly_name = "New token learned by {}".format(chargepoint_id) db_id_token.is_active = True db_id_token.owner_id = db_chargepoint.learn_user_id db_id_token.token = id_token["id_token"] db.add(db_id_token) id_token_info=IdTokenInfoType( status=AuthorizationStatusType.accepted ) owner_id = db_id_token.owner_id db_chargepoint.learn_user_id = None db_chargepoint.learn_until = None db.commit() else: owner_id = db_id_token.owner_id if db_id_token.is_active == False: id_token_info=IdTokenInfoType( status=AuthorizationStatusType.blocked ) else: id_token_info=IdTokenInfoType( status=AuthorizationStatusType.accepted ) return id_token_info, owner_id