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

fix: 修正逻辑问题

Browse Source
This commit is contained in:
Wang Zhiyu
2026-01-06 18:28:57 +08:00
parent e1c935f348
commit 22b41789eb
8 changed files with 109 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
data/repo/test/schedule.toml
View File

@@ -1,8 +1,10 @@
schedule = ["quick_review", "recognition", "final_review"]
[phases]
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"]]
final_review = [["FillBlank", "1.0"], ["SelectMeaning", "1.0"], ["Recognition", "1.0"]]
[annotation]
"quick_review" = "复习旧知"
"recognition" = "新知识"

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

@@ -27,7 +27,7 @@ logger = get_logger(__name__)
class MemScreen(Screen):
BINDINGS = [
("q", "pop_screen", "返回"),
("q", "go_back", "返回"),
("p", "prev", "查看上一个"),
("d", "toggle_dark", ""),
("v", "play_voice", "朗读"),
@@ -35,7 +35,8 @@ class MemScreen(Screen):
]
if config_var.get()["quick_pass"]:
BINDINGS.append(("k", "quick_pass", "跳过"))
BINDINGS.append(("k", "quick_pass", "正确应答"))
BINDINGS.append(("f", "quick_fail", "错误应答"))
rating = reactive(-1)
def __init__(
@@ -48,6 +49,7 @@ class MemScreen(Screen):
super().__init__(name, id, classes)
self.phaser = phaser
self.update_state()
self.fission: Fission
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
@@ -62,23 +64,43 @@ class MemScreen(Screen):
self.atom: pt.Atom = self.procession.current_atom # type: ignore
def on_mount(self):
self.fission = self.procession.get_fission()
self.mount_puzzle()
self.update_display()
def puzzle_widget(self):
try:
self.fission = self.procession.get_fission()
puzzle = self.fission.get_current_puzzle()
puzzle = self.fission.get_current_puzzle_inf()
return shim.puzzle2widget[puzzle["puzzle"]]( # type: ignore
atom=self.atom, alia=puzzle["alia"] # type: ignore
)
except (KeyError, StopIteration, AttributeError) as e:
except Exception as e:
logger.debug(f"调度展开出错: {e}")
return Static(f"无法生成谜题 {e}")
def _get_progress_text(self):
s = f"阶段: {self.procession.phase.name}\n"
s += f"当前进度: {self.procession.process() + 1}/{self.procession.total_length()}"
try:
alia = self.fission.get_current_puzzle_inf()['alia'] # type: ignore
s += f"谜题: {alia}\n"
except:
pass
try:
stat = self.phaser.__repr__('simple', '')
s += f"{stat}\n"
except:
pass
try:
stat = self.procession.__repr__('simple', '')
s += f"{stat}\n"
except:
pass
try:
stat = self.fission.__repr__('simple', '')
s += f"{stat}\n"
except Exception as e:
s = str(e)
#s += f"当前进度: {self.procession.process() + 1}/{self.procession.total_length()}"
return s
def update_display(self):
@@ -129,26 +151,38 @@ class MemScreen(Screen):
if new_rating == -1: # 安全值
return
self.fission.report(new_rating)
self.forward(new_rating)
self.rating = -1
def forward(self, rating):
self.update_state() # 刷新状态
if self.procession == None: # 已经完成记忆
return
forwards = 1 if rating >= 4 else 0 # 准许前进
self.rating = -1
logger.debug(f"试图前进: {"允许" if forwards else "禁止"}")
if forwards:
ret = self.procession.forward(1)
if ret == 0: # 若结束了此次队列
self.update_state()
if self.procession.phase == PhaserState.FINISHED: # 若所有队列都结束了
logger.debug(f"记忆进程结束")
self.mount_finished_widget()
return
else:
logger.debug(f"建立新队列 {self.procession.phase}")
self.update_state()
self.mount_puzzle()
else: # 若不通过
self.procession.append()
self.update_state()
allow_forward = 1 if rating >= 4 else 0
if allow_forward:
self.fission.forward()
if self.fission.state == 'retronly':
self.forward_atom(self.fission.get_quality())
self.update_state()
self.mount_puzzle()
self.update_display()
def forward_atom(self, quality):
logger.debug(f"Quality: {quality}")
if quality <= 3:
self.procession.append()
self.update_state() # 刷新状态
self.procession.forward(1)
self.update_state() # 刷新状态
if self.procession.phase == PhaserState.FINISHED: # 若所有队列都结束了
logger.debug(f"记忆进程结束")
self.mount_finished_widget()
return
self.fission = self.procession.get_fission()
def action_go_back(self):
self.app.pop_screen()
def action_quick_pass(self):
self.rating = 5
def action_quick_fail(self):
self.rating = 3

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

@@ -7,5 +7,5 @@ puzzle2widget = {
pz.RecognitionPuzzle: pzw.Recognition,
pz.ClozePuzzle: pzw.ClozePuzzle,
pz.MCQPuzzle: pzw.MCQPuzzle,
pz.BaseEvaluator: pzw.BasePuzzleWidget,
pz.BasePuzzle: pzw.BasePuzzleWidget,
}

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

@@ -9,7 +9,7 @@ orbital, 即轨道, 是定义队列式复习阶段流程的数据结构, 其实
orbital_example = {
"schedule": [列表 存储阶段(phases)名称]
"phases":{
阶段名称 = [["谜题(puzzle 现称 evaluator 评估器)名称", "概率系数 可大于1(整数部分为重复次数) 注意使用字符串包裹(toml 规范)"], ...],
阶段名称 = [["谜题(puzzle 现称 Puzzles 评估器)名称", "概率系数 可大于1(整数部分为重复次数) 注意使用字符串包裹(toml 规范)"], ...],
...
}
}

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

@@ -1,5 +1,5 @@
"""
Evaluator 模块 - 生成评估模块
Puzzles 模块 - 生成评估模块
提供多种类型的辅助评估生成器, 支持从字符串、字典等数据源导入题目
"""

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

@@ -5,6 +5,7 @@ import heurams.kernel.puzzles as puz
import heurams.kernel.particles as pt
from heurams.services.logger import get_logger
from tabulate import tabulate as tabu
from transitions import Machine
from .states import FissionState, PhaserState
@@ -17,20 +18,20 @@ class Fission(Machine):
self.phase = phase
self.cursor = 0
self.atom = atom
self.current_puzzle: puz.BasePuzzle
self.current_puzzle_inf: dict
# phase 为 PhaserState 枚举实例, 需要获取其value
phase_value = phase.value
orbital_schedule = atom.registry["orbital"]["phases"][phase_value] # type: ignore
orbital_puzzles = atom.registry["nucleon"]["puzzles"]
self.puzzles = list()
self.puzzles_inf = list()
self.min_ratings = []
for item, possibility in orbital_schedule: # type: ignore
self.logger.debug(f"开始处理: {item}")
logger.debug(f"开始处理: {item}")
if not isinstance(possibility, float):
possibility = float(possibility)
while possibility > 1:
self.puzzles.append(
self.puzzles_inf.append(
{
"puzzle": puz.puzzles[orbital_puzzles[item]["__origin__"]],
"alia": item,
@@ -39,16 +40,16 @@ class Fission(Machine):
possibility -= 1
if random.random() <= possibility:
self.puzzles.append(
self.puzzles_inf.append(
{
"puzzle": puz.puzzles[orbital_puzzles[item]["__origin__"]],
"alia": item,
}
)
self.current_puzzle_inf = self.puzzles_inf[0]
states = [
{"name": FissionState.EXAMMODE.value, "on_enter": "on_exammode"},
{"name": FissionState.RETRONLY.value, "on_enter": "on_retronly"},
{"name": FissionState.EXAMMODE.value},
{"name": FissionState.RETRONLY.value},
]
transitions = [
@@ -59,37 +60,56 @@ class Fission(Machine):
},
]
for i in range(len(self.puzzles_inf)):
self.min_ratings.append(0x3f3f3f3f)
Machine.__init__(
self,
states=states,
transitions=transitions,
initial="Evaluator_0",
initial=FissionState.EXAMMODE.value,
)
def get_puzzles(self):
def get_puzzles_inf(self):
if self.state == 'retronly':
return [puz.puzzles['recognition']]
return self.puzzles
return [{"puzzle": puz.puzzles['recognition'], "alia": "Recognition"}]
return self.puzzles_inf
def get_current_puzzle(self):
def get_current_puzzle_inf(self):
if self.state == 'retronly':
return puz.puzzles['recognition']
return self.current_puzzle
return {"puzzle": puz.puzzles['recognition'], "alia": "Recognition"}
return self.current_puzzle_inf
def report(self, rating):
self.min_ratings[self.cursor] = min(rating, self.min_ratings[self.cursor])
def get_quality(self):
if self.is_state("exammode", self):
logger.debug(f"CState: {self.state}")
if self.is_state("retronly", self):
return reduce(lambda x,y: min(x, y), self.min_ratings)
return -1
raise IndexError
def forward(self, step=1):
"""将谜题指针向前移动并依情况更新或完成"""
logger.debug("Procession.forward: step=%d, 当前 cursor=%d", step, self.cursor)
self.cursor += step
if self.cursor >= len(self.puzzles):
if self.cursor >= len(self.puzzles_inf):
if self.state != 'retronly':
self.finish()
else:
self.current_puzzle = self.puzzles[self.cursor]
self.current_puzzle_inf = self.puzzles_inf[self.cursor]
def __repr__(self, style="pipe", ends = "\n") -> str:
from heurams.services.textproc import truncate
dic = [
{
"Type": "Fission",
"Atom": truncate(self.atom.ident),
"State": self.state,
"Progress": f"{self.cursor + 1} / {len(self.puzzles_inf)}",
"Queue": list(map(lambda f: truncate(f['alia']), self.puzzles_inf)),
"Current Puzzle": f"{self.current_puzzle_inf['alia']}@{self.current_puzzle_inf['puzzle'].__name__}", # type: ignore
}
]
return str(tabu(dic, headers="keys", tablefmt=style)) + ends

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

@@ -128,7 +128,7 @@ class Phaser(Machine):
logger.debug("所有 Procession 已完成, 状态设置为 FINISHED")
return Procession([AtomPlaceholder()], PhaserState.FINISHED)
def __repr__(self):
def __repr__(self, style="pipe", ends = "\n"):
from heurams.services.textproc import truncate
from tabulate import tabulate as tabu
@@ -140,4 +140,4 @@ class Phaser(Machine):
"Current Procession": "None" if not self.current_procession() else self.current_procession().name_, # type: ignore
},
]
return str(tabu(tabular_data=lst, headers="keys", tablefmt="pipe")) + "\n"
return str(tabu(tabular_data=lst, headers="keys", tablefmt=style)) + ends

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

@@ -8,7 +8,6 @@ from .states import PhaserState, ProcessionState
logger = get_logger(__name__)
class Procession(Machine):
"""队列: 标识单次记忆流程"""
@@ -115,7 +114,7 @@ class Procession(Machine):
def get_fission(self):
return Fission(atom=self.current_atom, phase=self.phase) # type: ignore
def __repr__(self):
def __repr__(self, style="pipe", ends = "\n"):
from heurams.services.textproc import truncate
dic = [
@@ -128,4 +127,4 @@ class Procession(Machine):
"Current Atom": self.current_atom.ident, # type: ignore
}
]
return str(tabu(dic, headers="keys", tablefmt='pipe')) + "\n"
return str(tabu(dic, headers="keys", tablefmt=style)) + ends