import threading from tool.typing import * from tool.time import HGSTime, hgs_time_t from tool.location import HGSLocation, hgs_location_t class GarbageType: # 类别从 1 开始计数 GarbageTypeStrList: List = ["", "recyclable", "kitchen", "hazardous", "other"] GarbageTypeStrList_ch: List = ["", "可回收", "厨余", "有害", "其他"] recyclable: enum = 1 kitchen: enum = 2 hazardous: enum = 3 other: enum = 4 class GarbageBag: """垃圾袋""" def __init__(self, gid: gid_t): self._gid: gid_t = gid_t(gid) self._have_use: bool = False self._have_check: bool = False self._type: Union[enum, None] = None self._use_time: Union[HGSTime, None] = None self._user: Union[uid_t, None] = None self._loc: Union[HGSLocation, None] = None self._check = False self._checker: Union[uid_t, None] = None self._lock = threading.RLock() def __repr__(self): tmp = GarbageType.GarbageTypeStrList try: self._lock.acquire() if not self._have_use: info = f"GarbageBag: {self._gid} NOT USE" elif not self._have_check: info = (f"GarbageBag: {self._gid} " f"Type: {tmp[self._type]} " f"Time: {self._use_time.get_time()} " f"User: {self._user} " f"Location {self._loc.get_location()} " f"NOT CHECK") else: info = (f"GarbageBag: {self._gid} " f"Type: {tmp[self._type]} " f"Time: {self._use_time.get_time()} " f"User: {self._user} " f"Location {self._loc.get_location()} " f"Check: {self._check}" f"Checker: {self._checker}") finally: self._lock.release() return info def get_info(self) -> "Dict[str: str]": try: self._lock.acquire() info = { "gid": str(self._gid), "type": str(self._type), "use_time": "" if (self._use_time is None) else str(self._use_time.get_time()), "user": str(self._user), "loc": "" if (self._loc is None) else str(self._loc.get_location()), "check": '1' if self._check else '0', "checker": "None" if self._checker is None else self._checker, } finally: self._lock.release() return info def is_use(self) -> bool: try: self._lock.acquire() have_use = self._have_use finally: self._lock.release() return have_use def is_check(self) -> Tuple[bool, bool]: try: self._lock.acquire() have_check = self._have_check check = self._check finally: self._lock.release() if not have_check: return False, False else: return True, check def get_gid(self): try: self._lock.acquire() gid = self._gid finally: self._lock.release() return gid def get_user(self) -> uid_t: try: self._lock.acquire() assert self._have_use user = self._user finally: self._lock.release() return user def get_type(self): try: self._lock.acquire() assert self._have_use type_ = self._type finally: self._lock.release() return type_ def config_use(self, garbage_type: enum, use_time: hgs_time_t, user: uid_t, location: hgs_location_t): try: self._lock.acquire() assert not self._have_use self._have_use = True if not isinstance(use_time, HGSTime): use_time = HGSTime(use_time) if not isinstance(location, HGSLocation): location = HGSLocation(location) self._type: enum = enum(garbage_type) self._use_time: HGSTime = use_time self._user: uid_t = uid_t(user) self._loc: HGSLocation = location finally: self._lock.release() def config_check(self, use_right: bool, check_uid: uid_t): try: self._lock.acquire() assert self._have_use assert not self._have_check self._have_check = True self._check = bool(use_right) self._checker = uid_t(check_uid) finally: self._lock.release()