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

fix: 修复完成屏幕问题

Browse Source
This commit is contained in:
Wang Zhiyu
2026-01-06 19:58:46 +08:00
parent 22b41789eb
commit ca7ef92b05
6 changed files with 246 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -80,27 +80,28 @@ class MemScreen(Screen):
def _get_progress_text(self):
s = f"阶段: {self.procession.phase.name}\n"
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()}"
if config_var.get().get("debug_topline", 0):
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):
@@ -110,6 +111,9 @@ class MemScreen(Screen):
def mount_puzzle(self):
"""挂载当前谜题组件"""
if self.procession.phase == PhaserState.FINISHED:
self.mount_finished_widget()
return
container = self.query_one("#puzzle-container")
for i in container.children:
i.remove()
@@ -150,16 +154,20 @@ class MemScreen(Screen):
def watch_rating(self, old_rating, new_rating) -> None:
if new_rating == -1: # 安全值
return
self.update_state()
if self.procession.phase == PhaserState.FINISHED:
rating = -1
return
self.fission.report(new_rating)
self.forward(new_rating)
self.rating = -1
def forward(self, rating):
self.update_state()
allow_forward = 1 if rating >= 4 else 0
if allow_forward:
self.fission.forward()
if self.fission.state == 'retronly':
if self.fission.state == "retronly":
self.forward_atom(self.fission.get_quality())
self.update_state()
self.mount_puzzle()
@@ -172,15 +180,11 @@ class MemScreen(Screen):
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

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

@@ -11,6 +11,7 @@ from .states import FissionState, PhaserState
logger = get_logger(__name__)
class Fission(Machine):
"""单原子调度展开器"""
@@ -21,6 +22,26 @@ class Fission(Machine):
self.current_puzzle_inf: dict
# phase 为 PhaserState 枚举实例, 需要获取其value
phase_value = phase.value
states = [
{"name": FissionState.EXAMMODE.value},
{"name": FissionState.RETRONLY.value},
]
transitions = [
{
"trigger": "finish",
"source": FissionState.EXAMMODE.value,
"dest": FissionState.RETRONLY.value,
},
]
if phase == PhaserState.FINISHED:
Machine.__init__(
self,
states=states,
transitions=transitions,
initial=FissionState.EXAMMODE.value,
)
return
orbital_schedule = atom.registry["orbital"]["phases"][phase_value] # type: ignore
orbital_puzzles = atom.registry["nucleon"]["puzzles"]
self.puzzles_inf = list()
@@ -47,21 +68,9 @@ class Fission(Machine):
}
)
self.current_puzzle_inf = self.puzzles_inf[0]
states = [
{"name": FissionState.EXAMMODE.value},
{"name": FissionState.RETRONLY.value},
]
transitions = [
{
"trigger": "finish",
"source": FissionState.EXAMMODE.value,
"dest": FissionState.RETRONLY.value,
},
]
for i in range(len(self.puzzles_inf)):
self.min_ratings.append(0x3f3f3f3f)
self.min_ratings.append(0x3F3F3F3F)
Machine.__init__(
self,
@@ -71,35 +80,33 @@ class Fission(Machine):
)
def get_puzzles_inf(self):
if self.state == 'retronly':
return [{"puzzle": puz.puzzles['recognition'], "alia": "Recognition"}]
if self.state == "retronly":
return [{"puzzle": puz.puzzles["recognition"], "alia": "Recognition"}]
return self.puzzles_inf
def get_current_puzzle_inf(self):
if self.state == 'retronly':
return {"puzzle": puz.puzzles['recognition'], "alia": "Recognition"}
if self.state == "retronly":
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):
logger.debug(f"CState: {self.state}")
if self.is_state("retronly", self):
return reduce(lambda x,y: min(x, y), self.min_ratings)
return reduce(lambda x, y: min(x, y), self.min_ratings)
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_inf):
if self.state != 'retronly':
if self.state != "retronly":
self.finish()
else:
self.current_puzzle_inf = self.puzzles_inf[self.cursor]
def __repr__(self, style="pipe", ends = "\n") -> str:
def __repr__(self, style="pipe", ends="\n") -> str:
from heurams.services.textproc import truncate
dic = [
@@ -108,7 +115,7 @@ class Fission(Machine):
"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)),
"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
}
]

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

@@ -112,7 +112,7 @@ class Phaser(Machine):
for i in self.processions:
i: Procession
if i.state != ProcessionState.FINISHED.value:
#if i.phase == PhaserState.UNSURE: 此判断是不必要的 因为没有这种 Procession
# if i.phase == PhaserState.UNSURE: 此判断是不必要的 因为没有这种 Procession
if i.phase == PhaserState.QUICK_REVIEW:
self.to_quick_review()
elif i.phase == PhaserState.RECOGNITION:
@@ -128,7 +128,7 @@ class Phaser(Machine):
logger.debug("所有 Procession 已完成, 状态设置为 FINISHED")
return Procession([AtomPlaceholder()], PhaserState.FINISHED)
def __repr__(self, style="pipe", ends = "\n"):
def __repr__(self, style="pipe", ends="\n"):
from heurams.services.textproc import truncate
from tabulate import tabulate as tabu

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

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

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

@@ -12,12 +12,15 @@ class PhaserState(Enum):
FINAL_REVIEW = "final_review"
FINISHED = "finished"
class ProcessionState(Enum):
ACTIVE = "active"
FINISHED = "finished"
class FissionState(Enum):
EXAMMODE = "exammode"
RETRONLY = "retronly"
logger.debug("状态枚举定义已加载")