perf: 脏标记优化 Lict 对象操作复杂度

src/heurams/interface/screens/dashboard.py

@@ -105,20 +105,21 @@ class DashboardScreen(Screen):
         repo.progress = {
             "total": repo.data_length,
             "touched": 0,
+            "have_activated_ever": 0,
         }
         initial_time = float("inf")
         for i in range(repo.data_length):
             e = pt.Electron.from_data(electronic_data=repo.electronic_data_lict[i], algo_name=repo.config['algorithm'])
             n = pt.Nucleon.from_data(repo.nucleonic_data_lict[i])
             if e.is_activated():
+                repo.progress["have_activated_ever"] = 1
                 repo.progress["touched"] += 1
                 repo.nearest_review_time = min(repo.nearest_review_time, e.nextdate())
                 # initial_time = min(initial_time, e.)
         repo.need_review = timer.get_daystamp() >= repo.nearest_review_time
         repo.prompt = f"""{repo.manifest['title']} ({repo.config['algorithm']})
-  进度: {repo.progress['touched']}/{repo.progress['total']} ({round(repo.progress['touched']/repo.progress['total']*100, 1)}%)
-  {'需要学习' if repo.need_review else "无需操作"}
-        """
+  [d]进度: {repo.progress['touched']}/{repo.progress['total']} ({round(repo.progress['touched']/repo.progress['total']*100, 1)}%)[/d]
+  [d]{'需要复习' if repo.need_review else ("暂未开始" if not repo.progress['have_activated_ever'] else '无需操作')}[/d]"""
 
     def on_mount(self) -> None:
         """挂载组件时初始化"""

src/heurams/kernel/auxiliary/lict.py

@@ -1,124 +1,139 @@
-from collections import UserList
-from typing import Any, Iterator
+from collections.abc import MutableSequence
+from typing import Any, Iterator, Optional
 
-
-class Lict(UserList):  # TODO: 优化同步(惰性同步), 当前性能为 O(n)
-    """ "列典" 对象
-
-    同时兼容字典和列表大多数 API, 两边数据同步的容器
-    列表数据是 dictobj.items() 的格式
-    支持根据字典或列表初始化
-    限制要求:
-    - 键名一定唯一, 且仅能为字符串
-    - 值一定是引用对象
-    - 不使用并发
-    - 不在乎列表顺序语义(严格按键名字符序排列)和列表索引查找, 因此外部的 sort, index 等功能不可用
-    - append 的元组中, 表示键名的元素不能重复, 否则会导致覆盖行为
-
-    只有在 Python 3.7+ 中, forced_order 行为才能被取消.
-    """
-
-    def __init__(
-        self,
-        initlist: list | None = None,
-        initdict: dict | None = None,
-        forced_order=False,
-    ):
-        self.dicted_data = {}
-        if initdict != None:
-            initlist = list(initdict.items())
-        super().__init__(initlist=initlist)
+class Lict(MutableSequence):
+    
+    def __init__(self, initlist: Optional[list] = None, 
+        initdict: Optional[dict] = None,
+        forced_order: bool = False):
+        self._dict = {}
+        self._list = []
+        self._dict_dirty = False
+        self._list_dirty = False
         self.forced_order = forced_order
-        self._sync_based_on_list()
-        if self.forced_order:
-            self.data.sort()
+        
+        if initdict: # initdict 更优先
+            self._dict = initdict.copy()
+            self._list_dirty = True
+        elif initlist:
+            self._list = initlist.copy()
+            self._dict_dirty = True
+        
+        self._sync_if_needed()
 
-    def _sync_based_on_dict(self):
-        self.data = list(self.dicted_data.items())
-        if self.forced_order:
-            self.data.sort()
-
-    def _sync_based_on_list(self):
-        self.dicted_data = {}
-        for i in self.data:
-            self.dicted_data[i[0]] = i[1]
-
-    def __iter__(self) -> Iterator:
-        return self.data.__iter__()
-
-    def __getitem__(self, i):
-        if isinstance(i, str):
-            return self.dicted_data[i]
+    def _sync_if_needed(self):
+        """只在需要时同步"""
+        if self._dict_dirty:
+            self._dict = {k: v for k, v in self._list}
+            if self.forced_order:
+                self._list.sort()
+            self._dict_dirty = False
+        
+        if self._list_dirty:
+            self._list = list(self._dict.items())
+            if self.forced_order:
+                self._list.sort()
+            self._list_dirty = False
+    
+    def __getitem__(self, key):
+        if isinstance(key, str):
+            return self._dict[key]
         else:
-            return super().__getitem__(i)
-
-    def get_itemic_unit(self, ident):
-        return (ident, self.dicted_data[ident])
-
-    def __setitem__(self, i, item):
-        if isinstance(i, str):
-            self.dicted_data[i] = item
-            self._sync_based_on_dict()
+            self._sync_if_needed()
+            return self._list[key]
+    
+    def __setitem__(self, key, value):
+        """传入键值对时等同于操作字典, 传入索引+元组时等用于替换某索引的列表值为新元组"""
+        if isinstance(key, str):
+            self._dict[key] = value
+            self._list_dirty = True
         else:
-            if item != (item[0], item[1]):
+            if value != (value[0], value[1]):
                 raise NotImplementedError
-            super().__setitem__(i, item)
-            self._sync_based_on_list()
+            self._sync_if_needed()
+            old_key = self._list[key][0]
+            self._dict.pop(old_key)
+            self._list[key] = value
+            self._dict[value[0]] = value[1] # 避免全量同步
 
-    def __delitem__(self, i):
-        if isinstance(i, str):
-            del self.dicted_data[i]
-            self._sync_based_on_dict()
-        else:
-            super().__delitem__(i)
-            self._sync_based_on_list()
+    def __delitem__(self, key):
+        if isinstance(key, str): # 字符串键
+            del self._dict[key]
+            self._list_dirty = True
+        else: # 数字索引
+            self._sync_if_needed()
+            del_key = self._list[key][0]
+            del self._list[key]
+            del self._dict[del_key]
+    
+    def keys(self):
+        self._sync_if_needed()
+        return self._dict.keys()
+    
+    def values(self):
+        self._sync_if_needed()
+        return self._dict.values()
+    
+    def items(self):
+        self._sync_if_needed()
+        return self._list
+    
+    def __len__(self):
+        self._sync_if_needed()
+        return len(self._list)
+    
+    def __iter__(self):
+        self._sync_if_needed()
+        return iter(self._list)
 
     def __contains__(self, item):
-        return item in self.data or item in self.keys() or item in self.values()
-
-    def append(self, item: Any) -> None:
+        self._sync_if_needed()
+        return item in self._list or item in self.keys() or item in self.values()
+    
+    def append(self, item):
         if item != (item[0], item[1]):
             raise NotImplementedError
-        super().append(item)
-        self._sync_based_on_list()
-        if self.forced_order:
-            self.data.sort()
+        self._sync_if_needed() # 以防 forced_order
+        key, value = item
+        self._dict[key] = value
+        self._list.append(item) # 两端都已同步
+        self._sync_if_needed() # 以防 forced_order
 
-    def append_new(self, item: Any):
+    def append_if_it_donesnt_exist_before(self, item: Any):
         if item != (item[0], item[1]):
             raise NotImplementedError
+        self._sync_if_needed()
         if item[0] not in self:
             super().append(item)
-            self._sync_based_on_list()
-            if self.forced_order:
-                self.data.sort()
-
+        self._sync_if_needed()
+    
     def insert(self, i: int, item: Any) -> None:
-        if item != (item[0], item[1]):  # 确保 item 是遵从限制的元组
+        if item != (item[0], item[1]):
             raise NotImplementedError
-        super().insert(i, item)
-        self._sync_based_on_list()
-        if self.forced_order:
-            self.data.sort()
+        self._sync_if_needed()
+        self._list.insert(i, item)
+        self._dict_dirty = True
 
     def pop(self, i: int = -1) -> Any:
-        res = super().pop(i)
-        self._sync_based_on_list()
+        self._sync_if_needed()
+        res = self._list.pop(i)
+        self._dict_dirty = True
         return res
 
     def remove(self, item: Any) -> None:
         if isinstance(item, str):
-            item = (item, self.dicted_data[item])
+            item = (item, self._dict[item])
         if item != (item[0], item[1]):
             raise NotImplementedError
-        super().remove(item)
-        self._sync_based_on_list()
-        if self.forced_order:
-            self.data.sort()
+        self._sync_if_needed()
+        self._list.remove(item)
+        self._dict_dirty = True
 
     def clear(self) -> None:
-        super().clear()
-        self._sync_based_on_list()
+        self._dict.clear()
+        self._list.clear()
+        self._dict_dirty = False
+        self._list_dirty = False
 
     def index(self):
         raise NotImplementedError
@@ -133,15 +148,22 @@ class Lict(UserList):  # TODO: 优化同步(惰性同步), 当前性能为 O(n)
         raise NotImplementedError
 
     def keys(self):
-        return self.dicted_data.keys()
+        self._sync_if_needed()
+        return self._dict.keys()
 
     def values(self):
-        return self.dicted_data.values()
+        self._sync_if_needed()
+        return self._dict.values()
 
     def items(self):
-        return self.data
+        self._sync_if_needed()
+        return self._list
+
+    def get_itemic_unit(self, ident):
+        return (ident, self._dict[ident])
 
     def keys_equal_with(self, other):
+        self._sync_if_needed()
         return self.key_equality(self, other)
 
     @classmethod

src/heurams/kernel/auxiliary/lict_legacy.py

@@ -0,0 +1,140 @@
+from collections import UserList
+from typing import Any, Iterator
+
+
+class Lict(UserList):  # TODO: 优化同步(惰性同步), 当前性能为 O(n)
+    """ "列典" 对象
+
+    同时兼容字典和列表大多数 API, 两边数据同步的容器
+    列表数据是 dictobj.items() 的格式
+    支持根据字典或列表初始化
+    限制要求:
+    - 键名一定唯一, 且仅能为字符串
+    - 值一定是引用对象
+    - 不使用并发
+    - 不在乎列表顺序语义(严格按键名字符序排列)和列表索引查找, 因此外部的 sort, index 等功能不可用
+    - append 的元组中, 表示键名的元素不能重复, 否则会导致覆盖行为
+
+    """
+
+    def __init__(
+        self,
+        initlist: list | None = None,
+        initdict: dict | None = None,
+    ):
+        self.dicted_data = {}
+        if initdict != None:
+            initlist = list(initdict.items())
+        super().__init__(initlist=initlist)
+        self._sync_based_on_list()
+        self.data.sort()
+
+    def _sync_based_on_dict(self):
+        self.data = list(self.dicted_data.items())
+        self.data.sort()
+
+    def _sync_based_on_list(self):
+        self.dicted_data = {}
+        for i in self.data:
+            self.dicted_data[i[0]] = i[1]
+
+    def __iter__(self) -> Iterator:
+        return self.data.__iter__()
+
+    def __getitem__(self, i):
+        if isinstance(i, str):
+            return self.dicted_data[i]
+        else:
+            return super().__getitem__(i)
+
+    def get_itemic_unit(self, ident):
+        return (ident, self.dicted_data[ident])
+
+    def __setitem__(self, i, item):
+        if isinstance(i, str):
+            self.dicted_data[i] = item
+            self._sync_based_on_dict()
+        else:
+            if item != (item[0], item[1]):
+                raise NotImplementedError
+            super().__setitem__(i, item)
+            self._sync_based_on_list()
+
+    def __delitem__(self, i):
+        if isinstance(i, str):
+            del self.dicted_data[i]
+            self._sync_based_on_dict()
+        else:
+            super().__delitem__(i)
+            self._sync_based_on_list()
+
+    def __contains__(self, item):
+        return item in self.data or item in self.keys() or item in self.values()
+
+    def append(self, item: Any) -> None:
+        if item != (item[0], item[1]):
+            raise NotImplementedError
+        super().append(item)
+        self._sync_based_on_list()
+        self.data.sort()
+
+    def append_new(self, item: Any):
+        if item != (item[0], item[1]):
+            raise NotImplementedError
+        if item[0] not in self:
+            super().append(item)
+            self._sync_based_on_list()
+            self.data.sort()
+
+    def insert(self, i: int, item: Any) -> None:
+        if item != (item[0], item[1]):  # 确保 item 是遵从限制的元组
+            raise NotImplementedError
+        super().insert(i, item)
+        self._sync_based_on_list()
+        self.data.sort()
+
+    def pop(self, i: int = -1) -> Any:
+        res = super().pop(i)
+        self._sync_based_on_list()
+        return res
+
+    def remove(self, item: Any) -> None:
+        if isinstance(item, str):
+            item = (item, self.dicted_data[item])
+        if item != (item[0], item[1]):
+            raise NotImplementedError
+        super().remove(item)
+        self._sync_based_on_list()
+        self.data.sort()
+
+    def clear(self) -> None:
+        super().clear()
+        self._sync_based_on_list()
+
+    def index(self):
+        raise NotImplementedError
+
+    def extend(self):
+        raise NotImplementedError
+
+    def sort(self):
+        raise NotImplementedError
+
+    def reverse(self):
+        raise NotImplementedError
+
+    def keys(self):
+        return self.dicted_data.keys()
+
+    def values(self):
+        return self.dicted_data.values()
+
+    def items(self):
+        return self.data
+
+    def keys_equal_with(self, other):
+        return self.key_equality(self, other)
+
+    @classmethod
+    def key_equality(cls, a, b):
+        return a.keys() == b.keys()

src/heurams/kernel/particles/electron.py

@@ -5,6 +5,7 @@ import heurams.kernel.algorithms as algolib
 import heurams.services.timer as timer
 from heurams.kernel.algorithms import algorithms
 from heurams.services.logger import get_logger
+from heurams.context import config_var
 
 logger = get_logger(__name__)
 

src/heurams/kernel/repolib/repo.py

@@ -82,7 +82,7 @@ class Repo:
         self.data_length = len(self.nucleonic_data_lict)
         self.ident_index = self.nucleonic_data_lict.keys()
         for i in self.ident_index:
-            self.algodata.append_new((i, {}))
+            self.algodata.append_if_it_donesnt_exist_before((i, {}))
         self.electronic_data_lict = self.algodata
 
     def _nucleonic_proc(self, unit):