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

fix: 改进代码

Browse Source
This commit is contained in:
Wang Zhiyu
2026-01-04 04:46:19 +08:00
parent c585c79e73
commit 65486794b7
34 changed files with 87 additions and 1570 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,9 +17,9 @@ def environment_check():
from pathlib import Path
logger.debug("检查环境路径")
subdir = ['cache/voice', 'repo', 'global', 'config']
subdir = ["cache/voice", "repo", "global", "config"]
for i in subdir:
i = Path(config_var.get()['paths']['data']) / i
i = Path(config_var.get()["paths"]["data"]) / i
if not i.exists():
logger.info("创建目录: %s", i)
print(f"创建 {i}")

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

@@ -14,6 +14,8 @@ from heurams.kernel.particles import *
from heurams.kernel.repolib import *
from heurams.services.logger import get_logger
import heurams.kernel.particles as pt
from pathlib import Path
from .about import AboutScreen
from .preparation import PreparationScreen
@@ -42,7 +44,9 @@ class DashboardScreen(Screen):
yield Header(show_clock=True)
yield ScrollableContainer(
Label('欢迎使用 "潜进" 启发式辅助记忆调度器', classes="title-label"),
Label(f"当前 UNIX 日时间戳: {timer.get_daystamp()} (UTC+{config_var.get()["timezone_offset"] / 3600})"),
Label(
f"当前 UNIX 日时间戳: {timer.get_daystamp()} (UTC+{config_var.get()["timezone_offset"] / 3600})"
),
Label(f"全局算法设置: {config_var.get()['algorithm']['default']}"),
Label("选择待学习或待修改的项目:", classes="title-label"),
ListView(id="repo-list", classes="repo-list-view"),

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

@@ -41,9 +41,9 @@ class MemScreen(Screen):
def __init__(
self,
phaser: Phaser,
name = None,
id = None,
classes = None,
name=None,
id=None,
classes=None,
) -> None:
super().__init__(name, id, classes)
self.phaser = phaser
@@ -59,7 +59,7 @@ class MemScreen(Screen):
def update_state(self):
"""更新状态机"""
self.procession: Procession = self.phaser.current_procession() # type: ignore
self.atom: pt.Atom = self.procession.current_atom # type: ignore
self.atom: pt.Atom = self.procession.current_atom # type: ignore
def on_mount(self):
self.mount_puzzle()
@@ -69,14 +69,12 @@ class MemScreen(Screen):
try:
self.fission = self.procession.get_fission()
puzzle = self.fission.get_current_puzzle()
# logger.debug(puzzle_debug)
return shim.puzzle2widget[puzzle["puzzle"]]( # type: ignore
atom=self.atom, alia=puzzle["alia"] # type: ignore
return shim.puzzle2widget[puzzle["puzzle"]]( # type: ignore
atom=self.atom, alia=puzzle["alia"] # type: ignore
)
except (KeyError, StopIteration, AttributeError) as e:
logger.debug(f"调度展开出错: {e}")
return Static(f"无法生成谜题 {e}")
# logger.debug(shim.puzzle2widget[puzzle_debug["puzzle"]])
def _get_progress_text(self):
s = f"阶段: {self.procession.phase.name}\n"
@@ -117,32 +115,28 @@ class MemScreen(Screen):
from heurams.services.audio_service import play_by_path
from heurams.services.hasher import get_md5
path = Path(config_var.get()["paths"]['data']) / 'cache' / 'voice'
path = (
path
/ f"{get_md5(self.atom.registry['nucleon']["tts_text"])}.wav"
)
path = Path(config_var.get()["paths"]["data"]) / "cache" / "voice"
path = path / f"{get_md5(self.atom.registry['nucleon']["tts_text"])}.wav"
if path.exists():
play_by_path(path)
else:
from heurams.services.tts_service import convertor
convertor(
self.atom.registry["nucleon"]["tts_text"], path
)
convertor(self.atom.registry["nucleon"]["tts_text"], path)
play_by_path(path)
def watch_rating(self, old_rating, new_rating) -> None:
self.update_state() # 刷新状态
if self.procession == None: # 已经完成记忆
self.update_state() # 刷新状态
if self.procession == None: # 已经完成记忆
return
if new_rating == -1: # 安全值
if new_rating == -1: # 安全值
return
forwards = 1 if new_rating >= 4 else 0 # 准许前进
forwards = 1 if new_rating >= 4 else 0 # 准许前进
self.rating = -1
logger.debug(f"试图前进: {"允许" if forwards else "禁止"}")
if forwards:
ret = self.procession.forward(1)
if ret == 0: # 若结束了此次队列
if ret == 0: # 若结束了此次队列
self.update_state()
if self.procession.phase == PhaserState.FINISHED: # 若所有队列都结束了
logger.debug(f"记忆进程结束")

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

@@ -12,7 +12,8 @@ import heurams.kernel.particles as pt
import heurams.services.hasher as hasher
from heurams.context import *
cache_dir = pathlib.Path(config_var.get()["paths"]["data"]) / "cache" / 'voice'
cache_dir = pathlib.Path(config_var.get()["paths"]["data"]) / "cache" / "voice"
class PrecachingScreen(Screen):
"""预缓存音频文件屏幕
@@ -204,9 +205,7 @@ class PrecachingScreen(Screen):
from heurams.context import config_var, rootdir, workdir
shutil.rmtree(
cache_dir, ignore_errors=True
)
shutil.rmtree(cache_dir, ignore_errors=True)
self.update_status("已清空", "音频缓存已清空", 0)
except Exception as e:
self.update_status("错误", f"清空缓存失败: {e}")

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

@@ -128,6 +128,7 @@ class PreparationScreen(Screen):
atoms_to_provide.append(i)
from .memoqueue import MemScreen
import heurams.kernel.reactor as rt
pheser = rt.Phaser(atoms_to_provide)
memscreen = MemScreen(pheser)
self.app.push_screen(memscreen)

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

@@ -1,4 +1,5 @@
"""Kernel 操作辅助函数库"""
import heurams.interface.widgets as pzw
import heurams.kernel.evaluators as pz

2
src/heurams/interface/widgets/mcq_puzzle.py
View File

@@ -54,7 +54,7 @@ class MCQPuzzle(BasePuzzleWidget):
self._load()
def _load(self):
cfg = self.atom.registry["orbital"]["puzzles"][self.alia]
cfg = self.atom.registry["nucleon"]["puzzles"][self.alia]
self.puzzle = pz.MCQPuzzle(
cfg["mapping"], cfg["jammer"], int(cfg["max_riddles_num"]), cfg["prefix"]
)

6
src/heurams/kernel/algorithms/__init__.py
View File

@@ -1,11 +1,7 @@
from heurams.services.logger import get_logger
from .base import BaseAlgorithm
from .sm2 import SM2Algorithm
from .sm15m import SM15MAlgorithm
logger = get_logger(__name__)
__all__ = [
"SM2Algorithm",
"BaseAlgorithm",
@@ -17,5 +13,3 @@ algorithms = {
"SM-15M": SM15MAlgorithm,
"Base": BaseAlgorithm,
}
logger.debug("算法模块初始化完成, 注册的算法: %s", list(algorithms.keys()))

37
src/heurams/kernel/evaluators/__init__.py
View File

@@ -6,8 +6,6 @@ Evaluator 模块 - 生成评估模块
from heurams.services.logger import get_logger
logger = get_logger(__name__)
from .base import BaseEvaluator
from .cloze import ClozePuzzle
from .mcq import MCQPuzzle
@@ -26,38 +24,3 @@ puzzles = {
"recognition": RecognitionPuzzle,
"base": BaseEvaluator,
}
@staticmethod
def create_by_dict(config_dict: dict) -> BaseEvaluator:
"""
根据配置字典创建谜题
Args:
config_dict: 配置字典, 包含谜题类型和参数
Returns:
BasePuzzle: 谜题实例
Raises:
ValueError: 当配置无效时抛出
"""
logger.debug(
"puzzles.create_by_dict: config_dict keys=%s", list(config_dict.keys())
)
puzzle_type = config_dict.get("type")
if puzzle_type == "cloze":
return puzzles["cloze"](
text=config_dict["text"],
min_denominator=config_dict.get("min_denominator", 7),
)
elif puzzle_type == "mcq":
return puzzles["mcq"](
mapping=config_dict["mapping"],
jammer=config_dict.get("jammer", []),
max_riddles_num=config_dict.get("max_riddles_num", 2),
prefix=config_dict.get("prefix", ""),
)
else:
raise ValueError(f"未知的谜题类型: {puzzle_type}")

19
src/heurams/kernel/particles/__init__.py
View File

@@ -1,4 +1,21 @@
from .atom import Atom
from .electron import Electron
from .nucleon import Nucleon
#from .orbital import Orbital
from .placeholders import (
AtomPlaceholder,
NucleonPlaceholder,
ElectronPlaceholder,
orbital_placeholder,
)
# from .orbital import Orbital
__all__ = [
"Atom",
"Electron",
"Nucleon",
"AtomPlaceholder",
"NucleonPlaceholder",
"ElectronPlaceholder",
"orbital_placeholder",
]

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

@@ -13,9 +13,11 @@ class Nucleon:
def __init__(self, ident, payload, common):
self.ident = ident
env = {"payload": payload,
"default": config_var.get()['puzzles'],
"nucleon": (payload | common)}
env = {
"payload": payload,
"default": config_var.get()["puzzles"],
"nucleon": (payload | common),
}
self.evalizer = Evalizer(environment=env)
self.data: dict = self.evalizer(deepcopy((payload | common))) # type: ignore

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

@@ -5,8 +5,4 @@ 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("反应堆模块已加载")

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

@@ -20,7 +20,7 @@ class Fission:
phase_state.value if isinstance(phase_state, PhaserState) else phase_state
)
self.orbital_schedule = atom.registry['orbital']["phases"][phase_value] # type: ignore
self.orbital_schedule = atom.registry["orbital"]["phases"][phase_value] # type: ignore
self.orbital_puzzles = atom.registry["nucleon"]["puzzles"]
self.puzzles = list()
@@ -34,7 +34,6 @@ class Fission:
{
"puzzle": puz.puzzles[self.orbital_puzzles[item]["__origin__"]],
"alia": item,
"finished": 0,
}
)
possibility -= 1
@@ -44,7 +43,6 @@ class Fission:
{
"puzzle": puz.puzzles[self.orbital_puzzles[item]["__origin__"]],
"alia": item,
"finished": 0,
}
)
@@ -53,7 +51,7 @@ class Fission:
def get_puzzles(self):
return self.puzzles
def get_current_puzzle(self, forward = 0):
def get_current_puzzle(self, forward=0):
if forward:
if len(self.puzzles) <= self.cursor + 1:
return 0
@@ -61,10 +59,9 @@ class Fission:
return self.puzzles[self.cursor]
else:
return self.puzzles[self.cursor]
def check_passed(self):
for i in self.puzzles:
if i["finished"] == 0:
return 0
return 1
return 1

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

@@ -130,12 +130,13 @@ class Phaser(Machine):
def __repr__(self):
from heurams.services.textproc import truncate
from tabulate import tabulate as tabu
lst = [
{
"Type": "Phaser",
"State": self.state,
"Processions": list(map(lambda f: (f.name_), self.processions)),
"Current Procession": "None" if not self.current_procession() else self.current_procession().name_, # type: ignore
"Current Procession": "None" if not self.current_procession() else self.current_procession().name_, # type: ignore
},
]
return str(tabu(lst, headers="keys")) + '\n'
return str(tabu(lst, headers="keys")) + "\n"

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

@@ -63,8 +63,7 @@ class Procession(Machine):
logger.debug("Procession 进入 FINISHED 状态")
def forward(self, step=1):
"""将记忆原子指针向前移动并依情况更新原子(返回 1)或完成队列(返回 0)
"""
"""将记忆原子指针向前移动并依情况更新原子(返回 1)或完成队列(返回 0)"""
logger.debug("Procession.forward: step=%d, 当前 cursor=%d", step, self.cursor)
self.cursor += step
if self.cursor >= len(self.queue):
@@ -84,8 +83,7 @@ class Procession(Machine):
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")
@@ -118,10 +116,11 @@ class Procession(Machine):
return empty
def get_fission(self):
return Fission(atom=self.current_atom, phase_state=self.phase) # type: ignore
return Fission(atom=self.current_atom, phase_state=self.phase) # type: ignore
def __repr__(self):
from heurams.services.textproc import truncate
dic = [
{
"Type": "Procession",
@@ -129,7 +128,7 @@ class Procession(Machine):
"State": self.state,
"Progress": f"{self.cursor + 1} / {len(self.queue)}",
"Queue": list(map(lambda f: truncate(f.ident), self.queue)),
"Current Atom": self.current_atom.ident, # type: ignore
"Current Atom": self.current_atom.ident, # type: ignore
}
]
return str(tabu(dic, headers="keys")) + '\n'
return str(tabu(dic, headers="keys")) + "\n"

4
src/heurams/kernel/repolib/__init__.py
View File

@@ -1 +1,3 @@
from .repo import *
from .repo import Repo, RepoManifest
__all__ = ["Repo", "RepoManifest"]

5
src/heurams/kernel/repolib/aux.py
View File

@@ -1,5 +0,0 @@
from ...utils.lict import Lict
def merge(x: Lict, y: Lict):
return Lict(list(x.values()) + list(y.values()))

2
src/heurams/services/textproc.py
View File

@@ -1,4 +1,4 @@
def truncate(text):
if len(text) <= 3:
return text
return text[:3] + ">"
return text[:3] + ">"

5
src/heurams/utils/__init__.py
View File

@@ -0,0 +1,5 @@
from .evalizor import Evalizer
from .lict import Lict
from .refvar import RefVar
__all__ = ["Evalizer", "Lict", "RefVar"]