You've already forked HeurAMS-legacy
fix
This commit is contained in:
@@ -5,6 +5,7 @@ Particle 模块 - 粒子对象系统
|
||||
"""
|
||||
|
||||
from heurams.services.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
logger.debug("粒子模块已加载")
|
||||
|
||||
|
||||
@@ -12,6 +12,9 @@ from heurams.services.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
class AtomRegister_runtime(TypedDict):
|
||||
locked: bool # 只读锁定标识符
|
||||
min_rate: int # 最低评分
|
||||
|
||||
class AtomRegister(TypedDict):
|
||||
nucleon: Nucleon
|
||||
@@ -23,7 +26,7 @@ class AtomRegister(TypedDict):
|
||||
orbital: Orbital
|
||||
orbital_path: pathlib.Path
|
||||
orbital_fmt: str
|
||||
runtime: dict
|
||||
runtime: AtomRegister_runtime
|
||||
|
||||
|
||||
class Atom:
|
||||
@@ -51,6 +54,7 @@ class Atom:
|
||||
"orbital": None,
|
||||
"orbital_path": None, # 允许设置为 None, 此时使用 nucleon 文件内的推荐配置
|
||||
"orbital_fmt": "toml",
|
||||
"runtime": {"locked": False, "min_rate": 0x3f3f3f3f}
|
||||
}
|
||||
self.do_eval()
|
||||
logger.debug("Atom 初始化完成")
|
||||
@@ -65,6 +69,38 @@ class Atom:
|
||||
logger.error("尝试链接不受支持的键: '%s'", key)
|
||||
raise ValueError("不受支持的原子元数据链接操作")
|
||||
|
||||
def minimize(self, rating):
|
||||
"""效果等同于 self.registry['runtime']['min_rate'] = min(rating, self.registry['runtime']['min_rate'])
|
||||
|
||||
Args:
|
||||
rating (int): 评分
|
||||
"""
|
||||
self.registry['runtime']['min_rate'] = min(rating, self.registry['runtime']['min_rate'])
|
||||
|
||||
def lock(self, locked = -1):
|
||||
"""锁定, 效果等同于 self.registry['runtime']['locked'] = locked 或者返回是否锁定
|
||||
"""
|
||||
if locked == 1:
|
||||
self.registry['runtime']['locked'] = True
|
||||
return 1
|
||||
elif locked == 0:
|
||||
self.registry['runtime']['locked'] = False
|
||||
return 1
|
||||
elif locked == -1:
|
||||
return self.registry['runtime']["locked"]
|
||||
return 0
|
||||
|
||||
def revise(self):
|
||||
"""执行最终评分
|
||||
PuzzleWidget 的 handler 除了测试, 严禁直接执行 Electron 的 revisor 函数, 否则造成逻辑混乱
|
||||
"""
|
||||
if self.registry["runtime"]["locked"]:
|
||||
logger.debug(f"允许总评分: {self.registry['runtime']['min_rate']}")
|
||||
self.registry["electron"].revisor(self.registry['runtime']["min_rate"])
|
||||
else:
|
||||
logger.debug("禁止总评分")
|
||||
|
||||
|
||||
def do_eval(self):
|
||||
"""
|
||||
执行并以结果替换当前单元的所有 eval 语句
|
||||
@@ -85,13 +121,17 @@ class Atom:
|
||||
# 如果无法获取配置或元数据,使用空字典
|
||||
logger.debug("无法获取配置或元数据,使用空字典")
|
||||
pass
|
||||
try:
|
||||
try:
|
||||
eval_value = eval(s)
|
||||
if isinstance(eval_value, (list, dict)):
|
||||
ret = eval_value
|
||||
else:
|
||||
ret = str(eval_value)
|
||||
logger.debug("eval 执行成功: '%s' -> '%s'", s, str(ret)[:50] + '...' if len(ret) > 50 else ret)
|
||||
logger.debug(
|
||||
"eval 执行成功: '%s' -> '%s'",
|
||||
s,
|
||||
str(ret)[:50] + "..." if len(ret) > 50 else ret,
|
||||
)
|
||||
except Exception as e:
|
||||
ret = f"此 eval 实例发生错误: {e}"
|
||||
logger.warning("eval 执行错误: '%s' -> %s", s, e)
|
||||
@@ -117,16 +157,16 @@ class Atom:
|
||||
|
||||
# 如果 nucleon 存在且有 do_eval 方法,调用它
|
||||
nucleon = self.registry["nucleon"]
|
||||
if nucleon is not None and hasattr(nucleon, 'do_eval'):
|
||||
if nucleon is not None and hasattr(nucleon, "do_eval"):
|
||||
nucleon.do_eval()
|
||||
logger.debug("已调用 nucleon.do_eval")
|
||||
|
||||
|
||||
# 如果 electron 存在且其 algodata 包含 eval 字符串,遍历它
|
||||
electron = self.registry["electron"]
|
||||
if electron is not None and hasattr(electron, 'algodata'):
|
||||
if electron is not None and hasattr(electron, "algodata"):
|
||||
traverse(electron.algodata, eval_with_env)
|
||||
logger.debug("已处理 electron algodata eval")
|
||||
|
||||
|
||||
# 如果 orbital 存在且是字典,遍历它
|
||||
orbital = self.registry["orbital"]
|
||||
if orbital is not None and isinstance(orbital, dict):
|
||||
@@ -173,7 +213,9 @@ class Atom:
|
||||
raise KeyError(f"不支持的键: {key}")
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
logger.debug("Atom.__setitem__: key='%s', value type: %s", key, type(value).__name__)
|
||||
logger.debug(
|
||||
"Atom.__setitem__: key='%s', value type: %s", key, type(value).__name__
|
||||
)
|
||||
if key in self.registry:
|
||||
self.registry[key] = value
|
||||
logger.debug("键 '%s' 已设置", key)
|
||||
|
||||
@@ -17,7 +17,9 @@ class Electron:
|
||||
algodata: 算法数据字典, 包含算法的各项参数和设置
|
||||
algo: 使用的算法模块标识
|
||||
"""
|
||||
logger.debug("创建 Electron 实例,ident: '%s', algo_name: '%s'", ident, algo_name)
|
||||
logger.debug(
|
||||
"创建 Electron 实例,ident: '%s', algo_name: '%s'", ident, algo_name
|
||||
)
|
||||
self.algodata = algodata
|
||||
self.ident = ident
|
||||
self.algo = algorithms[algo_name]
|
||||
@@ -31,11 +33,15 @@ class Electron:
|
||||
self._default_init(self.algo.defaults)
|
||||
else:
|
||||
logger.debug("算法数据已存在,跳过默认初始化")
|
||||
logger.debug("Electron 初始化完成,algodata keys: %s", list(self.algodata.keys()))
|
||||
logger.debug(
|
||||
"Electron 初始化完成,algodata keys: %s", list(self.algodata.keys())
|
||||
)
|
||||
|
||||
def _default_init(self, defaults: dict):
|
||||
"""默认初始化包装"""
|
||||
logger.debug("Electron._default_init: 使用默认值,keys: %s", list(defaults.keys()))
|
||||
logger.debug(
|
||||
"Electron._default_init: 使用默认值,keys: %s", list(defaults.keys())
|
||||
)
|
||||
self.algodata[self.algo.algo_name] = defaults.copy()
|
||||
|
||||
def activate(self):
|
||||
@@ -88,10 +94,16 @@ class Electron:
|
||||
quality (int): 记忆保留率量化参数 (0-5)
|
||||
is_new_activation (bool): 是否为初次激活
|
||||
"""
|
||||
logger.debug("Electron.revisor: ident='%s', quality=%d, is_new_activation=%s",
|
||||
self.ident, quality, is_new_activation)
|
||||
logger.debug(
|
||||
"Electron.revisor: ident='%s', quality=%d, is_new_activation=%s",
|
||||
self.ident,
|
||||
quality,
|
||||
is_new_activation,
|
||||
)
|
||||
self.algo.revisor(self.algodata, quality, is_new_activation)
|
||||
logger.debug("revisor 完成,更新后的 algodata: %s", self.algodata.get(self.algo, {}))
|
||||
logger.debug(
|
||||
"revisor 完成,更新后的 algodata: %s", self.algodata.get(self.algo, {})
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return (
|
||||
|
||||
@@ -40,9 +40,7 @@ def load_nucleon(path: pathlib.Path, fmt="toml"):
|
||||
logger.debug("处理项目: %s", item)
|
||||
lst.append(
|
||||
(
|
||||
Nucleon(
|
||||
item, attr, deepcopy(nested_data["__metadata__"])
|
||||
),
|
||||
Nucleon(item, attr, deepcopy(nested_data["__metadata__"])),
|
||||
deepcopy(nested_data["__metadata__"]["orbital"]),
|
||||
)
|
||||
)
|
||||
|
||||
@@ -14,8 +14,12 @@ class Nucleon:
|
||||
payload: 记忆内容信息
|
||||
metadata: 可选元数据信息
|
||||
"""
|
||||
logger.debug("创建 Nucleon 实例,ident: '%s', payload keys: %s, metadata keys: %s",
|
||||
ident, list(payload.keys()) if payload else [], list(metadata.keys()) if metadata else [])
|
||||
logger.debug(
|
||||
"创建 Nucleon 实例,ident: '%s', payload keys: %s, metadata keys: %s",
|
||||
ident,
|
||||
list(payload.keys()) if payload else [],
|
||||
list(metadata.keys()) if metadata else [],
|
||||
)
|
||||
self.metadata = metadata
|
||||
self.payload = payload
|
||||
self.ident = ident
|
||||
@@ -28,7 +32,9 @@ class Nucleon:
|
||||
return self.ident
|
||||
if key in self.payload:
|
||||
value = self.payload[key]
|
||||
logger.debug("返回 payload['%s'], value type: %s", key, type(value).__name__)
|
||||
logger.debug(
|
||||
"返回 payload['%s'], value type: %s", key, type(value).__name__
|
||||
)
|
||||
return value
|
||||
else:
|
||||
logger.error("键 '%s' 未在 payload 中找到", key)
|
||||
@@ -59,7 +65,11 @@ class Nucleon:
|
||||
ret = str(eval_value)
|
||||
else:
|
||||
ret = eval_value
|
||||
logger.debug("eval 执行成功: '%s' -> '%s'", s, str(ret)[:50] + '...' if len(ret) > 50 else ret)
|
||||
logger.debug(
|
||||
"eval 执行成功: '%s' -> '%s'",
|
||||
s,
|
||||
str(ret)[:50] + "..." if len(ret) > 50 else ret,
|
||||
)
|
||||
except Exception as e:
|
||||
ret = f"此 eval 实例发生错误: {e}"
|
||||
logger.warning("eval 执行错误: '%s' -> %s", s, e)
|
||||
|
||||
Reference in New Issue
Block a user