feat: 非GUI实现

conf/__init__.py

@@ -4,3 +4,29 @@
 """
 
 import sys
+import warnings
+from font.noto import noto_font
+
+if len(sys.argv) != 4:
+    warnings.warn(f"参数不足: {len(sys.argv)}")
+    raise exit(1)
+
+MYSQL_URL = sys.argv[1]
+MYSQL_NAME = sys.argv[2]
+MYSQL_PASSWORD = sys.argv[3]
+
+passwd_salt = "HGSSystem"
+default_score = 10
+default_reputation = 300
+
+max_rubbish_week = 34
+limit_rubbish_week = 50
+
+base_location = "Guangdong-Guangzhou"
+
+font_dict = {
+    "noto": noto_font
+}
+
+capture_num = 0  # 摄像头号
+capture_arg = []

conf/font/noto/OFL.txt

@@ -0,0 +1,93 @@
+Copyright 2012 Google Inc. All Rights Reserved.
+
+This Font Software is licensed under the SIL Open Font License, Version 1.1.
+This license is copied below, and is also available with a FAQ at:
+http://scripts.sil.org/OFL
+
+
+-----------------------------------------------------------
+SIL OPEN FONT LICENSE Version 1.1 - 26 February 2007
+-----------------------------------------------------------
+
+PREAMBLE
+The goals of the Open Font License (OFL) are to stimulate worldwide
+development of collaborative font projects, to support the font creation
+efforts of academic and linguistic communities, and to provide a free and
+open framework in which fonts may be shared and improved in partnership
+with others.
+
+The OFL allows the licensed fonts to be used, studied, modified and
+redistributed freely as long as they are not sold by themselves. The
+fonts, including any derivative works, can be bundled, embedded, 
+redistributed and/or sold with any software provided that any reserved
+names are not used by derivative works. The fonts and derivatives,
+however, cannot be released under any other type of license. The
+requirement for fonts to remain under this license does not apply
+to any document created using the fonts or their derivatives.
+
+DEFINITIONS
+"Font Software" refers to the set of files released by the Copyright
+Holder(s) under this license and clearly marked as such. This may
+include source files, build scripts and documentation.
+
+"Reserved Font Name" refers to any names specified as such after the
+copyright statement(s).
+
+"Original Version" refers to the collection of Font Software components as
+distributed by the Copyright Holder(s).
+
+"Modified Version" refers to any derivative made by adding to, deleting,
+or substituting -- in part or in whole -- any of the components of the
+Original Version, by changing formats or by porting the Font Software to a
+new environment.
+
+"Author" refers to any designer, engineer, programmer, technical
+writer or other person who contributed to the Font Software.
+
+PERMISSION & CONDITIONS
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of the Font Software, to use, study, copy, merge, embed, modify,
+redistribute, and sell modified and unmodified copies of the Font
+Software, subject to the following conditions:
+
+1) Neither the Font Software nor any of its individual components,
+in Original or Modified Versions, may be sold by itself.
+
+2) Original or Modified Versions of the Font Software may be bundled,
+redistributed and/or sold with any software, provided that each copy
+contains the above copyright notice and this license. These can be
+included either as stand-alone text files, human-readable headers or
+in the appropriate machine-readable metadata fields within text or
+binary files as long as those fields can be easily viewed by the user.
+
+3) No Modified Version of the Font Software may use the Reserved Font
+Name(s) unless explicit written permission is granted by the corresponding
+Copyright Holder. This restriction only applies to the primary font name as
+presented to the users.
+
+4) The name(s) of the Copyright Holder(s) or the Author(s) of the Font
+Software shall not be used to promote, endorse or advertise any
+Modified Version, except to acknowledge the contribution(s) of the
+Copyright Holder(s) and the Author(s) or with their explicit written
+permission.
+
+5) The Font Software, modified or unmodified, in part or in whole,
+must be distributed entirely under this license, and must not be
+distributed under any other license. The requirement for fonts to
+remain under this license does not apply to any document created
+using the Font Software.
+
+TERMINATION
+This license becomes null and void if any of the above conditions are
+not met.
+
+DISCLAIMER
+THE FONT SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO ANY WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT
+OF COPYRIGHT, PATENT, TRADEMARK, OR OTHER RIGHT. IN NO EVENT SHALL THE
+COPYRIGHT HOLDER BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+INCLUDING ANY GENERAL, SPECIAL, INDIRECT, INCIDENTAL, OR CONSEQUENTIAL
+DAMAGES, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF THE USE OR INABILITY TO USE THE FONT SOFTWARE OR FROM
+OTHER DEALINGS IN THE FONT SOFTWARE.

conf/font/noto/README.txt

@@ -0,0 +1,99 @@
+Noto Sans Mono Variable Font
+============================
+
+This download contains Noto Sans Mono as both a variable font and static fonts.
+
+Noto Sans Mono is a variable font with these axes:
+  wdth
+  wght
+
+This means all the styles are contained in a single file:
+  NotoSansMono-VariableFont_wdth,wght.ttf
+
+If your app fully supports variable fonts, you can now pick intermediate styles
+that aren’t available as static fonts. Not all apps support variable fonts, and
+in those cases you can use the static font files for Noto Sans Mono:
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-Thin.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-ExtraLight.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-Light.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-Regular.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-Medium.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-SemiBold.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-Bold.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-ExtraBold.ttf
+  static/NotoSansMono_ExtraCondensed/NotoSansMono_ExtraCondensed-Black.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-Thin.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-ExtraLight.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-Light.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-Regular.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-Medium.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-SemiBold.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-Bold.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-ExtraBold.ttf
+  static/NotoSansMono_Condensed/NotoSansMono_Condensed-Black.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-Thin.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-ExtraLight.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-Light.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-Regular.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-Medium.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-SemiBold.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-Bold.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-ExtraBold.ttf
+  static/NotoSansMono_SemiCondensed/NotoSansMono_SemiCondensed-Black.ttf
+  static/NotoSansMono/NotoSansMono-Thin.ttf
+  static/NotoSansMono/NotoSansMono-ExtraLight.ttf
+  static/NotoSansMono/NotoSansMono-Light.ttf
+  static/NotoSansMono/NotoSansMono-Regular.ttf
+  static/NotoSansMono/NotoSansMono-Medium.ttf
+  static/NotoSansMono/NotoSansMono-SemiBold.ttf
+  static/NotoSansMono/NotoSansMono-Bold.ttf
+  static/NotoSansMono/NotoSansMono-ExtraBold.ttf
+  static/NotoSansMono/NotoSansMono-Black.ttf
+
+Get started
+-----------
+
+1. Install the font files you want to use
+
+2. Use your app's font picker to view the font family and all the
+available styles
+
+Learn more about variable fonts
+-------------------------------
+
+  https://developers.google.com/web/fundamentals/design-and-ux/typography/variable-fonts
+  https://variablefonts.typenetwork.com
+  https://medium.com/variable-fonts
+
+In desktop apps
+
+  https://theblog.adobe.com/can-variable-fonts-illustrator-cc
+  https://helpx.adobe.com/nz/photoshop/using/fonts.html#variable_fonts
+
+Online
+
+  https://developers.google.com/fonts/docs/getting_started
+  https://developer.mozilla.org/en-US/docs/Web/CSS/CSS_Fonts/Variable_Fonts_Guide
+  https://developer.microsoft.com/en-us/microsoft-edge/testdrive/demos/variable-fonts
+
+Installing fonts
+
+  MacOS: https://support.apple.com/en-us/HT201749
+  Linux: https://www.google.com/search?q=how+to+install+a+font+on+gnu%2Blinux
+  Windows: https://support.microsoft.com/en-us/help/314960/how-to-install-or-remove-a-font-in-windows
+
+Android Apps
+
+  https://developers.google.com/fonts/docs/android
+  https://developer.android.com/guide/topics/ui/look-and-feel/downloadable-fonts
+
+License
+-------
+Please read the full license text (OFL.txt) to understand the permissions,
+restrictions and requirements for usage, redistribution, and modification.
+
+You can use them freely in your products & projects - print or digital,
+commercial or otherwise.
+
+This isn't legal advice, please consider consulting a lawyer and see the full
+license for all details.

conf/font/noto/__init__.py

@@ -0,0 +1,3 @@
+import os
+font = os.path.dirname(os.path.abspath(__file__))
+noto_font = os.path.join(font, "NotoSansMono-VariableFont_wdth,wght.ttf")

control/__init__.py

@@ -0,0 +1,110 @@
+import conf
+import abc
+from tool.type_ import *
+from sql.db import DB, mysql_db
+from sql.user import update_user, find_user_by_id
+from sql.garbage import update_garbage
+from equipment.scan import HGSCapture, HGSQRCoder, QRCode, capture, qr_capture
+from equipment.scan_user import scan_user
+from equipment.scan_garbage import scan_garbage
+from core.user import User
+from core.garbage import GarbageBag, GarbageType, GarbageBagNotUse
+
+
+class ControlError(Exception):
+    ...
+
+
+class ControlNotLogin(ControlError):
+    ...
+
+
+class ThrowGarbageError(ControlError):
+    ...
+
+
+class CheckGarbageError(ControlError):
+    ...
+
+
+class ControlBase(metaclass=abc.ABCMeta):
+    __control = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls.__control is None:
+            cls.__control = super().__new__()
+        return cls.__control
+
+    def __init__(self,
+                 db: DB = mysql_db,
+                 cap: HGSCapture = capture,
+                 qr: HGSQRCoder = qr_capture,
+                 loc: location_t = conf.base_location):
+        self._db: DB = db
+        self._cap = cap
+        self._qr = qr
+        self._loc: location_t = loc
+        self._user: Optional[User] = None  # 操作者
+
+    def scan(self) -> bool:
+        """
+        处理扫码事务
+        二维码扫描的任务包括: 登录, 扔垃圾, 标记垃圾
+        :return:
+        """
+        self._cap.get_image()
+        qr_code = self._qr.get_qr_code()
+
+        user: Optional[User] = scan_user(qr_code, self._db)
+        if user is not None:
+            self.switch_user(user)
+            return True
+
+        garbage: Optional[GarbageBag] = scan_garbage(qr_code, self._db)
+        if garbage is not None:
+            if self._user is None:
+                raise ControlNotLogin
+            return True
+
+    def throw_garbage(self, garbage: GarbageBag):
+        garbage_type = self.get_rubbish_type()  # 获得垃圾类型
+        if not self._user.throw_rubbish(garbage, garbage_type, self._loc):
+            raise ThrowGarbageError
+        update_garbage(garbage, self._db)
+        update_user(self._user, self._db)
+
+    def check_garbage(self, garbage: GarbageBag):
+        check_result = self.get_rubbish_check()  # 获取垃圾检查结果
+        user = find_user_by_id(garbage.get_user(), self._db)
+        if user is None:
+            raise GarbageBagNotUse
+        if not self._user.check_rubbish(garbage, check_result, user):
+            raise CheckGarbageError
+        update_garbage(garbage, self._db)
+        update_user(self._user, self._db)
+        update_user(user, self._db)
+
+    @abc.abstractmethod
+    def get_rubbish_type(self) -> enum:
+        ...
+
+    @abc.abstractmethod
+    def get_rubbish_check(self) -> bool:
+        ...
+
+    def switch_user(self, user: User) -> bool:
+        """
+        切换用户: 退出/登录
+        :param user: 新用户
+        :return: 登录-True, 退出-False
+        """
+        if self._user is not None and self._user.get_uid() == user.get_uid():
+            self._user = None  # 退出登录
+            return False
+        self._user = user
+        return True  # 登录
+
+    def get_user_info(self):
+        if self._user is None:
+            raise ControlNotLogin
+        return self._user.get_info()

core/garbage.py

@@ -1,14 +1,14 @@
 from tool.type_ import *
 from tool.time_ import HGSTime, hgs_time_t
 from tool.location import HGSLocation, hgs_location_t
-from enum import Enum
 
 
 class GarbageBagNotUse(Exception):
     pass
 
 
-class GarbageType(Enum):
+class GarbageType:
+    GarbageTypeStrList: List = ["", "recyclable", "kitchen", "hazardous", "other"]
     recyclable: enum = 1
     kitchen: enum = 2
     hazardous: enum = 3
@@ -16,25 +16,73 @@ class GarbageType(Enum):
 
 
 class GarbageBag:
-    def __init__(self, gid: gid_t, last_time: time_t):
+    def __init__(self, gid: gid_t):
         self._gid: gid_t = gid
         self._have_use: bool = False
-        self.last_time: HGSTime = HGSTime(last_time)
+        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
+
+    def __repr__(self):
+        tmp = GarbageType.GarbageTypeStrList
+        if not self._have_use:
+            return f"GarbageBag: {self._gid} NOT USE"
+        elif not self._have_check:
+            return (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")
+        return (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}")
+
+    def get_info(self) -> dict[str: str]:
+        return {
+            "gid":      str(self._gid),
+            "type":     str(self._type),
+            "use_time": (self._use_time is None) if "" else str(self._use_time.get_time()),
+            "user":     str(self._user),
+            "loc":      (self._loc is None) if "" else str(self._loc.get_location()),
+            "check":    self._check if '1' else '0',
+            "checker":  str(self._checker),
+        }
+
     def is_use(self) -> bool:
         return self._have_use
 
+    def is_check(self) -> Tuple[bool, bool]:
+        if not self._have_check:
+            return False, False
+        else:
+            return True, self._check
+
+    def get_gid(self):
+        return self._gid
+
     def get_user(self) -> uid_t:
         if not self._have_use:
             raise GarbageBagNotUse
         return self._user
 
+    def get_type(self):
+        if not self._have_use:
+            raise GarbageBagNotUse
+        return self._type
+
     def config_use(self, garbage_type: enum, use_time: hgs_time_t, user: uid_t, location: hgs_location_t):
+        assert not self._have_use
         self._have_use = True
         if not isinstance(use_time, HGSTime):
             use_time = HGSTime(use_time)
@@ -45,5 +93,9 @@ class GarbageBag:
         self._user: uid_t = user
         self._loc: HGSLocation = location
 
-    def is_out_of_date(self) -> bool:
-        return self.last_time.is_out_of_date()
+    def config_check(self, use_right: bool, check_uid: uid_t):
+        assert self._have_use
+        assert not self._have_check
+        self._have_check = True
+        self._check = use_right
+        self._checker = check_uid

core/user.py

@@ -1,41 +1,72 @@
+import abc
+import conf
 from tool.type_ import *
-from enum import Enum
+from tool.time_ import HGSTime
+from .garbage import GarbageBag, GarbageType
+from sql.garbage import update_garbage
 
 
-class UserNotSupportError(Exception):
+class UserException(Exception):
     pass
 
 
-class UserType(Enum):
+class UserNotSupportError(UserException):
+    pass
+
+
+class UserNotScoreException(UserException):
+    pass
+
+
+class UserRubbishException(UserException):
+    pass
+
+
+class UserType:
+    UserTypeStrList: List = ["", "NORMAL", "MANAGER"]
     normal: enum = 1
     manager: enum = 2
 
 
-class User:
+class User(metaclass=abc.ABCMeta):
     def __init__(self, name: uname_t, uid: uid_t, user_type: enum):
         self._name: uname_t = name
         self._uid: uid_t = uid
         self._type: enum = user_type
 
+    def is_manager(self):
+        return self._type == UserType.manager
+
     def get_uid(self) -> uid_t:
         return self._uid
 
     def get_name(self) -> uname_t:
         return self._name
 
-    def get_info(self) -> Dict[str: uid_t, str: uname_t]:
+    def get_user_type(self) -> enum:
+        return self._type
+
+    def get_info(self) -> Dict[str, str]:
         return {
-            "name": self._name,
-            "uid": self._uid,
+            "name": str(self._name),
+            "uid": str(self._uid),
+            "manager": self._type == UserType.manager if '1' else '0'
         }
 
-    def evaluate(self) -> score_t:
+    def __repr__(self):
+        tmp = UserType.UserTypeStrList
+        return f"User {self._uid} {self._name} is {tmp[self._type]}"
+
+    def evaluate(self, is_right: bool) -> score_t:
         raise UserNotSupportError
 
     def add_score(self, score: score_t) -> score_t:
         raise UserNotSupportError
 
-    def throw_rubbish(self) -> count_t:
+    def throw_rubbish(self, garbage: GarbageBag, garbage_type: enum, loc: location_t) -> bool:
+        raise UserNotSupportError
+
+    def check_rubbish(self, garbage: GarbageBag, right: bool, user) -> bool:
         raise UserNotSupportError
 
 
@@ -46,35 +77,133 @@ class NormalUser(User):
         self._rubbish = rubbish
         self._score = score
 
-    def get_info(self) -> Dict[str: uid_t, str: uname_t, str: score_t, str: count_t]:
+    def __repr__(self):
+        return (f"User {self._uid} {self._name} "
+                f"reputation {self._reputation} "
+                f"rubbish {self._rubbish} "
+                f"score {self._score} "
+                f"is NORMAL")
+
+    def get_info(self) -> Dict[str, str]:
         """
         获取当前用户的简单信息
         :return: 用户信息字典
         """
         return {
-            "name": self._name,
-            "uid": self._uid,
-            "reputation": self._reputation,
-            "rubbish": self._rubbish,
-            "score": self._score
+            "name": str(self._name),
+            "uid": str(self._uid),
+            "manager": self._type == UserType.manager if '1' else '0',
+            "reputation": str(self._reputation),
+            "rubbish": str(self._rubbish),
+            "score": str(self._score)
         }
 
-    def evaluate(self) -> score_t:
+    def evaluate(self, is_right: bool) -> score_t:
         """
         评估信誉积分
+        使用朴素贝叶斯公式为基础
+        总分值: 1000分
+        实际分: P(A|B) * 1000
+        A=能正确仍对垃圾的概率
+        B=本次仍对垃圾的概率
+
+        初始概率:
+          P(A) = 0.3  (default_reputation = 300)
+          P(B|A) = 0.6
+          P(B|^A) = 0.3
+          P(B) = 0.8 * 0.3 + 0.1 * 0.7 = 0.31
+          p(^B) = 0.69
+
+          P(A|B) = P(A) * (P(B|A) / P(B)) = P(A) * 2.5806
+          P(A|^B) = P(A) * (P(^B|A) / P(^B)) = P(A) * (0.2/0.96) = P(A) * 0.2083
         :return: 信誉积分
         """
-        ...
+
+        if is_right and self._rubbish > conf.max_rubbish_week:
+            return self._reputation
+
+        pa = self._reputation / 1000  # P(A)
+        if pa < 0.01:
+            pa = 0.01
+        p_a = 1 - pa  # P(^A)
+        pba, p_ba, pb_a, p_b_a = 0.6, 0.4, 0.3, 0.7  # P(B|A), P(^B|A), P(B|^A), P(^B|^A)
+        pb = pba * pa + pb_a * p_a  # P(B) = P(B|A) * P(A) + P(B|^A) * P(^A)
+        p_b = p_ba * pa + p_b_a * p_a  # P(^B) = P(^B|A) * P(A) + P(^B|^A) * P(^A)
+
+        if is_right:
+            new_pa = pa * (pba / pb)  # P(A|B)
+        else:
+            new_pa = pa * (p_ba / p_b)  # P(A|^B)
+        new_pa = new_pa * 1000
+        if int(new_pa) == 0:
+            new_pa = 1
+        if int(new_pa) > 1000:
+            new_pa = 999
+
+        amplitude = new_pa - self._reputation  # 分差
+        amplitude_top = 1000 - self._reputation  # 距离总分分差
+
+        if is_right:
+            if amplitude >= 20:
+                amplitude = amplitude * (amplitude_top / 1000)  # 涨分抑制
+        else:
+            if amplitude <= -20:
+                amplitude = amplitude * (self._reputation / 1000)  # 总分分差月小扣分越高
+
+        self._reputation += int(amplitude)
+        if self._reputation <= 5:
+            self._reputation = 5
+        return self._reputation
 
     def add_score(self, score: score_t) -> score_t:
+        if self._score + score < 0:
+            self._score = 0
+            raise UserNotScoreException
+
         self._score += score
         return self._score
 
-    def throw_rubbish(self) -> count_t:
+    def throw_rubbish(self, garbage: GarbageBag, garbage_type: enum, loc: location_t = conf.base_location) -> bool:
+        if self._rubbish > conf.max_rubbish_week:
+            try:
+                self.add_score(-3)
+            except UserNotScoreException:
+                raise UserRubbishException
+        elif self._rubbish > conf.limit_rubbish_week:
+            raise UserRubbishException
+
+        if garbage.is_use() or not garbage.is_check():
+            return False
+        garbage.config_use(garbage_type, HGSTime(), self._uid, loc)
+
         self._rubbish += 1
-        return self._rubbish
+        return True
+
+    def check_rubbish(self, garbage: GarbageBag, right: bool, user: User) -> bool:
+        raise UserNotSupportError
 
 
 class ManagerUser(User):
     def __init__(self, name: uname_t, uid: uid_t):
-        super(ManagerUser, self).__init__(name, uid, UserType.normal)
+        super(ManagerUser, self).__init__(name, uid, UserType.manager)
+
+    def check_rubbish(self, garbage: GarbageBag, right: bool, user: User) -> bool:
+        if not garbage.is_use() or garbage.is_check() or user.get_uid() != garbage.get_user():
+            return False
+        garbage.config_check(right, self._uid)
+        user.evaluate(right)
+
+        try:
+            if right:
+                if garbage.get_type() == GarbageType.recyclable:
+                    user.add_score(3)
+                elif garbage.get_type() == GarbageType.kitchen or garbage.get_type() == GarbageType.hazardous:
+                    user.add_score(2)
+                else:
+                    user.add_score(1)
+            else:
+                user.add_score(-4)
+        except UserNotScoreException:
+            ...
+
+        return True

equipment/scan.py

@@ -0,0 +1,102 @@
+import time
+import conf
+import cv2 as cv2
+import qrcode
+from tool.type_ import *
+
+
+class HGSCapture:
+    def __init__(self, capnum: int = 0, *args, **kwargs):
+        if cv2.CAP_DSHOW not in args:
+            args = *args, cv2.CAP_DSHOW
+        self._capture = cv2.VideoCapture(capnum, *args, **kwargs)
+        self._frame = None
+
+    def get_image(self):
+        ret, frame = self._capture.read()
+        if ret:
+            self._frame = frame
+        return ret
+
+    def show_image_wait_second(self, wait_second: int = 10):
+        cv2.imshow('frame', self._frame)
+        if wait_second != 0:
+            return cv2.waitKey(wait_second * 1000)
+        return None
+
+    def show_image(self, wait: int = 10):
+        cv2.imshow('frame', self._frame)
+        if wait != 0:
+            return cv2.waitKey(wait)
+        return None
+
+    def get_frame(self):
+        return self._frame
+
+
+class QRCode:
+    def __init__(self, data, get_time: Optional[time_t] = None):
+        self._data: str = data
+        if get_time is None:
+            get_time = time.time()
+        self._time: time_t = get_time
+
+    def __repr__(self):
+        return f"QR:'{self._data}'"
+
+    def get_data(self):
+        return self._data
+
+    def get_time(self):
+        return self._time
+
+    def make_img(self, image: str) -> bool:
+        try:
+            with open(image, "wb") as f:
+                img = qrcode.make(self._data)
+                img.save(f)
+        except (IOError, FileExistsError, FileNotFoundError):
+            return False
+        else:
+            return True
+
+
+class HGSQRCoder:
+    def __init__(self, cap: HGSCapture):
+        self._cap = cap
+        self._last_qr: Optional[QRCode] = None
+
+    def get_qr_code(self) -> Optional[QRCode]:
+        re = self.is_qr_code()
+        if re:
+            return self._last_qr
+        return None
+
+    def is_qr_code(self) -> bool:
+        frame = self._cap.get_frame()
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        coder = cv2.QRCodeDetector()
+        data, points, straight_qrcode = coder.detectAndDecode(gray)
+        old_qr: Optional[QRCode] = self._last_qr
+
+        if len(data) > 0:
+            self._last_qr = QRCode(data)
+            return old_qr is None or data != old_qr.get_data()
+        elif len(data) == 0 and old_qr is not None:
+            if time.time() - old_qr.get_time() >= 1.5:  # 时间间隔大于2s 自动取消
+                self._last_qr = QRCode(data)
+        return False
+
+
+capture = HGSCapture(capnum=conf.capture_num, *conf.capture_arg)
+qr_capture = HGSQRCoder(capture)
+if __name__ == '__main__':
+    while True:
+        capture.get_image()
+        qr_data = qr_capture.get_qr_code()
+        if qr_data is not None:
+            print(qr_data)
+            qr_data.make_img("img.png")
+        capture.show_image(0)
+        if cv2.waitKey(1) == ord('q'):
+            break

equipment/scan_garbage.py

@@ -0,0 +1,87 @@
+from core.garbage import GarbageBag
+from sql.db import DB, mysql_db
+from sql.garbage import find_garbage
+from type_ import *
+from scan import QRCode, capture, qr_capture
+from tool.pic import write_text
+import re
+import os.path
+
+
+qr_user_pattern = re.compile(r'HGSSystem-QR-GARBAGE:([a-z0-9])+-END', re.I)
+
+
+def scan_gid(code: QRCode) -> gid_t:
+    data = code.get_data()
+    res = re.match(qr_user_pattern, data)
+    if res is None:
+        return ""
+    else:
+        return res.group(1)
+
+
+def scan_garbage(code: QRCode, db: DB) -> Optional[GarbageBag]:
+    gid = scan_gid(code)
+    if len(gid) == 0:
+        return None
+    return find_garbage(gid, db)
+
+
+def __get_gid_qr_file_name(gid: gid_t, path: str):
+    path = os.path.join(path, f"gar-{gid}.png")
+    dir_ = os.path.split(path)[0]
+    if len(dir_) > 0:
+        os.makedirs(dir_, exist_ok=True)  # 生成输出目录
+    return path
+
+
+def make_gid_image(gid: gid_t, path: str):
+    qr = QRCode(f"HGSSystem-QR-GARBAGE:{gid}-END")
+    res = qr.make_img(path)
+    if not res:
+        return False
+    write_text((30, 5), "noto", f"GID: {gid}", path)
+    return True
+
+
+def write_uid_qr(gid: gid_t, path: str, db: DB) -> Tuple[str, Optional[GarbageBag]]:
+    user = find_garbage(gid, db)
+    if user is None:
+        return "", None
+
+    path = __get_gid_qr_file_name(gid, path)
+    if make_gid_image(gid, path):
+        return path, user
+    return "", None
+
+
+def write_all_gid_qr(path: str, db: DB, where: str = "") -> List[Tuple[str,]]:
+    if len(where) > 0:
+        where = f"WHERE {where}"
+
+    cur = db.search(f"SELECT gid FROM garbage {where};")
+    if cur is None:
+        return []
+
+    re_list = []
+    for _ in range(cur.rowcount):
+        res = cur.fetchone()
+        assert len(res) == 1
+        path_ = __get_gid_qr_file_name(res[0], path)
+        if make_gid_image(res[0], path_):
+            re_list.append((path_,))
+    return re_list
+
+
+if __name__ == '__main__':
+    write_all_gid_qr("gid", mysql_db)
+    while True:
+        capture.get_image()
+        qr_data = qr_capture.get_qr_code()
+        if qr_data is not None:
+            gar = scan_garbage(qr_data, mysql_db)
+            if gar is not None:
+                print(gar)
+        if capture.show_image(1) == ord('q'):
+            break
+

equipment/scan_user.py

@@ -0,0 +1,94 @@
+from core.user import User
+from sql.db import DB, mysql_db
+from sql.user import find_user_by_id
+from type_ import *
+from scan import QRCode, capture, qr_capture
+from tool.pic import write_text
+import re
+import os.path
+
+
+qr_user_pattern = re.compile(r'HGSSystem-QR-USER:([a-z0-9]{32})-END', re.I)
+
+
+def scan_uid(code: QRCode) -> uid_t:
+    data = code.get_data()
+    res = re.match(qr_user_pattern, data)
+    if res is None:
+        return ""
+    else:
+        return res.group(1)
+
+
+def scan_user(code: QRCode, db: DB) -> Optional[User]:
+    uid = scan_uid(code)
+    if len(uid) == 0:
+        return None
+    return find_user_by_id(uid, db)
+
+
+def __get_uid_qr_file_name(uid: uid_t, name: str, path: str, name_type="nu"):
+    if name_type == "nu":
+        path = os.path.join(path, f"{name}-f{uid}.png")
+    elif name_type == "n":
+        path = os.path.join(path, f"{name}.png")
+    else:
+        path = os.path.join(path, f"{uid}.png")
+
+    dir_ = os.path.split(path)[0]
+    if len(dir_) > 0:
+        os.makedirs(dir_, exist_ok=True)  # 生成输出目录
+    return path
+
+
+def make_uid_image(uid: uid_t, name: uname_t, path: str):
+    qr = QRCode(f"HGSSystem-QR-USER:{uid}-END")
+    res = qr.make_img(path)
+    if not res:
+        return False
+    write_text((60, 5), "noto", f"User: {name} {uid[0: 12]}", path)
+    return True
+
+
+def write_uid_qr(uid: uid_t, path: str, db: DB, name="nu") -> Tuple[str, Optional[User]]:
+    user = find_user_by_id(uid, db)
+    if user is None:
+        return "", None
+
+    user_name = user.get_name()
+    path = __get_uid_qr_file_name(uid, user_name, path, name)
+    if make_uid_image(uid, user_name, path):
+        return path, user
+    return "", None
+
+
+def write_all_uid_qr(path: str, db: DB, name="nu", where: str = "") -> List[Tuple[str,]]:
+    if len(where) > 0:
+        where = f"WHERE {where}"
+
+    cur = db.search(f"SELECT uid, name FROM user {where};")
+    if cur is None:
+        return []
+
+    re_list = []
+    for _ in range(cur.rowcount):
+        res = cur.fetchone()
+        assert len(res) == 2
+        path_ = __get_uid_qr_file_name(res[0], res[1], path, name)
+        if make_uid_image(res[0], res[1], path_):
+            re_list.append((path_,))
+    return re_list
+
+
+if __name__ == '__main__':
+    write_all_uid_qr("uid", mysql_db)
+    while True:
+        capture.get_image()
+        qr_data = qr_capture.get_qr_code()
+        if qr_data is not None:
+            usr = scan_user(qr_data, mysql_db)
+            if usr is not None:
+                print(usr)
+        if capture.show_image(1) == ord('q'):
+            break
+

sql/db.py

@@ -0,0 +1,105 @@
+import pymysql
+from conf import MYSQL_URL, MYSQL_NAME, MYSQL_PASSWORD
+from tool.type_ import *
+
+
+class DBException(Exception):
+    ...
+
+
+class DBDataException(Exception):
+    ...
+
+
+class DBCloseException(Exception):
+    ...
+
+
+class DBDoneException(Exception):
+    ...
+
+
+class DBBit:
+    BIT_0 = b'\x00'
+    BIT_1 = b'\x01'
+
+
+class DB:
+    def __init__(self, host: str = MYSQL_URL, name: str = MYSQL_NAME, passwd: str = MYSQL_PASSWORD):
+        try:
+            self._db = pymysql.connect(user=name, password=passwd, host=host, database="hgssystem")
+        except pymysql.err.OperationalError:
+            raise DBException
+        self._cursor = self._db.cursor()
+
+    def __del__(self):
+        self.close()
+
+    def close(self):
+        if self._cursor is not None:
+            self._cursor.close()
+        if self._db is not None:
+            self._db.close()
+        self._db = None
+        self._cursor = None
+
+    def is_connect(self) -> bool:
+        if self._cursor is None or self._db is None:
+            return False
+        return True
+
+    def get_cursor(self) -> pymysql.cursors.Cursor:
+        if self._cursor is None or self._db is None:
+            raise DBCloseException
+        return self._cursor
+
+    def search(self, sql) -> Union[None, pymysql.cursors.Cursor]:
+        if self._cursor is None or self._db is None:
+            raise DBCloseException
+
+        try:
+            self._cursor.execute(sql)
+        except pymysql.MySQLError:
+            return None
+        return self._cursor
+
+    def done(self, sql) -> Union[None, pymysql.cursors.Cursor]:
+        if self._cursor is None or self._db is None:
+            raise DBCloseException
+
+        try:
+            self._cursor.execute(sql)
+        except pymysql.MySQLError:
+            self._db.rollback()
+            return None
+        finally:
+            self._db.commit()
+        return self._cursor
+
+    def done_(self, sql) -> Union[None, pymysql.cursors.Cursor]:
+        if self._cursor is None or self._db is None:
+            raise DBCloseException
+        try:
+            self._cursor.execute(sql)
+        except pymysql.MySQLError:
+            self._db.rollback()
+            raise DBDoneException
+        return self._cursor
+
+    def done_commit(self):
+        self._db.commit()
+
+
+mysql_db = DB()
+
+if __name__ == '__main__':
+    # 测试程序
+    mysql_db = DB()
+    mysql_db.search("SELECT * FROM user;")
+    res = mysql_db.get_cursor().fetchall()
+    print(res)
+
+    mysql_db.search("SELECT * FROM user WHERE uid = 0;")
+    res = mysql_db.get_cursor().fetchall()
+    print(res)
+    mysql_db.close()

sql/garbage.py

@@ -0,0 +1,164 @@
+import time
+from decimal import Decimal
+from db import DB, mysql_db, DBBit, DBDataException, DBDoneException
+from tool.type_ import *
+from tool.time_ import HGSTime
+from core.garbage import GarbageBag, GarbageType
+
+
+class GarbageDBException(DBDataException):
+    ...
+
+
+def countGarbageByTime(uid: uid_t, db: DB):
+    ti: time_t = time.time()
+    start = ti - 3.5 * 24 * 60 * 60  # 前后3.5天
+    end = ti + 3.5 * 24 * 60 * 60
+    cur = db.search(f"SELECT gid FROM garbage_time WHERE uid = '{uid}' AND use_time BETWEEN {start} AND {end};")
+    if cur is None:
+        return -1
+    return cur.rowcount
+
+
+def __find_garbage(sql: str, res_len, db: DB):
+    cur = db.search(sql)
+    if cur is None or cur.rowcount == 0:
+        return [None, tuple()]
+    assert cur.rowcount == 1
+    res = cur.fetchone()
+    assert len(res) == res_len
+    return GarbageBag(str(res[0])), res
+
+
+def find_not_use_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
+    return __find_garbage(f"SELECT gid FROM garbage_n WHERE gid = {gid};", 1, db)[0]
+
+
+def find_wait_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
+    res: Tuple[int, bytes, Decimal, str, str]
+    gb: GarbageBag
+    gb, res = __find_garbage(f"SELECT gid, type, use_time, uid, loc FROM garbage_c WHERE gid = {gid};", 5, db)
+    if gb is None:
+        return None
+
+    garbage_type: enum = int(res[1].decode())
+    use_time: time_t = float(res[2])
+    uid: uid_t = res[3]
+    loc: location_t = res[4]
+
+    gb.config_use(garbage_type, use_time, uid, loc)
+    return gb
+
+
+def find_use_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
+    res: Tuple[int, bytes, Decimal, str, str, bytes]
+    gb: GarbageBag
+    gb, res = __find_garbage(f"SELECT gid, type, use_time, uid, loc, right_use, check_uid "
+                             f"FROM garbage_u WHERE gid = {gid};", 7, db)
+    if gb is None:
+        return None
+
+    garbage_type: enum = int(res[1].decode())
+    use_time: time_t = float(res[2])
+    uid: uid_t = res[3]
+    loc: location_t = res[4]
+    check: bool = res[5] == DBBit.BIT_1
+    check_uid: uid_t = res[6]
+
+    gb.config_use(garbage_type, use_time, uid, loc)
+    gb.config_check(check, check_uid)
+    return gb
+
+
+def find_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
+    res: Tuple[bool, int] = is_garbage_exists(gid, db)
+    if not res[0]:
+        return None
+    elif res[1] == 0:
+        re = find_not_use_garbage(gid, db)
+    elif res[1] == 1:
+        re = find_wait_garbage(gid, db)
+    elif res[1] == 2:
+        re = find_use_garbage(gid, db)
+    else:
+        re = None
+    assert re is not None
+    return re
+
+
+def is_garbage_exists(gid: gid_t, db: DB) -> Tuple[bool, int]:
+    cur = db.search(f"SELECT gid, flat FROM garbage WHERE gid = {gid};")
+    if cur is None or cur.rowcount == 0:
+        return False, 0
+    assert cur.rowcount == 1
+    res: Tuple[int, int] = cur.fetchone()
+    return True, res[1]
+
+
+def update_garbage(garbage: GarbageBag, db: DB) -> bool:
+    re = find_garbage(garbage.get_gid(), db)
+    if re is None:
+        return False
+
+    if re.is_use() and not garbage.is_use() or re.is_check() and not garbage.is_check():
+        return False
+
+    if not garbage.is_use() and not garbage.is_check():
+        return True  # 不做任何修改
+
+    gid = garbage.get_gid()
+    info = garbage.get_info()
+
+    try:
+        db.done_(f"DELETE FROM garbage_n WHERE gid = {gid};")
+        db.done_(f"DELETE FROM garbage_c WHERE gid = {gid};")
+        db.done_(f"DELETE FROM garbage_u WHERE gid = {gid};")
+
+        if garbage.is_check()[0]:
+            db.done_(f"UPDATE garbage SET flat = 2 WHERE gid = {gid};")
+            db.done_(f"INSERT INTO garbage_u(gid, uid, use_time, type, loc, right_use, check_uid) "
+                     f"VALUES ({info['gid']} , '{info['user']}'  , {info['use_time']}, {info['type']}, "
+                     f"       '{info['loc']}', {info['check']}, '{info['checker']}');")
+        else:
+            db.done_(f"UPDATE garbage SET flat = 1 WHERE gid = {gid};")
+            db.done_(f"INSERT INTO garbage_c(gid, uid, use_time, type, loc) "
+                     f"VALUES ({info['gid']} , '{info['user']}', {info['use_time']}, {info['type']}, "
+                     f"       '{info['loc']}');")
+    except DBDoneException:
+        return False
+    finally:
+        db.done_commit()
+    return True
+
+
+def creat_new_garbage(db: DB) -> Optional[GarbageBag]:
+    cur = db.done("INSERT INTO garbage() VALUES ();")
+    if cur is None:
+        return None
+    assert cur.rowcount == 1
+    gid = cur.lastrowid
+    cur = db.done(f"INSERT INTO garbage_n(gid) VALUES ({gid});")
+    if cur is None:
+        return None
+    assert cur.rowcount == 1
+    return GarbageBag(str(gid))
+
+
+if __name__ == '__main__':
+    bag = creat_new_garbage(mysql_db)
+    print(bag)
+    bag.config_use(GarbageType.recyclable, HGSTime(), "1e1d30a1f9b78c8fa852d19b4cfaee79", "HuaDu")
+    update_garbage(bag, mysql_db)
+    print(bag)
+
+    bag = find_garbage(bag.get_gid(), mysql_db)
+    print(bag)
+
+    bag.config_check(True, "048529ca5c6accf594b74e6f73ee1bf0")
+    update_garbage(bag, mysql_db)
+    print(bag)
+
+    bag = find_garbage(bag.get_gid(), mysql_db)
+    print(bag)
+
+    print(countGarbageByTime("1e1d30a1f9b78c8fa852d19b4cfaee79", mysql_db))

sql/user.py

@@ -0,0 +1,103 @@
+from db import DB, mysql_db, DBBit
+from tool.type_ import *
+from tool.login import creat_uid, randomPassword
+from core.user import NormalUser, ManagerUser, User
+import conf
+from .garbage import countGarbageByTime
+
+
+def find_user_by_id(uid: uid_t, db: DB) -> Optional[User]:
+    cur = db.search(f"SELECT uid, name, manager, score, reputation FROM user WHERE uid = '{uid}';")
+    if cur is None or cur.rowcount == 0:
+        return None
+    assert cur.rowcount == 1
+    res = cur.fetchone()
+    assert len(res) == 5
+
+    uid: uid_t = res[0]
+    name: uname_t = str(res[1])
+    manager: bool = res[2] == DBBit.BIT_1
+
+    if manager:
+        return ManagerUser(name, uid)
+    else:
+        score: score_t = res[3]
+        reputation: score_t = res[4]
+        rubbish: count_t = countGarbageByTime(uid, db)
+        return NormalUser(name, uid, reputation, rubbish, score)  # rubbish 实际计算
+
+
+def find_user_by_name(name: uname_t, passwd: passwd_t, db: DB) -> Optional[User]:
+    uid = creat_uid(name, passwd)
+    return find_user_by_id(uid, db)
+
+
+def is_user_exists(uid: uid_t, db: DB) -> bool:
+    cur = db.search(f"SELECT uid FROM user WHERE uid = '{uid}';")
+    if cur is None or cur.rowcount == 0:
+        return False
+    assert cur.rowcount == 1
+    return True
+
+
+def update_user(user: User, db: DB) -> bool:
+    if not is_user_exists(user.get_uid(), db):
+        return False
+
+    uid = user.get_uid()
+    info: dict = user.get_info()
+    is_manager = info['manager']
+    if is_manager == '1':
+        cur = db.done(f"UPDATE user "
+                      f"SET manager = {is_manager} "
+                      f"WHERE uid = '{uid}';")
+    else:
+        score = info['score']
+        reputation = info['reputation']
+        cur = db.done(f"UPDATE user "
+                      f"SET manager = {is_manager},"
+                      f"    score = {score},"
+                      f"    reputation = {reputation} "
+                      f"WHERE uid = '{uid}';")
+    return cur is not None
+
+
+def creat_new_user(name: Optional[uname_t], passwd: Optional[passwd_t], phone: phone_t, manager: bool, db: DB) -> Optional[User]:
+    if name is None:
+        name = randomPassword()
+
+    if passwd is None:
+        passwd = randomPassword()
+
+    uid = creat_uid(name, passwd)
+    if is_user_exists(uid, db):
+        return None
+    is_manager = manager if '1' else '0'
+    cur = db.done(f"INSERT INTO user(uid, name, manager, phone, score, reputation) "
+                  f"VALUES ('{uid}', '{name}', {is_manager}, '{phone}', {conf.default_score}, "
+                  f"{conf.default_reputation});")
+    if cur is None:
+        return None
+    if is_manager:
+        return ManagerUser(name, uid)
+    return NormalUser(name, uid, conf.default_reputation, 0, conf.default_score)
+
+
+if __name__ == '__main__':
+    name_ = 'Huan8'
+    usr = find_user_by_name(name_, "123", mysql_db)
+    if usr is None:
+        usr = creat_new_user(name_, "123", "12345678900", False, mysql_db)
+    print(usr)
+
+    for i in range(90):
+        usr.evaluate(False)
+        print(usr)
+
+    for i in range(90):
+        usr.evaluate(True)
+        print(usr)
+
+    update_user(usr, mysql_db)
+    usr = find_user_by_name(name_, "123", mysql_db)
+    print(usr)

tool/location.py

@@ -1,12 +1,12 @@
-from type_ import *
+from .type_ import *
 
 
 class HGSLocation:
     def __init__(self, location: location_t):
-        self.loc: location_t = location
+        self._loc: location_t = location
 
     def get_location(self) -> location_t:
-        return self.loc
+        return self._loc
 
 
-hgs_location_t = NewType("hgs_location_t", Union[HGSLocation, location_t])
+hgs_location_t = "hgs_location_t", Union[HGSLocation, location_t]

tool/login.py

@@ -1,6 +1,20 @@
-from type_ import *
+from .type_ import *
 import hashlib
+import conf
+from random import randint
 
 
-def check_login(uid: uid_t, name: uname_t, passwd: passwd_t, salt: str) -> bool:
-    ...
+def creat_uid(name: uname_t, passwd: passwd_t, salt: str = conf.passwd_salt) -> str:
+    return hashlib.md5(f"HGSSystem-USER{name}-PASSWORD:{passwd}-{salt}-END".encode('utf-8')).hexdigest()
+
+
+def check_login(uid: uid_t, name: uname_t, passwd: passwd_t, salt: str = conf.passwd_salt) -> bool:
+    return uid == creat_uid(name, passwd, salt)
+
+
+def randomPassword():
+    passwd_char = 'qwertyuiopasdfghjklzxcvbnmQWERTYUIOPASDFGHJKLZXCVBNM1234567890<>,.?/:;''!@#$%^&*()-_=+*'
+    passwd = []
+    for i in range(randint(16, 22)):
+        passwd.append(passwd_char[randint(0, len(passwd_char) - 1)])
+    return "".join(passwd)

tool/pic.py

@@ -0,0 +1,11 @@
+import conf
+from .type_ import *
+from PIL import Image, ImageDraw, ImageFont
+
+
+def write_text(pos: Tuple[int, int], font: str, text: str, path: str):
+    image = Image.open(path)
+    draw = ImageDraw.Draw(image)
+    font = ImageFont.truetype(font=conf.font_dict[font], size=20)
+    draw.text(pos, text, font=font)
+    image.save(path)

tool/time_.py

@@ -1,4 +1,4 @@
-from type_ import *
+from .type_ import *
 import time
 
 
@@ -10,8 +10,11 @@ class HGSTime:
             self._time: time_t = second
         self._time_local: time.struct_time = time.localtime(self._time)
 
+    def get_time(self) -> time_t:
+        return self._time
+
     def is_out_of_date(self):
         return time.time() > self._time
 
 
-hgs_time_t = NewType("hgs_time_t", Union[HGSTime, time_t])
+hgs_time_t = Union[HGSTime, time_t]

tool/type_.py

@@ -1,15 +1,16 @@
-from typing import Dict, Union, NewType
+from typing import Dict, List, Tuple, Union, Optional
 
-gid_t = NewType("gid_t", str)  # garbage bag id 类型
-uid_t = NewType("uid_t", str)  # user id 类型
+gid_t = str  # garbage bag id 类型
+uid_t = str  # user id 类型
 
-uname_t = NewType("uname_t", str)  # user name 类型
-passwd_t = NewType("passwd_t", str)  # user password 类型
+uname_t = str  # user name 类型
+passwd_t = str  # user password 类型
+phone_t = str
 
-count_t = NewType("count_t", int)  # 计数类型 (垃圾计数)
-score_t = NewType("score_t", int)  # 积分类型
-location_t = NewType("location_t", str)
-time_t = NewType("time_t", float)  # 时间类型
-day_t = NewType("day_t", int)
+count_t = int  # 计数类型 (垃圾计数)
+score_t = int  # 积分类型
+location_t = str
+time_t = float  # 时间类型
+day_t = int
 
-enum = NewType("enum", int)  # 枚举类型
+enum = int  # 枚举类型