You've already forked HeurAMS-legacy
style: 代码格式化
This commit is contained in:
@@ -21,4 +21,4 @@ __all__ = [
|
||||
"load_nucleon",
|
||||
"load_electron",
|
||||
"atom_registry",
|
||||
]
|
||||
]
|
||||
|
||||
@@ -9,6 +9,7 @@ import json
|
||||
import bidict
|
||||
from heurams.context import config_var
|
||||
|
||||
|
||||
class AtomRegister(TypedDict):
|
||||
nucleon: Nucleon
|
||||
nucleon_path: pathlib.Path
|
||||
@@ -21,7 +22,8 @@ class AtomRegister(TypedDict):
|
||||
orbital_fmt: str
|
||||
runtime: dict
|
||||
|
||||
class Atom():
|
||||
|
||||
class Atom:
|
||||
"""
|
||||
统一处理一系列对象的所有信息与持久化:
|
||||
关联电子 (算法数据)
|
||||
@@ -30,11 +32,11 @@ class Atom():
|
||||
以及关联路径
|
||||
"""
|
||||
|
||||
def __init__(self, ident = ""):
|
||||
def __init__(self, ident=""):
|
||||
self.ident = ident
|
||||
atom_registry[ident] = self
|
||||
# self.is_evaled = False
|
||||
self.registry: AtomRegister = { # type: ignore
|
||||
self.registry: AtomRegister = { # type: ignore
|
||||
"nucleon": None,
|
||||
"nucleon_path": None,
|
||||
"nucleon_fmt": "toml",
|
||||
@@ -42,7 +44,7 @@ class Atom():
|
||||
"electron_path": None,
|
||||
"electron_fmt": "json",
|
||||
"orbital": None,
|
||||
"orbital_path": None, # 允许设置为 None, 此时使用 nucleon 文件内的推荐配置
|
||||
"orbital_path": None, # 允许设置为 None, 此时使用 nucleon 文件内的推荐配置
|
||||
"orbital_fmt": "toml",
|
||||
}
|
||||
self.do_eval()
|
||||
@@ -53,16 +55,17 @@ class Atom():
|
||||
self.do_eval()
|
||||
else:
|
||||
raise ValueError("不受支持的原子元数据链接操作")
|
||||
|
||||
|
||||
def do_eval(self):
|
||||
"""
|
||||
执行并以结果替换当前单元的所有 eval 语句
|
||||
TODO: 带有限制的 eval, 异步/多线程执行避免堵塞
|
||||
"""
|
||||
|
||||
# eval 环境设置
|
||||
def eval_with_env(s: str):
|
||||
try:
|
||||
nucleon = self.registry['nucleon']
|
||||
nucleon = self.registry["nucleon"]
|
||||
default = config_var.get()["puzzles"]
|
||||
metadata = nucleon.metadata
|
||||
except:
|
||||
@@ -72,7 +75,7 @@ class Atom():
|
||||
except Exception as e:
|
||||
ret = f"此 eval 实例发生错误: {e}"
|
||||
return ret
|
||||
|
||||
|
||||
def traverse(data, modifier):
|
||||
if isinstance(data, dict):
|
||||
for key, value in data.items():
|
||||
@@ -89,10 +92,9 @@ class Atom():
|
||||
if data.startswith("eval:"):
|
||||
return modifier(data[5:])
|
||||
return data
|
||||
|
||||
|
||||
traverse(self.registry["nucleon"], eval_with_env)
|
||||
traverse(self.registry["orbital"], eval_with_env)
|
||||
|
||||
|
||||
def persist(self, key):
|
||||
path: pathlib.Path | None = self.registry[key + "_path"]
|
||||
@@ -109,7 +111,7 @@ class Atom():
|
||||
raise KeyError("不受支持的持久化格式")
|
||||
else:
|
||||
raise TypeError("对未初始化的路径对象操作")
|
||||
|
||||
|
||||
def __getitem__(self, key):
|
||||
if key in self.registry:
|
||||
return self.registry[key]
|
||||
@@ -124,5 +126,6 @@ class Atom():
|
||||
@staticmethod
|
||||
def placeholder():
|
||||
return (Electron.placeholder(), Nucleon.placeholder(), {})
|
||||
|
||||
|
||||
|
||||
atom_registry: bidict.bidict[str, Atom] = bidict.bidict()
|
||||
|
||||
@@ -2,12 +2,13 @@ import heurams.services.timer as timer
|
||||
from heurams.context import config_var
|
||||
from heurams.kernel.algorithms import algorithms
|
||||
|
||||
|
||||
class Electron:
|
||||
"""电子: 记忆分析元数据及算法"""
|
||||
|
||||
def __init__(self, ident: str, algodata: dict = {}, algo_name: str = "supermemo2"):
|
||||
"""初始化电子对象 (记忆数据)
|
||||
|
||||
|
||||
Args:
|
||||
ident: 算法的唯一标识符, 用于区分不同的算法实例, 使用 algodata[ident] 获取
|
||||
algodata: 算法数据字典, 包含算法的各项参数和设置
|
||||
@@ -28,34 +29,34 @@ class Electron:
|
||||
|
||||
def activate(self):
|
||||
"""激活此电子"""
|
||||
self.algodata[self.algo]['is_activated'] = 1
|
||||
self.algodata[self.algo]['last_modify'] = timer.get_timestamp()
|
||||
self.algodata[self.algo]["is_activated"] = 1
|
||||
self.algodata[self.algo]["last_modify"] = timer.get_timestamp()
|
||||
|
||||
def modify(self, var: str, value):
|
||||
"""修改 algodata[algo] 中子字典数据"""
|
||||
if var in self.algodata[self.algo]:
|
||||
self.algodata[self.algo][var] = value
|
||||
self.algodata[self.algo]['last_modify'] = timer.get_timestamp()
|
||||
self.algodata[self.algo]["last_modify"] = timer.get_timestamp()
|
||||
else:
|
||||
print(f"警告: '{var}' 非已知元数据字段")
|
||||
|
||||
def is_due(self):
|
||||
"""是否应该复习"""
|
||||
return self.algo.is_due(self.algodata)
|
||||
|
||||
|
||||
def is_activated(self):
|
||||
return self.algodata[self.algo]['is_activated']
|
||||
|
||||
return self.algodata[self.algo]["is_activated"]
|
||||
|
||||
def rate(self):
|
||||
"评价"
|
||||
return self.algo.rate(self.algodata)
|
||||
|
||||
|
||||
def nextdate(self) -> int:
|
||||
return self.algo.nextdate(self.algodata)
|
||||
|
||||
def revisor(self, quality: int = 5, is_new_activation: bool = False):
|
||||
"""算法迭代决策机制实现
|
||||
|
||||
|
||||
Args:
|
||||
quality (int): 记忆保留率量化参数 (0-5)
|
||||
is_new_activation (bool): 是否为初次激活
|
||||
@@ -93,7 +94,7 @@ class Electron:
|
||||
if key == "ident":
|
||||
raise AttributeError("ident 应为只读")
|
||||
self.algodata[self.algo][key] = value
|
||||
self.algodata[self.algo]['last_modify'] = timer.get_timestamp()
|
||||
self.algodata[self.algo]["last_modify"] = timer.get_timestamp()
|
||||
|
||||
def __len__(self):
|
||||
"""仅返回当前算法的配置数量"""
|
||||
@@ -102,4 +103,4 @@ class Electron:
|
||||
@staticmethod
|
||||
def placeholder():
|
||||
"""生成一个电子占位符"""
|
||||
return Electron("电子对象样例内容", {})
|
||||
return Electron("电子对象样例内容", {})
|
||||
|
||||
@@ -6,17 +6,18 @@ import toml
|
||||
import json
|
||||
from copy import deepcopy
|
||||
|
||||
def load_nucleon(path: pathlib.Path, fmt = "toml"):
|
||||
|
||||
def load_nucleon(path: pathlib.Path, fmt="toml"):
|
||||
with open(path, "r") as f:
|
||||
dictdata = dict()
|
||||
dictdata = toml.load(f) # type: ignore
|
||||
dictdata = toml.load(f) # type: ignore
|
||||
lst = list()
|
||||
nested_data = dict()
|
||||
# 修正 toml 解析器的不管嵌套行为
|
||||
for key, value in dictdata.items():
|
||||
if "__metadata__" in key: # 以免影响句号
|
||||
if '.' in key:
|
||||
parts = key.split('.')
|
||||
if "__metadata__" in key: # 以免影响句号
|
||||
if "." in key:
|
||||
parts = key.split(".")
|
||||
current = nested_data
|
||||
for part in parts[:-1]:
|
||||
if part not in current:
|
||||
@@ -29,23 +30,31 @@ def load_nucleon(path: pathlib.Path, fmt = "toml"):
|
||||
for item, attr in nested_data.items():
|
||||
if item == "__metadata__":
|
||||
continue
|
||||
lst.append((Nucleon(hasher.hash(item), attr, deepcopy(nested_data['__metadata__'])), deepcopy(nested_data["__metadata__"]["orbital"])))
|
||||
lst.append(
|
||||
(
|
||||
Nucleon(
|
||||
hasher.hash(item), attr, deepcopy(nested_data["__metadata__"])
|
||||
),
|
||||
deepcopy(nested_data["__metadata__"]["orbital"]),
|
||||
)
|
||||
)
|
||||
return lst
|
||||
|
||||
def load_electron(path: pathlib.Path, fmt = "json") -> dict:
|
||||
|
||||
def load_electron(path: pathlib.Path, fmt="json") -> dict:
|
||||
"""从文件路径加载电子对象
|
||||
|
||||
Args:
|
||||
path (pathlib.Path): 路径
|
||||
fmt (str): 文件格式(可选, 默认 json)
|
||||
|
||||
|
||||
Returns:
|
||||
dict: 键名是电子对象名称, 值是电子对象
|
||||
"""
|
||||
with open(path, "r") as f:
|
||||
dictdata = dict()
|
||||
dictdata = json.load(f) # type: ignore
|
||||
dictdata = json.load(f) # type: ignore
|
||||
dic = dict()
|
||||
for item, attr in dictdata.items():
|
||||
dic[item] = (Electron(hasher.hash(item), attr))
|
||||
return dic
|
||||
dic[item] = Electron(hasher.hash(item), attr)
|
||||
return dic
|
||||
|
||||
@@ -29,12 +29,13 @@ class Nucleon:
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.ident)
|
||||
|
||||
|
||||
def do_eval(self):
|
||||
"""
|
||||
执行并以结果替换当前单元的所有 eval 语句
|
||||
TODO: 带有限制的 eval, 异步/多线程执行避免堵塞
|
||||
"""
|
||||
|
||||
# eval 环境设置
|
||||
def eval_with_env(s: str):
|
||||
try:
|
||||
@@ -43,7 +44,7 @@ class Nucleon:
|
||||
except Exception as e:
|
||||
ret = f"此 eval 实例发生错误: {e}"
|
||||
return ret
|
||||
|
||||
|
||||
def traverse(data, modifier):
|
||||
if isinstance(data, dict):
|
||||
for key, value in data.items():
|
||||
@@ -60,9 +61,10 @@ class Nucleon:
|
||||
if data.startswith("eval:"):
|
||||
return modifier(data[5:])
|
||||
return data
|
||||
|
||||
|
||||
traverse(self.payload, eval_with_env)
|
||||
traverse(self.metadata, eval_with_env)
|
||||
|
||||
@staticmethod
|
||||
def placeholder():
|
||||
"""生成一个占位原子核"""
|
||||
|
||||
@@ -1,14 +1,16 @@
|
||||
from typing import TypedDict
|
||||
|
||||
|
||||
class OrbitalSchedule(TypedDict):
|
||||
quick_review: list
|
||||
recognition: list
|
||||
final_review: list
|
||||
|
||||
|
||||
class Orbital(TypedDict):
|
||||
schedule: OrbitalSchedule
|
||||
puzzles: dict
|
||||
|
||||
|
||||
|
||||
"""一份示例
|
||||
["__metadata__.orbital.puzzles"] # 谜题定义
|
||||
@@ -20,4 +22,4 @@ class Orbital(TypedDict):
|
||||
quick_review = [["FillBlank", "1.0"], ["SelectMeaning", "0.5"], ["recognition", "1.0"]]
|
||||
recognition = [["recognition", "1.0"]]
|
||||
final_review = [["FillBlank", "0.7"], ["SelectMeaning", "0.7"], ["recognition", "1.0"]]
|
||||
"""
|
||||
"""
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from heurams.context import config_var
|
||||
import pathlib
|
||||
|
||||
|
||||
def probe_by_filename(filename):
|
||||
"""探测指定文件 (无扩展名) 的所有信息"""
|
||||
paths: dict = config_var.get().get("paths")
|
||||
@@ -8,17 +9,18 @@ def probe_by_filename(filename):
|
||||
result = {}
|
||||
for item, attr in paths.items():
|
||||
for i in formats:
|
||||
attr: pathlib.Path = pathlib.Path(attr) / filename + '.' + i
|
||||
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):
|
||||
|
||||
def probe_all(is_stem=1):
|
||||
"""依据目录探测所有信息
|
||||
|
||||
|
||||
Args:
|
||||
is_stem (boolean): 是否**删除**文件扩展名
|
||||
|
||||
|
||||
Returns:
|
||||
dict: 有三项, 每一项的键名都是文件组类型, 值都是文件组列表, 只包含文件名
|
||||
"""
|
||||
@@ -35,7 +37,9 @@ def probe_all(is_stem = 1):
|
||||
result[item.replace("_dir", "")].append(str(i.name))
|
||||
return result
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import os
|
||||
|
||||
print(os.getcwd())
|
||||
print(probe_all())
|
||||
|
||||
Reference in New Issue
Block a user