115 lines
2.9 KiB
Python
115 lines
2.9 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""
|
||
@File : common_model
|
||
@Author : xuxingchen
|
||
@Version : 1.0
|
||
@Contact : xuxingchen@sinochem.com
|
||
@Desc : 公用的数据结构
|
||
"""
|
||
import threading
|
||
from enum import Enum
|
||
from typing import Optional
|
||
|
||
from pydantic import BaseModel
|
||
|
||
|
||
# 错误码
|
||
class ErrorCode(Enum):
|
||
SUCCESS = (0, "全部完成下发")
|
||
PART_SUCCESS = (10, "部分完成下发")
|
||
FAILURE = (20, "一个都没下发成功")
|
||
INPUT_TYPE_ERROR = (30, "入参数据类型异常")
|
||
NEVER_POST_FACE = (40, "用户未下发过人脸")
|
||
UNKNOWN = (90, "未知异常")
|
||
|
||
def __new__(cls, code, message):
|
||
obj = object.__new__(cls)
|
||
obj._value_ = code
|
||
obj.message = message
|
||
return obj
|
||
|
||
|
||
class BaseInfo(BaseModel):
|
||
def check(self):
|
||
for attr in self.__dict__.keys():
|
||
# if property can be null, default value should not be set to None
|
||
if self.__dict__[attr] is None:
|
||
raise ValueError(f"{attr} not allowed to be set to None")
|
||
|
||
|
||
class UserData:
|
||
def __init__(self):
|
||
self.topic: Optional[str] = None
|
||
self.topics: list = []
|
||
self.table_handler = None
|
||
self.code = None
|
||
self.token = None
|
||
self.status: dict = {}
|
||
self.clients: dict = {}
|
||
self.client_id = None
|
||
self.lock = threading.Lock() # 添加一个锁用于线程同步
|
||
|
||
def set_topic(self, value: str):
|
||
with self.lock:
|
||
self.topic = value
|
||
|
||
def set_topics(self, value: list):
|
||
with self.lock:
|
||
self.topics = value
|
||
|
||
def set_table_handler(self, value):
|
||
with self.lock:
|
||
self.table_handler = value
|
||
|
||
def set_code(self, value):
|
||
with self.lock:
|
||
self.code = value
|
||
|
||
def set_token(self, value):
|
||
with self.lock:
|
||
self.token = value
|
||
|
||
def set_status_add(self, key, value):
|
||
with self.lock:
|
||
self.status[key] = value
|
||
|
||
def set_status_remove(self, key):
|
||
with self.lock:
|
||
if self.status and key in self.status.keys():
|
||
self.status.pop(key)
|
||
|
||
def get_status(self, key):
|
||
if self.status and key in self.status.keys():
|
||
return self.status[key]
|
||
|
||
def set_clients(self, value: dict):
|
||
with self.lock:
|
||
self.clients = value
|
||
|
||
def set_client_add(self, key, value):
|
||
with self.lock:
|
||
self.clients[key] = value
|
||
|
||
def set_client_id(self, key):
|
||
with self.lock:
|
||
self.client_id = key
|
||
|
||
|
||
class Project(BaseModel):
|
||
project_name: str
|
||
device_id: str
|
||
subscribe_topic: str
|
||
publish_topic: str
|
||
gateway_id: str
|
||
gateway_sct: str
|
||
register_type: int
|
||
|
||
|
||
class UserInfo(BaseModel):
|
||
user_id: str
|
||
device_id: str
|
||
name: str
|
||
user_type: int # 0:业主 1:访客
|
||
qrcode: str = None # 当使用二维码时，该值就是美安设备对应的userid
|
||
face_url: str = None
|