部分移植

.gitignore

@@ -12,10 +12,11 @@ build/
 dist/
 
 # Project specific directories
-config/
+# config/
 data/cache/
-data/electrion/
+data/electron/
 data/nucleon/
+data/orbital/
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

config/config.toml

@@ -0,0 +1,21 @@
+# [调试] 将更改保存到文件
+persist_to_file = 1
+
+# [调试] 覆写时间, 设为 -1 以禁用
+daystamp_override = -1
+timestamp_override = -1
+
+# [调试] 一键通过
+quick_pass = 0
+
+# 对于每个项目的默认新记忆原子数量
+tasked_number = 8
+
+# UTC 时间戳修正 仅用于 UNIX 日时间戳的生成修正, 单位为秒
+timezone_offset = +28800 # 中国标准时间 (UTC+8)
+
+[paths] # 相对于配置文件的 ".." (即工作目录) 而言 或绝对路径
+nucleon_dir = "./data/nucleon"
+electron_dir = "./data/electron"
+orbital_dir = "./data/orbital"
+cache_dir = "./data/cache"

src/heurams/context.py

@@ -1,11 +1,15 @@
 """
 全局上下文管理模块
+以及基准路径
 """
 from contextvars import ContextVar
-from typing import Optional
+import pathlib
 from heurams.services.config import ConfigFile
 
-config_var: ContextVar[ConfigFile] = ContextVar('config_var', default=ConfigFile("")) # 配置文件
+root_dir = pathlib.Path(__file__).parent
+working_dir = pathlib.Path.cwd()
+
+config_var: ContextVar[ConfigFile] = ContextVar('config_var', default=ConfigFile(working_dir / "config" / "config.toml")) # 配置文件
 
 runtime_var: ContextVar = ContextVar('runtime_var', default=None) # 运行时共享数据
 
@@ -13,6 +17,7 @@ class ConfigContext:
     """
     功能完备的上下文管理器
     用于临时切换配置的作用域, 支持嵌套使用
+
     Example:
         >>> with ConfigContext(test_config):
         ...     get_daystamp()  # 使用 test_config

src/heurams/kernel/particles/__init__.py

@@ -1,4 +1,4 @@
 from .electron import Electron
 from .nucleon import Nucleon
-from .unions import ElectronUnion, NucleonUnion
-from .atom import Atom
+from .atom import Atom
+from .probe import probe_all, probe_by_filename

src/heurams/kernel/particles/atom.py

@@ -1,13 +1,17 @@
-from electron import Electron
-from nucleon import Nucleon
+from .electron import Electron
+from .nucleon import Nucleon
+import pathlib
+import typing
+import toml
+import json
 
 class Atom():
     """
-    一个静态类, 包含一个原子对象的所有信息:
+    统一处理一系列对象的所有信息与持久化:
         关联电子 (算法数据)
         关联核子 (内容数据)
         关联轨道 (策略数据)
-        关联路径 ()
+        以及关联路径
     """
 
     def __init__(self, ident = ""):
@@ -15,15 +19,36 @@ class Atom():
         self.register = {
             "nucleon": None,
             "nucleon_path": None,
+            "nucleon_fmt": "toml",
             "electron": None,
             "electron_path": None,
-            "orbital": None,            
+            "electron_fmt": "json",
+            "orbital": None,
+            "orbital_path": None, # 允许设置为 None, 此时使用 nucleon 文件内的推荐配置
+            "orbital_fmt": "toml",
         }
 
     def link(self, key, value):
-        self.register[key] = value
-    
+        if key in self.register.keys():
+            self.register[key] = value
+        else:
+            raise ValueError("不受支持的原子元数据链接操作")
     
+    def persist(self, key):
+        path: pathlib.Path | None = self.register[key + "_path"]
+        if isinstance(path, pathlib.Path):
+            path = typing.cast(pathlib.Path, path)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            if self.register[key + "_fmt"] == "toml":
+                with open(path, "w") as f:
+                    toml.dump(self.register[key], f)
+            elif self.register[key + "_fmt"] == "json":
+                with open(path, "w") as f:
+                    json.dump(self.register[key], f)
+            else:
+                raise KeyError("不受支持的持久化格式")
+        else:
+            raise TypeError("对未初始化的路径对象操作")
 
     @staticmethod
     def placeholder():

src/heurams/kernel/particles/electron.py

@@ -117,11 +117,11 @@ class Electron:
         if key in self.algodata[self.algo]:
             return self.algodata[self.algo][key]
         else:
-            raise KeyError(f"Key '{key}' not found in algodata[self.algo].")
+            raise KeyError(f"键 '{key}' 未在 algodata[self.algo] 中")
 
     def __setitem__(self, key, value):
         if key == "ident":
-            raise AttributeError("ident should be readonly")
+            raise AttributeError("ident 应为只读")
         self.algodata[self.algo][key] = value
         self.algodata[self.algo]['last_modify'] = timer.get_timestamp()
 

src/heurams/kernel/particles/loader.py

@@ -0,0 +1,16 @@
+from .nucleon import Nucleon
+from .electron import Electron
+import heurams.services.hasher as hasher
+import pathlib
+import toml
+import json
+
+def loader_nucleon(path: pathlib.Path, fmt = "toml"):
+    with open(path, "r") as f:
+        dictdata = dict()
+        toml.load(f, dictdata) # type: ignore
+        lst = list()
+        for item, attr in dictdata.items():
+            if item == "__metadata__":
+                continue
+            lst.append(Nucleon(hasher.hash(item), attr))

src/heurams/kernel/particles/nucleon.py

@@ -7,7 +7,6 @@ class Nucleon:
         Args:
             ident: 唯一标识符
             metadata: 记忆内容信息
-            orbital: 记忆策略信息 (电子轨道)
         """
         self.metadata = metadata
         self.ident = ident

src/heurams/kernel/particles/probe.py

@@ -0,0 +1,34 @@
+from heurams.context import config_var
+import pathlib
+
+def probe_by_filename(filename):
+    """探测指定文件的所有信息"""
+    paths: dict = config_var.get().get("paths")
+    formats = ["toml", "json"]
+    result = {}
+    for item, attr in paths.items():
+        for i in formats:
+            attr: pathlib.Path = pathlib.Path(attr) / filename + '.' + i
+            if attr.exists():
+                result[item.replace("_dir", "")] = str(attr)
+    return result
+
+def probe_all(is_stem = 1):
+    """依据目录探测所有信息"""
+    paths: dict = config_var.get().get("paths")
+    result = {}
+    for item, attr in paths.items():
+        attr: pathlib.Path = pathlib.Path(attr)
+        result[item.replace("_dir", "")] = list()
+        for i in attr.iterdir():
+            if not i.is_dir():
+                if is_stem:
+                    result[item.replace("_dir", "")].append(str(i.stem))
+                else:
+                    result[item.replace("_dir", "")].append(str(i.name))
+    return result
+
+if __name__ == "__main__":
+    import os
+    print(os.getcwd())
+    print(probe_all())

src/heurams/kernel/particles/unions.py

@@ -1,90 +0,0 @@
-from electron import Electron
-from nucleon import Nucleon
-import pathlib
-import toml
-from typing import List
-import heurams.services.timer as timer
-
-class NucleonUnion():
-    """
-    替代原有 NucleonFile 类, 支持复杂逻辑
-
-    Attributes:
-        path (Path): 对应于 NucleonUnion 实例的文件路径。
-        name (str): 核联对象的显示名称，从文件名中派生。
-        nucleons (list): 内部核子对象的列表。
-        nucleons_dict (dict): 内部核子对象的字典，以核子内容作为键。
-        keydata (dict): 核子对象字典键名的翻译。
-        testdata (dict): 记忆测试项目的元数据。
-
-    Parameters:
-        path (Path): 包含核子数据的文件路径。
-    """
-
-    def __init__(self, path: pathlib.Path):
-        self.path = path
-        self.name = path.name.replace(path.suffix, "")
-        with open(path, 'r') as f:
-            all = toml.load(f)
-            lst = list()
-            for i in all.keys():
-                if "attr" in i:
-                    continue
-                if "data" in i:
-                    continue
-                lst.append(Nucleon(i, all[i]))
-            self.keydata = all["keydata"]
-            self.testdata = all["testdata"]
-            self.nucleons: List[Nucleon] = lst
-            self.nucleons_dict = {i.content: i for i in lst}
-
-    def __len__(self):
-        return len(self.nucleons)
-    
-    def linked_electron_union(self):
-        if (self.path.parent / '..' / 'electron' / self.path.name).exists():
-            return ElectronUnion(self.path.parent / '..' / 'electron' / self.path.name)
-        else:
-            return 0
-
-    def save(self):
-        with open(self.path, 'w') as f:
-            tmp = {i.content: i.metadata for i in self.nucleons}
-            toml.dump(tmp, f)
-
-
-class ElectronUnion:
-    """取代原有 ElectronFile 类, 以支持复杂逻辑"""
-    def __init__(self, path):
-        self.path = path
-        print(path)
-        self.name = path.name.replace(path.suffix, "")
-        with open(path, 'r') as f:
-            all = toml.load(f)
-            lst = list()
-            for i in all.keys():
-                if i != "total":
-                    lst.append(Electron(i, all[i]))
-            self.total = all.get("total", {"last_date": 0})
-            self.electrons = lst
-            self.electrons_dict = {i.content: i for i in lst}
-
-    def sync(self):
-        """同步 electrons_dict 中新增对到 electrons 中, 仅用于缺省初始化不存在映射时调用"""
-        self.electrons = self.electrons_dict.values()
-    
-    def __len__(self):
-        return len(self.electrons)
-    
-    def linked_nucleon_union(self):
-        return NucleonUnion(self.path.parent / '..' / 'nucleon' / self.path.name)
-
-    def save(self):
-        # print(1)
-        self.total["last_date"] = timer.get_daystamp()
-        with open(self.path, 'w') as f:
-            tmp = {i.content: i.metadata for i in self.electrons}
-            tmp["total"] = self.total
-            # print(tmp)
-            toml.dump(tmp, f)
-

src/heurams/services/config.py

@@ -4,8 +4,8 @@ import toml
 import typing
 
 class ConfigFile:
-    def __init__(self, path: str):
-        self.path = pathlib.Path(path)
+    def __init__(self, path: pathlib.Path):
+        self.path = path
         if not self.path.exists():
             self.path.touch()
         self.data = dict()

src/heurams/services/hasher.py

@@ -2,4 +2,7 @@
 import hashlib
 
 def get_md5(text):
+    return hashlib.md5(text.encode('utf-8')).hexdigest()
+
+def hash(text):
     return hashlib.md5(text.encode('utf-8')).hexdigest()