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

feat: 改进状态机

Browse Source
This commit is contained in:
Wang Zhiyu
2026-01-03 13:08:08 +08:00
parent aacf4fdbdf
commit 94aaef386b
24 changed files with 275 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/heurams/context.py
View File

@@ -26,23 +26,30 @@ if pathlib.Path(workdir / "data" / "config" / "config_dev.toml").exists():
print("使用开发设置")
logger.debug("使用开发设置")
config_var: ContextVar[ConfigFile] = ContextVar(
"config_var", default=ConfigFile(workdir / "data" / "config" / "config_dev.toml")
"config_var",
default=ConfigFile(workdir / "data" / "config" / "config_dev.toml"),
)
else:
try:
config_var: ContextVar[ConfigFile] = ContextVar(
"config_var", default=ConfigFile(workdir / "data" / "config" / "config.toml")
"config_var",
default=ConfigFile(workdir / "data" / "config" / "config.toml"),
) # 配置文件
except Exception as e:
input("按下回车以创建新的配置文件, 或按下 Ctrl + C 以终止程序 ")
(workdir / "data" / 'config').mkdir(parents=True, exist_ok=True)
(workdir / "data" / 'config' / 'config').unlink(missing_ok=True)
shutil.copy((rootdir / 'default' / 'config' / 'config.toml'), workdir / "data" / "config" / "config.toml")
(workdir / "data" / "config").mkdir(parents=True, exist_ok=True)
(workdir / "data" / "config" / "config").unlink(missing_ok=True)
shutil.copy(
(rootdir / "default" / "config" / "config.toml"),
workdir / "data" / "config" / "config.toml",
)
finally:
config_var: ContextVar[ConfigFile] = ContextVar(
"config_var", default=ConfigFile(workdir / "data" / "config" / "config.toml")
"config_var",
default=ConfigFile(workdir / "data" / "config" / "config.toml"),
) # 配置文件
class ConfigContext:
"""
功能完备的上下文管理器

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

@@ -12,6 +12,7 @@ from .screens.synctool import SyncScreen
logger = get_logger(__name__)
def environment_check():
from pathlib import Path

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

@@ -5,8 +5,7 @@ import pathlib
from textual.app import ComposeResult
from textual.containers import ScrollableContainer
from textual.screen import Screen
from textual.widgets import (Button, Footer, Header, Label, ListItem, ListView,
Static)
from textual.widgets import Button, Footer, Header, Label, ListItem, ListView, Static
import heurams.services.timer as timer
import heurams.services.version as version
@@ -48,29 +47,33 @@ class DashboardScreen(Screen):
Label(f"使用算法: {config_var.get()['algorithm']['default']}"),
Label("选择待学习或待修改的仓库:", classes="title-label"),
ListView(id="repo-list", classes="repo-list-view"),
Label(
f'"潜进" 启发式辅助记忆调度器 | 版本 {version.ver} '
),
Label(f'"潜进" 启发式辅助记忆调度器 | 版本 {version.ver} '),
)
yield Footer()
def _load_data(self):
self.repo_dirs = Repo.probe_vaild_repos_in_dir(Path(config_var.get()['paths']['data']) / 'repo')
self.repo_dirs = Repo.probe_vaild_repos_in_dir(
Path(config_var.get()["paths"]["data"]) / "repo"
)
for repo_dir in self.repo_dirs:
repo = Repo.create_from_repodir(repo_dir)
self._analyse_repo(repo)
def _analyse_repo(self, repo: Repo):
dirname = repo.source.name # type: ignore
dirname = repo.source.name # type: ignore
title = repo.manifest["title"]
is_due = 0
unit_sum = len(repo)
activated_sum = 0
nextdate = 0x3f3f3f3f
is_unfinished = (unit_sum > activated_sum)
nextdate = 0x3F3F3F3F
is_unfinished = unit_sum > activated_sum
for i in repo.ident_index:
nucleon = pt.Nucleon.create_on_nucleonic_data(nucleonic_data=repo.nucleonic_data_lict.get_itemic_unit(i))
electron = pt.Electron.create_on_electonic_data(electronic_data=repo.electronic_data_lict.get_itemic_unit(i))
nucleon = pt.Nucleon.create_on_nucleonic_data(
nucleonic_data=repo.nucleonic_data_lict.get_itemic_unit(i)
)
electron = pt.Electron.create_on_electonic_data(
electronic_data=repo.electronic_data_lict.get_itemic_unit(i)
)
if electron.is_activated():
activated_sum += 1
if electron.is_due():
@@ -102,10 +105,10 @@ class DashboardScreen(Screen):
# 按下次复习时间排序
repodirs = sorted(
self.repo_dirs,
key=lambda f: self.repostat[f.name]['nextdate'],
key=lambda f: self.repostat[f.name]["nextdate"],
reverse=True,
)
repotitles = map(lambda f: self.repostat[f.name]['title'], repodirs)
repotitles = map(lambda f: self.repostat[f.name]["title"], repodirs)
# 填充列表
if not repodirs:
repo_list_widget.append(
@@ -120,11 +123,11 @@ class DashboardScreen(Screen):
return
for repotitle in repotitles:
prompt = self.repostat[self.title2dirname[repotitle]]['prompt']
prompt = self.repostat[self.title2dirname[repotitle]]["prompt"]
list_item = ListItem(Label(prompt))
repo_list_widget.append(list_item)
#if not self.stay_enabled[repodir]:
# if not self.stay_enabled[repodir]:
# list_item.disabled = True
def on_list_view_selected(self, event) -> None:
@@ -141,10 +144,13 @@ class DashboardScreen(Screen):
# 提取文件名
selected_repotitle = label_text.partition("\0")[0].replace("*", "")
selected_repo = self.title2repo[label_text.partition("\0")[0].replace("*", "")]
# 跳转到准备屏幕
self.app.push_screen(PreparationScreen(selected_repo, self.repostat[self.title2dirname[selected_repotitle]]))
self.app.push_screen(
PreparationScreen(
selected_repo, self.repostat[self.title2dirname[selected_repotitle]]
)
)
def action_quit_app(self) -> None:
"""退出应用程序"""

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

@@ -88,7 +88,7 @@ class PrecachingScreen(Screen):
"""预缓存单段文本的音频"""
from heurams.context import config_var, rootdir, workdir
cache_dir = pathlib.Path(config_var.get()["paths"]["data"]) / 'cache'
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():
@@ -149,7 +149,7 @@ class PrecachingScreen(Screen):
from heurams.context import config_var, rootdir, workdir
from heurams.kernel.repolib import Repo
repo_path = pathlib.Path(config_var.get()["paths"]["data"]) / 'repo'
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)
@@ -159,7 +159,11 @@ class PrecachingScreen(Screen):
for repo in repos:
try:
for i in repo.ident_index:
nucleon_list.append(pt.Nucleon.create_on_nucleonic_data(repo.nucleonic_data_lict.get_itemic_unit(i)))
nucleon_list.append(
pt.Nucleon.create_on_nucleonic_data(
repo.nucleonic_data_lict.get_itemic_unit(i)
)
)
except:
continue
self.total = len(nucleon_list)

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

@@ -73,7 +73,9 @@ class PreparationScreen(Screen):
def _get_full_content(self):
content = ""
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))
n = pt.Nucleon.create_on_nucleonic_data(
nucleonic_data=self.repo.nucleonic_data_lict.get_itemic_unit(i)
)
content += f"- {n['content']} \n"
return content
@@ -85,8 +87,14 @@ class PreparationScreen(Screen):
lst = list()
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"])
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):
@@ -98,8 +106,12 @@ class PreparationScreen(Screen):
if event.button.id == "start_memorizing_button":
atoms = list()
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))
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)
@@ -107,7 +119,7 @@ class PreparationScreen(Screen):
left_new = self.scheduled_num
for i in atoms:
i: pt.Atom
if i.registry['electron'].is_activated():
if i.registry["electron"].is_activated():
if i.registry["electron"].is_due():
atoms_to_provide.append(i)
else:

3
src/heurams/interface/screens/repocreator.py
View File

@@ -6,8 +6,7 @@ import toml
from textual.app import ComposeResult
from textual.containers import ScrollableContainer
from textual.screen import Screen
from textual.widgets import (Button, Footer, Header, Input, Label, Markdown,
Select)
from textual.widgets import Button, Footer, Header, Input, Label, Markdown, Select
from heurams.context import config_var
from heurams.services.version import ver

17
src/heurams/kernel/algorithms/sm15m.py
View File

@@ -14,13 +14,22 @@ import pathlib
from typing import TypedDict
from heurams.context import config_var
from heurams.kernel.algorithms.sm15m_calc import (MAX_AF, MIN_AF, NOTCH_AF,
RANGE_AF, RANGE_REPETITION,
SM, THRESHOLD_RECALL, Item)
from heurams.kernel.algorithms.sm15m_calc import (
MAX_AF,
MIN_AF,
NOTCH_AF,
RANGE_AF,
RANGE_REPETITION,
SM,
THRESHOLD_RECALL,
Item,
)
# 全局状态文件路径
_GLOBAL_STATE_FILE = os.path.expanduser(
pathlib.Path(config_var.get()["paths"]["data"]) / 'global' / "sm15m_global_state.json"
pathlib.Path(config_var.get()["paths"]["data"])
/ "global"
/ "sm15m_global_state.json"
)

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

@@ -99,5 +99,6 @@ class Atom:
def __repr__(self):
from pprint import pformat
s = pformat(self.registry, indent=4)
return s

1
src/heurams/kernel/particles/electron.py
View File

@@ -31,6 +31,7 @@ class Electron:
def __repr__(self):
from pprint import pformat
s = pformat(self.algodata, indent=4)
return s

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

@@ -13,7 +13,7 @@ class Nucleon:
self.ident = ident
env = {"payload": payload}
self.evalizer = Evalizer(environment=env)
self.data: dict = self.evalizer(deepcopy((payload | common))) # type: ignore
self.data: dict = self.evalizer(deepcopy((payload | common))) # type: ignore
def __getitem__(self, key):
if isinstance(key, str):
@@ -45,6 +45,7 @@ class Nucleon:
def __repr__(self):
from pprint import pformat
s = pformat(self.data, indent=4)
return s

27
src/heurams/kernel/particles/orbital.py
View File

@@ -1,16 +1,17 @@
from heurams.utils.evalizor import Evalizer
from heurams.utils.lict import Lict
"""轨道对象"""
# 似乎没有实现这个类的必要...
# 那不妨在这儿写点文档
class Orbital:
@classmethod
def create_orbital(cls, schedule: list, phase_def: dict):
phase_def = Lict(initdict=phase_def) # type: ignore
orbital = Lict()
for i in schedule:
orbital[i] = Lict(phase_def[i])
return orbital
"""
orbital, 即轨道, 是定义队列式复习阶段流程的数据结构, 其实就是个字典, 至于为何不用typeddict, 因为懒.
@classmethod
def create_orbital_on_orbitic_data(cls, orbitic_data):
return cls.create_orbital(orbitic_data["schedule"], orbitic_data["phases"])
orbital_example = {
"schedule": [列表 存储阶段(phases)名称]
"phases":{
阶段名称 = [["谜题(puzzle 现称 evaluator 评估器)名称", "概率系数 可大于1(整数部分为重复次数) 注意使用字符串包裹(toml 规范)"], ...],
...
}
}
至于谜题定义 放在 nucleon['puzzles'], 这样设计是为了兼容多种不同谜题实现的记忆单元, 尽管如此, 你也可见其谜题调度方式必须是相同的.
"""

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

@@ -9,4 +9,4 @@ logger = get_logger(__name__)
__all__ = ["PhaserState", "ProcessionState", "Procession", "Fission", "Phaser"]
logger.debug("反应堆模块已加载")
logger.debug("反应堆模块已加载")

30
src/heurams/kernel/reactor/fission.py
View File

@@ -8,24 +8,26 @@ 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"]
# NOTE: phase 为 PhaserState 枚举实例，需要获取其value
phase_value = (
phase_state.value if isinstance(phase_state, PhaserState) else phase_state
)
self.orbital_schedule = atom.registry["phases"][phase_value] # type: ignore
self.orbital_puzzles = atom.registry["nucleon"]["puzzles"]
self.puzzles = list()
for item, possibility in self.orbital_schedule: # type: ignore
self.logger.debug(f"开始处理 orbital 项: {item}")
self.logger.debug(f"开始处理: {item}")
if not isinstance(possibility, float):
possibility = float(possibility)
while possibility > 1:
self.puzzles.append(
{
@@ -34,7 +36,7 @@ class Fission:
}
)
possibility -= 1
if random.random() <= possibility:
self.puzzles.append(
{
@@ -42,8 +44,8 @@ class Fission:
"alia": item,
}
)
self.logger.debug(f"orbital 项处理完成: {item}")
def generate(self):
yield from self.puzzles
def get_puzzles_list(self):
yield from self.puzzles

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

@@ -13,20 +13,20 @@ class Phaser(Machine):
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: 改进为基于配置文件的可变复习阶段管理
# TODO: 改进为基于配置文件的可变复习阶段管理
if len(old_atoms):
self.processions.append(
Procession(old_atoms, PhaserState.QUICK_REVIEW, "初始复习")
@@ -42,27 +42,47 @@ class Phaser(Machine):
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'}
{"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}
{"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)
Machine.__init__(
self,
states=states,
transitions=transitions,
initial=PhaserState.UNSURE.value,
)
self.to_unsure()
def on_unsure(self):
@@ -97,15 +117,15 @@ class Phaser(Machine):
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):
"""获取当前状态值"""
@@ -114,4 +134,4 @@ class Phaser(Machine):
for phase in PhaserState:
if phase.value == current_state:
return phase
return PhaserState.UNSURE
return PhaserState.UNSURE

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

@@ -17,27 +17,39 @@ class Procession(Machine):
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}
states = [
{"name": ProcessionState.RUNNING.value, "on_enter": "on_running"},
{"name": ProcessionState.FINISHED.value, "on_enter": "on_finished"},
]
Machine.__init__(self, states=states, transitions=transitions,
initial=ProcessionState.RUNNING.value)
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):
@@ -51,7 +63,7 @@ class Procession(Machine):
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() # 触发状态转换
@@ -61,16 +73,19 @@ class Procession(Machine):
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")
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))
@@ -97,16 +112,16 @@ class Procession(Machine):
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()
self.finish()

2
src/heurams/kernel/reactor/states.py
View File

@@ -18,4 +18,4 @@ class ProcessionState(Enum):
FINISHED = "finished"
logger.debug("状态枚举定义已加载")
logger.debug("状态枚举定义已加载")

11
src/heurams/kernel/repolib/repo.py
View File

@@ -9,11 +9,13 @@ import heurams.kernel.particles as pt
from ...utils.lict import Lict
class RepoManifest(TypedDict):
title: str
author: str
desc: str
class Repo:
file_mapping = {
"schedule": "schedule.toml",
@@ -43,7 +45,7 @@ class Repo:
source=None,
) -> None:
self.schedule: dict = schedule
self.manifest: RepoManifest = manifest # type: ignore
self.manifest: RepoManifest = manifest # type: ignore
self.typedef: dict = typedef
self.payload: Lict = payload
self.algodata: Lict = algodata
@@ -61,9 +63,7 @@ class Repo:
def generate_particles_data(self):
self.nucleonic_data_lict = Lict(
initlist=list(map(
self._nucleonic_proc,
self.payload))
initlist=list(map(self._nucleonic_proc, self.payload))
)
self.orbitic_data = self.schedule
self.ident_index = self.nucleonic_data_lict.keys()
@@ -88,6 +88,7 @@ class Repo:
def __repr__(self):
from pprint import pformat
s = pformat(self.database, indent=4)
return s
@@ -172,4 +173,4 @@ class Repo:
if i.is_dir():
if cls.check_repodir(i):
lst.append(i)
return lst
return lst

2
src/heurams/utils/evalizor.py
View File

@@ -1,4 +1,4 @@
class Evalizer():
class Evalizer:
"""几乎无副作用的模板系统
接受环境信息并创建一个模板解析工具, 工具传入参数支持list, dict及其嵌套