feat: 新增通过率分析

tk_ui/admin_event.py

@@ -737,3 +737,40 @@ class ScoreReputationDistributedEvent(AdminEventBase):
         else:
             lst = [int(i[0]) for i in res]
             self._program.show_result(lst)
+
+
+class PassingRateEvent(AdminEventBase):
+    def func(self, columns, where, where_select, order_by):
+        where_str = " AND ".join(["g.CheckResult is not null", *where_select])
+        columns += [f"get_avg(count(GarbageID), "
+                    f"(SELECT count(g.GarbageID) "
+                    f"FROM garbage AS g WHERE {where_str})) AS count"]
+        where += ["CheckResult is not null", "CheckResult=1"]
+        cur = self._db.search(columns=columns,
+                              table="garbage",
+                              where=where,
+                              order_by=[(i, "DESC") for i in order_by] + [("count", "DESC")],
+                              group_by=order_by if len(order_by) != 0 else None)
+        if cur is None:
+            return None
+        return cur.fetchall()
+
+    def __init__(self, gb_station):
+        super().__init__(gb_station)
+        self.thread = None
+        self._program: Optional[admin_program.StatisticsScoreDistributedProgram] = None
+
+    def start(self, columns, where, where_select, order_by, program):
+        self.thread = TkThreading(self.func, columns, where, where_select, order_by)
+        self._program = program
+        return self
+
+    def is_end(self) -> bool:
+        return not self.thread.is_alive()
+
+    def done_after_event(self):
+        res: Optional[List] = self.thread.wait_event()
+        if res is None:
+            self.station.show_warning("数据分析", "数据获取时发生错误")
+        else:
+            self._program.show_result(res)

tk_ui/admin_menu.py

@@ -214,6 +214,7 @@ class StatisticsMenu(AdminMenu):
         super().conf_gui(color, n)
         self.btn[0]['command'] = self.statistics_time_command
         self.btn[1]['command'] = self.statistics_user_command
+        self.btn[2]['command'] = self.statistics_pass_command
 
     def statistics_time_command(self):
         self.station.to_menu("时段分析")
@@ -221,6 +222,9 @@ class StatisticsMenu(AdminMenu):
     def statistics_user_command(self):
         self.station.to_menu("积分信用分析")
 
+    def statistics_pass_command(self):
+        self.station.to_menu("通过率")
+
 
 class StatisticsTimeMenu(AdminMenu):
     def __init__(self, station, win, color):
@@ -287,5 +291,31 @@ class StatisticsUserMenu(AdminMenu):
         self.station.to_program("垃圾分类信用分布")
 
 
+class StatisticsPassMenu(AdminMenu):
+    def __init__(self, station, win, color):
+        super().__init__(station, win, color, "通过率")
+        self.btn: List[tk.Button] = [tk.Button(self.frame) for _ in range(4)]
+        self.btn_name = ["全局", "按类型", "按区域", "按类型和区域"]
+
+    def conf_gui(self, color: str, n: int = 1):
+        super().conf_gui(color, n)
+        self.btn[0]['command'] = self.by_global
+        self.btn[1]['command'] = self.by_type
+        self.btn[2]['command'] = self.by_loc
+        self.btn[3]['command'] = self.by_type_loc
+
+    def by_global(self):
+        self.station.to_program("通过率-全局")
+
+    def by_type(self):
+        self.station.to_program("通过率-按类型")
+
+    def by_loc(self):
+        self.station.to_program("通过率-按区域")
+
+    def by_type_loc(self):
+        self.station.to_program("通过率-按类型和区域")
+
+
 all_menu = [MainMenu, CreateMenu, DeleteMenu, SearchMenu, UpdateMenu, StatisticsMenu, StatisticsTimeMenu,
-            StatisticsUserMenu]
+            StatisticsUserMenu, StatisticsPassMenu]

tk_ui/admin_program.py

@@ -3,6 +3,8 @@ import tkinter as tk
 import tkinter.ttk as ttk
 from tkinter.filedialog import askdirectory, askopenfilename, asksaveasfilename
 import random
+
+import matplotlib.pyplot
 from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk
 from matplotlib.axes import Axes
 import numpy as np
@@ -2362,6 +2364,225 @@ class StatisticsReputationDistributedProgram(StatisticsUserBaseProgram):
         self.station.push_event(event)
 
 
+class StatisticsPassRateGlobalProgram(StatisticsUserBaseProgram):
+    def __init__(self, station, win, color):
+        super(StatisticsPassRateGlobalProgram, self).__init__(station, win, color, "通过率-全局")
+
+    def show_result(self, lst: np.array):
+        passing = float(lst[0][0])
+        not_passing = 1 - passing
+        data = [passing, not_passing]
+        label = ["通过", "未通过"]
+
+        res = self.plt.pie(data, radius=1, pctdistance=0.7, textprops=dict(color='w'),  # 不显示文字
+                           shadow=True, startangle=45, autopct="%6.3f%%", wedgeprops=dict(width=0.6, edgecolor="w"))
+        self.plt.legend(res[0], label, loc="lower left")
+        self.plt.set_title("全局垃圾分类通过率")  # 设置标题以及其位置和字体大小
+        self.canvas.draw()
+        self.toolbar.update()
+
+    def export(self):
+        self.station.show_msg("保存数据", f"数据不支持导出")
+        return
+
+    def to_program(self):
+        self.refresh()
+
+    def refresh(self, _=None):
+        self.plt.cla()
+        event = tk_event.PassingRateEvent(self.station).start([], [], [], [], self)
+        self.station.push_event(event)
+
+
+class StatisticsPassRateTypeProgram(StatisticsUserBaseProgram):
+    def __init__(self, station, win, color):
+        super(StatisticsPassRateTypeProgram, self).__init__(station, win, color, "通过率-按类型")
+
+    def show_result(self, lst: List[Tuple[bytes, any]]):
+        data_1, data_2, data_3, data_4 = [1.0, 0.0], [1.0, 0.0], [1.0, 0.0], [1.0, 0.0]
+
+        for i in lst:
+            tmp: bytes = i[0]
+            type_ = tmp.decode('utf-8')
+            if type_ == '1':
+                data_1 = [float(i[1]), 1 - float(i[1])]
+            elif type_ == '2':
+                data_2 = [float(i[1]), 1 - float(i[1])]
+            elif type_ == '3':
+                data_3 = [float(i[1]), 1 - float(i[1])]
+            elif type_ == '4':
+                data_4 = [float(i[1]), 1 - float(i[1])]
+
+        legend_text = []
+        for data, r, s in zip([data_1, data_2, data_3, data_4], [0.3, 0.6, 0.9, 1.2], [0, 15, 30, 45]):
+            res = self.plt.pie(data, radius=r, pctdistance=0.7,  # 不显示文字
+                               shadow=True, startangle=s, autopct="%6.3f%%", wedgeprops=dict(width=0.3, edgecolor="w"))
+            legend_text += res[0]
+
+        label = []
+        for i in GarbageType.GarbageTypeStrList_ch[1:]:
+            label.append(f"{i}-通过")
+            label.append(f"{i}-不通过")
+
+        self.plt.legend(legend_text, label)
+        self.plt.set_title("全局垃圾分类通过率")  # 设置标题以及其位置和字体大小
+        self.canvas.draw()
+        self.toolbar.update()
+
+    def export(self):
+        self.station.show_msg("保存数据", f"数据不支持导出")
+        return
+
+    def to_program(self):
+        self.refresh()
+
+    def refresh(self, _=None):
+        self.plt.cla()
+        event = tk_event.PassingRateEvent(self.station).start(["GarbageType"],
+                                                              [],
+                                                              ["g.GarbageType=garbage.GarbageType"],
+                                                              ["GarbageType"], self)
+        self.station.push_event(event)
+
+
+class StatisticsPassRateLocProgram(StatisticsUserBaseProgram):
+    def __init__(self, station, win, color):
+        super(StatisticsPassRateLocProgram, self).__init__(station, win, color, "通过率-按区域")
+        self.loc_frame = tk.Frame(self.frame)
+        self.loc_title = tk.Label(self.loc_frame)
+        self.loc_enter = tk.Entry(self.loc_frame), tk.StringVar()
+
+    def conf_gui(self, n: int = 1):
+        super(StatisticsPassRateLocProgram, self).conf_gui(n)
+        title_font = make_font(size=16)
+
+        self.loc_frame['bg'] = self.bg_color
+        self.loc_frame['bd'] = 5
+        self.loc_frame['relief'] = "ridge"
+        self.loc_frame.place(relx=0.0, rely=0.92, relwidth=0.33, relheight=0.07)
+
+        self.loc_title['font'] = title_font
+        self.loc_title['text'] = "区域:"
+        self.loc_title['bg'] = self.bg_color
+        self.loc_title['anchor'] = 'e'
+
+        self.loc_enter[0]['font'] = title_font
+        self.loc_enter[0]['textvariable'] = self.loc_enter[1]
+
+        self.loc_title.place(relx=0.0, rely=0.02, relwidth=0.3, relheight=0.96)
+        self.loc_enter[0].place(relx=0.3, rely=0.02, relwidth=0.7, relheight=0.96)
+
+    def show_result(self, lst: np.array):
+        passing = float(lst[0][0])
+
+        label = ["通过", "未通过"]
+        not_passing = 1 - passing
+        data = [passing, not_passing]
+
+        res = self.plt.pie(data, radius=1, pctdistance=0.7, textprops=dict(color='w'),  # 不显示文字
+                           shadow=True, startangle=45, autopct="%6.3f%%", wedgeprops=dict(width=0.6, edgecolor="w"))
+        self.plt.legend(res[0], label, loc="lower left")
+        self.canvas.draw()
+        self.toolbar.update()
+
+    def to_program(self):
+        self.refresh()
+
+    def refresh(self, _=None):
+        where = self.loc_enter[1].get()
+        if len(where) == 0:
+            where = "全局"
+            where_ = []
+        else:
+            where_ = [f"Location='{where}'"]
+
+        self.plt.cla()
+        self.plt.set_title(f"{where}垃圾分类通过率")  # 设置标题以及其位置和字体大小
+        event = tk_event.PassingRateEvent(self.station).start([], where_, where_, [], self)
+        self.station.push_event(event)
+
+
+class StatisticsPassRateTypeAndLocProgram(StatisticsUserBaseProgram):
+    def __init__(self, station, win, color):
+        super(StatisticsPassRateTypeAndLocProgram, self).__init__(station, win, color, "通过率-按类型和区域")
+        self.loc_frame = tk.Frame(self.frame)
+        self.loc_title = tk.Label(self.loc_frame)
+        self.loc_enter = tk.Entry(self.loc_frame), tk.StringVar()
+
+    def conf_gui(self, n: int = 1):
+        super(StatisticsPassRateTypeAndLocProgram, self).conf_gui(n)
+        title_font = make_font(size=16)
+
+        self.loc_frame['bg'] = self.bg_color
+        self.loc_frame['relief'] = "ridge"
+        self.loc_frame['bd'] = 5
+        self.loc_frame.place(relx=0.0, rely=0.92, relwidth=0.33, relheight=0.07)
+
+        self.loc_title['font'] = title_font
+        self.loc_title['bg'] = self.bg_color
+        self.loc_title['text'] = "区域:"
+        self.loc_title['anchor'] = 'e'
+
+        self.loc_enter[0]['font'] = title_font
+        self.loc_enter[0]['textvariable'] = self.loc_enter[1]
+
+        self.loc_title.place(relx=0.0, rely=0.02, relwidth=0.3, relheight=0.96)
+        self.loc_enter[0].place(relx=0.3, rely=0.02, relwidth=0.7, relheight=0.96)
+
+    def show_result(self, lst: List[Tuple[bytes, any]]):
+        data_1, data_2, data_3, data_4 = [1.0, 0.0], [1.0, 0.0], [1.0, 0.0], [1.0, 0.0]
+
+        for i in lst:
+            tmp: bytes = i[0]
+            type_ = tmp.decode('utf-8')
+            if type_ == '4':
+                data_4 = [float(i[1]), 1 - float(i[1])]
+            elif type_ == '3':
+                data_3 = [float(i[1]), 1 - float(i[1])]
+            elif type_ == '2':
+                data_2 = [float(i[1]), 1 - float(i[1])]
+            elif type_ == '1':
+                data_1 = [float(i[1]), 1 - float(i[1])]
+
+        legend_text = []
+        for data, r, s in zip([data_1, data_2, data_3, data_4], [0.3, 0.6, 0.9, 1.2], [5, 20, 35, 50]):
+            res = self.plt.pie(data, radius=r, pctdistance=0.7,  # 不显示文字
+                               shadow=True, startangle=s, autopct="%6.3f%%", wedgeprops=dict(width=0.3, edgecolor="w"))
+            legend_text += res[0]
+
+        label = []
+        for i in GarbageType.GarbageTypeStrList_ch[1:]:
+            label.append(f"{i}-通过")
+            label.append(f"{i}-不通过")
+
+        self.plt.legend(legend_text, label)
+        self.canvas.draw()
+        self.toolbar.update()
+
+    def export(self):
+        self.station.show_msg("保存数据", f"数据不支持导出")
+        return
+
+    def to_program(self):
+        self.refresh()
+
+    def refresh(self, _=None):
+        where = self.loc_enter[1].get()
+        if len(where) == 0:
+            where = "全局"
+            where_ = []
+        else:
+            where_ = [f"Location='{where}'"]
+
+        self.plt.cla()
+        self.plt.set_title(f"{where}垃圾分类通过率")  # 设置标题以及其位置和字体大小
+        event = tk_event.PassingRateEvent(self.station).start(["GarbageType"],
+                                                              where_,
+                                                              where_ + ["g.GarbageType=garbage.GarbageType"],
+                                                              ["GarbageType"], self)
+        self.station.push_event(event)
+
+
 all_program = [WelcomeProgram, CreateNormalUserProgram, CreateManagerUserProgram, CreateAutoNormalUserProgram,
                CreateGarbageProgram, DeleteUserProgram, DeleteUsersProgram, DeleteGarbageProgram,
                DeleteGarbageMoreProgram, DeleteAllGarbageProgram, SearchUserProgram, SearchUserAdvancedProgram,
@@ -2371,4 +2592,6 @@ all_program = [WelcomeProgram, CreateNormalUserProgram, CreateManagerUserProgram
                StatisticsTimeLocProgram, StatisticsTimeTypeProgram, StatisticsTimeTypeLocProgram,
                StatisticsTimeCheckResultProgram, StatisticsTimeCheckResultAndTypeProgram,
                StatisticsTimeCheckResultAndLocProgram, StatisticsTimeDetailProgram, StatisticsUserTinyProgram,
-               StatisticsUserLargeProgram, StatisticsScoreDistributedProgram, StatisticsReputationDistributedProgram]
+               StatisticsUserLargeProgram, StatisticsScoreDistributedProgram, StatisticsReputationDistributedProgram,
+               StatisticsPassRateGlobalProgram, StatisticsPassRateTypeProgram, StatisticsPassRateLocProgram,
+               StatisticsPassRateTypeAndLocProgram]