ajax/HeurAMS
1
0
Fork 0
Code Releases 4 Wiki Activity

feat(kernel): 状态机改进

Browse Source
This commit is contained in:
Wang Zhiyu
2026-01-03 05:05:41 +08:00
parent eced6130f1
commit aacf4fdbdf
19 changed files with 428 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
examples/data/config/config.toml Normal file
View File

@@ -0,0 +1,56 @@
# [调试] 将更改保存到文件
persist_to_file = 1
# [调试] 覆写时间, 设为 -1 以禁用
daystamp_override = -1
timestamp_override = -1
# [调试] 一键通过
quick_pass = 1
# 对于每个项目的默认新记忆原子数量
scheduled_num = 8
# UTC 时间戳修正 仅用于 UNIX 日时间戳的生成修正, 单位为秒
timezone_offset = +28800 # 中国标准时间 (UTC+8)
[interface]
[interface.memorizor]
autovoice = true # 自动语音播放, 仅限于 recognition 组件
[algorithm]
default = "SM-2" # 主要算法; 可选项: SM-2, SM-15M, FSRS
[puzzles] # 谜题默认配置
[puzzles.mcq]
max_riddles_num = 2
[puzzles.cloze]
min_denominator = 3
[paths] # 相对于配置文件的 ".." (即工作目录) 而言 或绝对路径
data = "./data"
[services] # 定义服务到提供者的映射
audio = "playsound" # 可选项: playsound(通用), termux(仅用于支持 Android Termux), mpg123(TODO)
tts = "edgetts" # 可选项: edgetts
llm = "openai" # 可选项: openai
sync = "webdav" # 可选项: 留空, webdav
[providers.tts.edgetts] # EdgeTTS 设置
voice = "zh-CN-XiaoxiaoNeural" # 可选项: zh-CN-YunjianNeural (男声), zh-CN-XiaoxiaoNeural (女声)
[providers.llm.openai] # 与 OpenAI 相容的语言模型接口服务设置
url = ""
key = ""
[providers.sync.webdav] # WebDAV 同步设置
url = ""
username = ""
password = ""
remote_path = "/heurams/"
verify_ssl = true
[sync]

14
examples/jiebatest.py Normal file
View File

@@ -0,0 +1,14 @@
# encoding=utf-8
import jieba
#jieba.enable_paddle()# 启动paddle模式。 0.40版之后开始支持，早期版本不支持
strs=["我来到北京清华大学","乒乓球拍卖完了","中国科学技术大学"]
#for str in strs:
# seg_list = jieba.cut(str,use_paddle=True) # 使用paddle模式
# print("Paddle Mode: " + '/'.join(list(seg_list)))
seg_list = jieba.cut("秦孝公据崤函之固, 拥雍州之地", cut_all=False)
print("Default Mode: " + "/ ".join(seg_list)) # 精确模式
seg_list = jieba.cut("他来到了网易杭研大厦") # 默认是精确模式
print(", ".join(seg_list))

7
examples/simplemem.py
View File

@@ -5,8 +5,9 @@ repo = repolib.Repo.create_from_repodir(Path('./test_repo'))
for i in repo.ident_index:
n = pt.Nucleon.create_on_nucleonic_data(nucleonic_data=repo.nucleonic_data_lict.get_itemic_unit(i))
e = pt.Electron.create_on_electonic_data(electronic_data=repo.electronic_data_lict.get_itemic_unit(i))
a = pt.Atom(n, e, repo.orbitic_data)
e.activate()
e.revisor(5, True)
print(repr(n))
print(repr(e))
print(repo)
print(repr(a))
#print(repr(e))
#print(repo)

2
src/heurams/interface/__init__.py
View File

@@ -38,7 +38,7 @@ class HeurAMSApp(App):
("d", "toggle_dark", "切换色调"),
("1", "app.push_screen('dashboard')", "仪表盘"),
("2", "app.push_screen('precache_all')", "缓存管理器"),
("3", "app.push_screen('nucleon_creator')", "创建新仓库"),
("3", "app.push_screen('repo_creator')", "创建新仓库"),
# ("4", "app.push_screen('synctool')", "同步工具"),
("0", "app.push_screen('about')", "版本信息"),
]

2
src/heurams/interface/screens/about.py
View File

@@ -20,7 +20,7 @@ class AboutScreen(Screen):
版本 {version.ver} {version.stage.capitalize()}
开发代号: {version.codename.capitalize()}
开发代号: {version.codename.capitalize()} {version.codename_cn}
一个基于启发式算法的开放源代码记忆调度器, 旨在帮助用户更高效地进行记忆工作与学习规划.

20
src/heurams/interface/screens/dashboard.py
View File

@@ -146,26 +146,6 @@ class DashboardScreen(Screen):
# 跳转到准备屏幕
self.app.push_screen(PreparationScreen(selected_repo, self.repostat[self.title2dirname[selected_repotitle]]))
def on_button_pressed(self, event) -> None:
"""处理按钮点击事件"""
button_id = event.button.id
if button_id == "new_nucleon_button":
from .repocreator import RepoCreatorScreen
new_screen = RepoCreatorScreen()
self.app.push_screen(new_screen)
elif button_id == "precache_all_button":
from .precache import PrecachingScreen
precache_screen = PrecachingScreen()
self.app.push_screen(precache_screen)
elif button_id == "about_button":
about_screen = AboutScreen()
self.app.push_screen(about_screen)
def action_quit_app(self) -> None:
"""退出应用程序"""
self.app.exit()

2
src/heurams/interface/screens/memoqueue.py
View File

@@ -47,8 +47,6 @@ class MemScreen(Screen):
) -> None:
super().__init__(name, id, classes)
self.atoms = atoms
for i in self.atoms:
i.do_eval()
self.phaser = Phaser(atoms)
# logger.debug(self.phaser.state)
self.procession: Procession = self.phaser.current_procession() # type: ignore

48
src/heurams/interface/screens/precache.py
View File

@@ -38,12 +38,6 @@ class PrecachingScreen(Screen):
self.precache_worker = None
self.cancel_flag = 0
self.desc = desc
for i in nucleons:
i: pt.Nucleon
atom = pt.Atom()
atom.link("nucleon", i)
atom.do_eval()
# print("完成 EVAL")
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
@@ -94,7 +88,7 @@ class PrecachingScreen(Screen):
"""预缓存单段文本的音频"""
from heurams.context import config_var, rootdir, workdir
cache_dir = pathlib.Path(config_var.get()["paths"]["cache_dir"])
cache_dir = pathlib.Path(config_var.get()["paths"]["data"]) / 'cache'
cache_dir.mkdir(parents=True, exist_ok=True)
cache_file = cache_dir / f"{hasher.get_md5(text)}.wav"
if not cache_file.exists():
@@ -110,10 +104,8 @@ class PrecachingScreen(Screen):
def precache_by_nucleon(self, nucleon: pt.Nucleon):
"""依据 Nucleon 缓存"""
# print(nucleon.metadata['formation']['tts_text'])
ret = self.precache_by_text(nucleon.metadata["formation"]["tts_text"])
ret = self.precache_by_text(nucleon["tts_text"])
return ret
# print(f"TTS 缓存: {nucleon.metadata['formation']['tts_text']}")
def precache_by_list(self, nucleons: list):
"""依据 Nucleons 列表缓存"""
@@ -122,7 +114,7 @@ class PrecachingScreen(Screen):
worker = get_current_worker()
if worker and worker.is_cancelled: # 函数在worker中执行且已被取消
return False
text = nucleon.metadata["formation"]["tts_text"]
text = nucleon["tts_text"]
# self.current_item = text[:30] + "..." if len(text) > 50 else text
# print(text)
self.processed += 1
@@ -152,38 +144,26 @@ class PrecachingScreen(Screen):
# print(f"返回 {ret}")
return ret
def precache_by_filepath(self, path: pathlib.Path):
"""预缓存单个文件的所有内容"""
lst = list()
for i in pt.load_nucleon(path):
lst.append(i[0])
return self.precache_by_list(lst)
def precache_all_files(self):
"""预缓存所有文件"""
from heurams.context import config_var, rootdir, workdir
from heurams.kernel.repolib import Repo
nucleon_path = pathlib.Path(config_var.get()["paths"]["nucleon_dir"])
nucleon_files = [
f for f in nucleon_path.iterdir() if f.suffix == ".toml"
] # TODO: 解耦合
repo_path = pathlib.Path(config_var.get()["paths"]["data"]) / 'repo'
repo_dirs = Repo.probe_vaild_repos_in_dir(repo_path)
repos = map(Repo.create_from_repodir, repo_dirs)
# 计算总项目数
self.total = 0
nu = list()
for file in nucleon_files:
nucleon_list = list()
for repo in repos:
try:
for i in pt.load_nucleon(file):
nu.append(i[0])
for i in repo.ident_index:
nucleon_list.append(pt.Nucleon.create_on_nucleonic_data(repo.nucleonic_data_lict.get_itemic_unit(i)))
except:
continue
self.total = len(nu)
for i in nu:
i: pt.Nucleon
atom = pt.Atom()
atom.link("nucleon", i)
atom.do_eval()
return self.precache_by_list(nu)
self.total = len(nucleon_list)
return self.precache_by_list(nucleon_list)
def on_button_pressed(self, event: Button.Pressed) -> None:
event.stop()
@@ -221,7 +201,7 @@ class PrecachingScreen(Screen):
from heurams.context import config_var, rootdir, workdir
shutil.rmtree(
f"{config_var.get()["paths"]["cache_dir"]}", ignore_errors=True
f"{config_var.get()["paths"]["data"]}/'cache'", ignore_errors=True
)
self.update_status("已清空", "音频缓存已清空", 0)
except Exception as e:

43
src/heurams/interface/screens/preparation.py
View File

@@ -84,9 +84,9 @@ class PreparationScreen(Screen):
from ..screens.precache import PrecachingScreen
lst = list()
for i in self.nucleons_with_orbital:
lst.append(i[0])
precache_screen = PrecachingScreen(lst)
for i in self.repo.ident_index:
lst.append(pt.Nucleon.create_on_nucleonic_data(self.repo.nucleonic_data_lict.get_itemic_unit(i)))
precache_screen = PrecachingScreen(nucleons=lst, desc=self.repo.manifest["title"])
self.app.push_screen(precache_screen)
def action_quit_app(self):
@@ -97,38 +97,27 @@ class PreparationScreen(Screen):
logger.debug("按下按钮")
if event.button.id == "start_memorizing_button":
atoms = list()
for nucleon, orbital in self.nucleons_with_orbital:
atom = pt.Atom(nucleon.ident)
atom.link("nucleon", nucleon)
try:
atom.link("electron", self.electrons[nucleon.ident])
except KeyError:
atom.link("electron", pt.Electron(nucleon.ident))
atom.link("orbital", orbital)
atom.link("nucleon_fmt", "toml")
atom.link("electron_fmt", "json")
atom.link("orbital_fmt", "toml")
atom.link("nucleon_path", self.nucleon_file)
atom.link("electron_path", self.electron_file)
atom.link("orbital_path", None)
atoms.append(atom)
for i in self.repo.ident_index:
n = pt.Nucleon.create_on_nucleonic_data(nucleonic_data=self.repo.nucleonic_data_lict.get_itemic_unit(i))
e = pt.Electron.create_on_electonic_data(electronic_data=self.repo.electronic_data_lict.get_itemic_unit(i))
a = pt.Atom(n, e, self.repo.orbitic_data)
atoms.append(a)
atoms_to_provide = list()
left_new = self.scheduled_num
for i in atoms:
i: pt.Atom
if i.registry["electron"].is_due():
atoms_to_provide.append(i)
if i.registry['electron'].is_activated():
if i.registry["electron"].is_due():
atoms_to_provide.append(i)
else:
if i.registry["electron"].is_activated():
pass
else:
left_new -= 1
if left_new >= 0:
atoms_to_provide.append(i)
logger.debug(f"ATP: {atoms_to_provide}")
left_new -= 1
if left_new >= 0:
atoms_to_provide.append(i)
from .memoqueue import MemScreen
memscreen = MemScreen(atoms_to_provide)
self.app.push_screen(memscreen)
elif event.button.id == "precache_button":
self.action_precache()

10
src/heurams/kernel/particles/atom.py
View File

@@ -1,9 +1,5 @@
import json
import pathlib
import typing
from typing import TypedDict
import toml
from heurams.services.logger import get_logger
@@ -22,6 +18,7 @@ class AtomRegister_runtime(TypedDict):
class AtomRegister(TypedDict):
nucleon: Nucleon
electron: Electron
orbital: dict
runtime: AtomRegister_runtime
@@ -99,3 +96,8 @@ class Atom:
if key == "ident":
raise AttributeError("应为只读")
self.registry[key] = value
def __repr__(self):
from pprint import pformat
s = pformat(self.registry, indent=4)
return s

11
src/heurams/kernel/particles/nucleon.py
View File

@@ -13,12 +13,15 @@ class Nucleon:
self.ident = ident
env = {"payload": payload}
self.evalizer = Evalizer(environment=env)
self.data = self.evalizer(deepcopy((payload | common)))
self.data: dict = self.evalizer(deepcopy((payload | common))) # type: ignore
def __getitem__(self, key):
if key == "ident":
return self.ident
return self.data[key]
if isinstance(key, str):
if key == "ident":
return self.ident
return self.data[key]
else:
raise AttributeError
def __setitem__(self, key, value):
raise AttributeError("应为只读")

1
src/heurams/kernel/reactor/README.md Normal file
View File

@@ -0,0 +1 @@
# Reactor - 记忆过程状态机模块

12
src/heurams/kernel/reactor/__init__.py
View File

@@ -0,0 +1,12 @@
from heurams.services.logger import get_logger
from .fission import Fission
from .phaser import Phaser
from .procession import Procession
from .states import PhaserState, ProcessionState
logger = get_logger(__name__)
__all__ = ["PhaserState", "ProcessionState", "Procession", "Fission", "Phaser"]
logger.debug("反应堆模块已加载")

49
src/heurams/kernel/reactor/fission.py Normal file
View File

@@ -0,0 +1,49 @@
import random
import heurams.kernel.evaluators as puz
import heurams.kernel.particles as pt
from heurams.services.logger import get_logger
from .states import PhaserState
class Fission:
"""裂变器: 单原子调度展开器"""
def __init__(self, atom: pt.Atom, phase_state=PhaserState.RECOGNITION):
self.logger = get_logger(__name__)
self.atom = atom
#NOTE: phase 为 PhaserState 枚举实例，需要获取其value
phase_value = phase_state.value if isinstance(phase_state, PhaserState) else phase_state
self.orbital_schedule = atom.registry["orbital"]["schedule"][phase_value] # type: ignore
self.orbital_puzzles = atom.registry["orbital"]["puzzles"]
self.puzzles = list()
for item, possibility in self.orbital_schedule: # type: ignore
self.logger.debug(f"开始处理 orbital 项: {item}")
if not isinstance(possibility, float):
possibility = float(possibility)
while possibility > 1:
self.puzzles.append(
{
"puzzle": puz.puzzles[self.orbital_puzzles[item]["__origin__"]],
"alia": item,
}
)
possibility -= 1
if random.random() <= possibility:
self.puzzles.append(
{
"puzzle": puz.puzzles[self.orbital_puzzles[item]["__origin__"]],
"alia": item,
}
)
self.logger.debug(f"orbital 项处理完成: {item}")
def generate(self):
yield from self.puzzles

133
src/heurams/kernel/reactor/phaser.py
View File

@@ -1,24 +1,117 @@
from enum import Enum
from typing import Any, Sequence, Type
from transitions import Machine, State, Event, EventData
import heurams.kernel.particles as pt
from heurams.services.logger import get_logger
from transitions import Machine
from .procession import Procession
from .states import PhaserState, ProcessionState
logger = get_logger(__name__)
class Phaser(Machine):
def __init__(self, schedule: list):
state_words = ["init"] + schedule.copy()
state_objects = list()
for name in state_words:
state_objects.append(State(
name=name,
on_enter=["on_enter"]
))
Machine.__init__(self, states=state_objects, initial="init", send_event=True)
self.add_ordered_transitions(loop=False)
"""全局调度阶段管理器"""
def on_enter(self, event_data: EventData):
print(event_data.transition.source, "->", event_data.transition.dest) # type: ignore
def __init__(self, atoms: list[pt.Atom]) -> None:
logger.debug("Phaser.__init__: 原子数量=%d", len(atoms))
new_atoms = list()
old_atoms = list()
for i in atoms:
if not i.registry["electron"].is_activated():
new_atoms.append(i)
else:
old_atoms.append(i)
logger.debug("新原子数量=%d, 旧原子数量=%d", len(new_atoms), len(old_atoms))
self.processions = list()
#TODO: 改进为基于配置文件的可变复习阶段管理
if len(old_atoms):
self.processions.append(
Procession(old_atoms, PhaserState.QUICK_REVIEW, "初始复习")
)
logger.debug("创建初始复习 Procession")
if __name__ == "__main__":
p = Phaser(["a", "b"])
p.next_state()
p.next_state()
print(p.state)
if len(new_atoms):
self.processions.append(
Procession(new_atoms, PhaserState.RECOGNITION, "新记忆")
)
logger.debug("创建新记忆 Procession")
self.processions.append(Procession(atoms, PhaserState.FINAL_REVIEW, "总体复习"))
logger.debug("创建总体复习 Procession")
logger.debug("Phaser 初始化完成, processions 数量=%d", len(self.processions))
# 设置transitions状态机
states = [
{'name': PhaserState.UNSURE.value, 'on_enter': 'on_unsure'},
{'name': PhaserState.QUICK_REVIEW.value, 'on_enter': 'on_quick_review'},
{'name': PhaserState.RECOGNITION.value, 'on_enter': 'on_recognition'},
{'name': PhaserState.FINAL_REVIEW.value, 'on_enter': 'on_final_review'},
{'name': PhaserState.FINISHED.value, 'on_enter': 'on_finished'}
]
transitions = [
{'trigger': 'to_unsure', 'source': '*', 'dest': PhaserState.UNSURE.value},
{'trigger': 'to_quick_review', 'source': '*', 'dest': PhaserState.QUICK_REVIEW.value},
{'trigger': 'to_recognition', 'source': '*', 'dest': PhaserState.RECOGNITION.value},
{'trigger': 'to_final_review', 'source': '*', 'dest': PhaserState.FINAL_REVIEW.value},
{'trigger': 'to_finished', 'source': '*', 'dest': PhaserState.FINISHED.value}
]
Machine.__init__(self, states=states, transitions=transitions,
initial=PhaserState.UNSURE.value)
self.to_unsure()
def on_unsure(self):
"""进入UNSURE状态时的回调"""
logger.debug("Phaser 进入 UNSURE 状态")
def on_quick_review(self):
"""进入QUICK_REVIEW状态时的回调"""
logger.debug("Phaser 进入 QUICK_REVIEW 状态")
def on_recognition(self):
"""进入RECOGNITION状态时的回调"""
logger.debug("Phaser 进入 RECOGNITION 状态")
def on_final_review(self):
"""进入FINAL_REVIEW状态时的回调"""
logger.debug("Phaser 进入 FINAL_REVIEW 状态")
def on_finished(self):
"""进入FINISHED状态时的回调"""
logger.debug("Phaser 进入 FINISHED 状态")
def current_procession(self):
logger.debug("Phaser.current_procession 被调用")
for i in self.processions:
i: Procession
if i.state != ProcessionState.FINISHED.value:
# 根据当前procession的phase更新Phaser状态
if i.phase == PhaserState.QUICK_REVIEW:
self.to_quick_review()
elif i.phase == PhaserState.RECOGNITION:
self.to_recognition()
elif i.phase == PhaserState.FINAL_REVIEW:
self.to_final_review()
logger.debug("找到未完成的 Procession: phase=%s", i.phase)
return i
# 所有Procession都已完成
self.to_finished()
logger.debug("所有 Procession 已完成, 状态设置为 FINISHED")
return None
@property
def state(self):
"""获取当前状态值"""
current_state = self.get_model_state(self)
# 将字符串状态转换为PhaserState枚举
for phase in PhaserState:
if phase.value == current_state:
return phase
return PhaserState.UNSURE

130
src/heurams/kernel/reactor/procession.py
View File

@@ -1,24 +1,112 @@
from enum import Enum
from typing import Any, Sequence, Type
from transitions import Machine, State, Event, EventData
import heurams.kernel.particles as pt
from heurams.services.logger import get_logger
from transitions import Machine
class Phaser(Machine):
def __init__(self, schedule: list):
state_words = ["init"] + schedule.copy()
state_objects = list()
for name in state_words:
state_objects.append(State(
name=name,
on_enter=["on_enter"]
))
Machine.__init__(self, states=state_objects, initial="init", send_event=True)
self.add_ordered_transitions(loop=False)
from .states import PhaserState, ProcessionState
def on_enter(self, event_data: EventData):
print(event_data.transition.source, "->", event_data.transition.dest) # type: ignore
logger = get_logger(__name__)
if __name__ == "__main__":
p = Phaser(["a", "b"])
p.next_state()
p.next_state()
print(p.state)
class Procession(Machine):
"""队列: 标识单次记忆流程"""
def __init__(self, atoms: list, phase_state: PhaserState, name: str = ""):
logger.debug(
"Procession.__init__: 原子数量=%d, phase=%s, name='%s'",
len(atoms),
phase_state.value,
name,
)
# 初始化原子队列
self.atoms = atoms
self.queue = atoms.copy()
self.current_atom = atoms[0] if atoms else None
self.cursor = 0
self.name = name
self.phase = phase_state
# 设置transitions状态机
states = [{'name': ProcessionState.RUNNING.value, 'on_enter': 'on_running'},
{'name': ProcessionState.FINISHED.value, 'on_enter': 'on_finished'}]
transitions = [
{'trigger': 'finish', 'source': ProcessionState.RUNNING.value, 'dest': ProcessionState.FINISHED.value},
{'trigger': 'restart', 'source': ProcessionState.FINISHED.value, 'dest': ProcessionState.RUNNING.value}
]
Machine.__init__(self, states=states, transitions=transitions,
initial=ProcessionState.RUNNING.value)
logger.debug("Procession 初始化完成, 队列长度=%d", len(self.queue))
def on_running(self):
"""进入RUNNING状态时的回调"""
logger.debug("Procession 进入 RUNNING 状态")
def on_finished(self):
"""进入FINISHED状态时的回调"""
logger.debug("Procession 进入 FINISHED 状态")
def forward(self, step=1):
logger.debug("Procession.forward: step=%d, 当前 cursor=%d", step, self.cursor)
self.cursor += step
if self.cursor >= len(self.queue):
if self.state != ProcessionState.FINISHED.value:
self.finish() # 触发状态转换
logger.debug("Procession 已完成")
else:
if self.state != ProcessionState.RUNNING.value:
self.restart() # 确保在RUNNING状态
self.current_atom = self.queue[self.cursor]
logger.debug("cursor 更新为: %d", self.cursor)
logger.debug("当前原子更新为: %s", self.current_atom.ident if self.current_atom else "None")
return 1 # 成功
return 0
def append(self, atom=None):
if atom is None:
atom = self.current_atom
logger.debug("Procession.append: atom=%s", atom.ident if atom else "None")
if not self.queue or self.queue[-1] != atom or len(self) <= 1:
self.queue.append(atom)
logger.debug("原子已追加到队列, 新队列长度=%d", len(self.queue))
else:
logger.debug("原子未追加(重复或队列长度<=1)")
def __len__(self):
if not self.queue:
return 0
length = len(self.queue) - self.cursor
logger.debug("Procession.__len__: 剩余长度=%d", length)
return length
def process(self):
logger.debug("Procession.process: cursor=%d", self.cursor)
return self.cursor
def total_length(self):
total = len(self.queue)
logger.debug("Procession.total_length: %d", total)
return total
def is_empty(self):
empty = len(self.queue) == 0
logger.debug("Procession.is_empty: %s", empty)
return empty
@property
def state(self):
"""获取当前状态值"""
return self.get_model_state(self)
@state.setter
def state(self, value):
"""设置状态值"""
if value == ProcessionState.RUNNING.value:
self.restart()
elif value == ProcessionState.FINISHED.value:
self.finish()

21
src/heurams/kernel/reactor/states.py Normal file
View File

@@ -0,0 +1,21 @@
from enum import Enum, auto
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class PhaserState(Enum):
UNSURE = "unsure"
QUICK_REVIEW = "quick_review"
RECOGNITION = "recognition"
FINAL_REVIEW = "final_review"
FINISHED = "finished"
class ProcessionState(Enum):
RUNNING = "running"
FINISHED = "finished"
logger.debug("状态枚举定义已加载")

3
src/heurams/services/version.py
View File

@@ -5,6 +5,7 @@ logger = get_logger(__name__)
ver = "0.5.0"
stage = "prototype"
codename = "fulcrom" # 支点
codename = "fulcrom"
codename_cn = "支点"
logger.info("HeurAMS 版本: %s (%s), 阶段: %s", ver, codename, stage)

4
src/heurams/utils/lict.py
View File

@@ -52,8 +52,8 @@ class Lict(UserList): # TODO: 优化同步(惰性同步), 当前性能为 O(n)
else:
return super().__getitem__(i)
def get_itemic_unit(self, i):
return (i, self.dicted_data[i])
def get_itemic_unit(self, ident):
return (ident, self.dicted_data[ident])
def __setitem__(self, i, item):
if isinstance(i, str):