pluv/HeurAMS
1
0
Fork 0
Code Issues Pull Requests Projects Releases 1 Activity

feat: 开发 unifront 前端会话模块

Browse Source
This commit is contained in:
Wang Zhiyu
2026-04-21 16:52:04 +08:00
parent fc70aa07f6
commit 093034828b
26 changed files with 191 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
data/config/global.toml
View File

@@ -1,4 +1,4 @@
zmq_debug = false
zmq_debug = true
_zmq_debug_desc = "[调试] ZeroMQ 调试服务器, 这会在 zmq_debug_port 上打开一个服务器\n调试工具可远程在 HeurAMS 内执行任意 python 代码, 无必要请关闭"
zmq_debug_port = 5555
_zmq_debug_port_desc = "[调试] ZeroMQ 调试服务器端口"

4
data/config/interface/_.toml
View File

@@ -1,4 +1,4 @@
_global_desc = "用户界面通用设置"
_widgets_desc = "各组件设置"
_screens_desc = "各界面设置"
_widgets_desc = "组件设置"
_screens_desc = "界面设置"
_puzzles_desc = "谜题生成器设置"

16
data/config/interface/global.toml
View File

@@ -1,3 +1,9 @@
show_header = true
_show_header_desc = "展示界面顶部的标题栏\n如果您想节省这一行空间, 可以禁用它"
clock_on_header = true
_clock_on_header_desc = "在界面顶部的标题栏显示时间"
change_window_title = true
_change_window_title_desc = "更改终端模拟器窗口的标题\n如果禁用了 header, 则建议启用"
persist_to_file = true
_persist_to_file_desc = "[调试] 将记忆更改保存到文件"
quick_pass = true
@@ -5,13 +11,15 @@ _quick_pass_desc = "[调试] 启用快速应答功能(跳过测验)"
auto_pass = false
_auto_pass_desc = "[调试] 自动通过测试模式"
scheduled_num = 420
_scheduled_num_desc = "默认记忆单元数量(可被单元集设置覆盖)"
_scheduled_num_desc = "默认记忆单元数量\n可被单元集设置覆盖"
refresh_on_resume = true
_refresh_on_resume_desc = "[调试] 每当 Screen 激活后都刷新状态"
algorithm = "SM-2"
_algorithm_desc = "默认记忆调度算法(可被单元集设置覆盖)"
_algorithm_desc = "默认记忆调度算法\n可被单元集设置覆盖"
[_algorithm_candidate]
NSP-0 = "筛选用非间隔重复调度器"
none = "不设置默认调度器"
SM-2 = "第二代 SuperMemo 简单间隔重复调度器"
SM-15M = "第15代 SuperMemo 复杂间隔重复调度器 (不稳定且逆向工程)"
SM-2 = "第二代 SuperMemo 简单间隔重复调度器\nWozniak 于 1987 年提出, Anki 的默认算法"
SM-15M = "类第15代 SuperMemo 复杂间隔重复调度器\n不稳定且逆向工程"
FSRS = "先进开放间隔重复调度器"

2
data/repo/cngk-t/schedule.toml
View File

@@ -1,6 +1,6 @@
schedule = ["quick_review", "recognition", "final_review"]
[phases]
[routes]
quick_review = [["FillBlank", "1.0"], ["Recognition", "1.0"]]
recognition = [["FillBlank", "1.0"]]
final_review = [["FillBlank", "1.0"], ["Recognition", "1.0"]]

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

@@ -25,6 +25,7 @@ from .screens.navigator import NavigatorScreen
from .screens.precache import PrecachingScreen
from .screens.setting import SettingScreen
from .screens.synctool import SyncScreen
from . import shim
_end = perf_counter()
print(f"已完成! (耗时: {round(1000 * (_end - _start))}ms)")
@@ -56,6 +57,9 @@ class HeurAMSApp(App):
"setting": SettingScreen,
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def on_mount(self) -> None:
self.push_screen("dashboard")

2
src/heurams/interface/css/screens/dashboard.tcss
View File

@@ -8,7 +8,7 @@
.repo-list-item {
layout: grid;
grid-size: 2;
grid-size: 1;
height: 3;
}

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

@@ -5,6 +5,8 @@ from textual.containers import ScrollableContainer
from textual.screen import Screen
from textual.widgets import Button, Footer, Header, Label, Markdown, Static
from textual import events, on
import heurams.services.version as version
from heurams.context import *
import platform
@@ -19,8 +21,19 @@ class AboutScreen(Screen):
("q", "go_back", "返回"),
]
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer(id="about_container"):
yield Label("[b]关于与版本信息[/b]")

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

@@ -10,6 +10,8 @@ from textual.containers import ScrollableContainer, Container, Horizontal, Verti
from textual.screen import Screen
from textual.widgets import Button, Footer, Header, Label, ListItem, ListView, Static
from textual.layouts import horizontal
from textual import events, on
from textual.reactive import reactive
import heurams.kernel.particles as pt
import heurams.services.timer as timer
@@ -37,6 +39,8 @@ class DashboardScreen(Screen):
CSS_PATH = rootdir / "interface" / "css" / "screens" / "dashboard.tcss"
repolink = reactive({})
def __init__(
self,
name: str | None = None,
@@ -44,12 +48,12 @@ class DashboardScreen(Screen):
classes: str | None = None,
) -> None:
super().__init__(name, id, classes)
self.repolink = {}
self._load_data()
def compose(self) -> ComposeResult:
"""组合界面组件"""
self._load_data()
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer():
yield Horizontal( # 顶部的状态
Vertical(
@@ -58,7 +62,7 @@ class DashboardScreen(Screen):
f"应用时区修正: UTC+{str(config_var.get()['services']['timer']['timezone_offset'] / 3600).rstrip('.0')}"
),
Label(
f"默认算法设置: {config_var.get()['interface']['global']['algorithm']}"
f"默认算法设置: {config_var.get()['interface']['global']['algorithm']}",
),
classes="left",
),
@@ -81,6 +85,13 @@ class DashboardScreen(Screen):
yield Label(f"版本 {version.ver} {version.stage.capitalize()}") # 版本信息
yield Footer()
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
# https://github.com/Textualize/textual/discussions/4268
# self.refresh(recompose=True) 此函数有问题且官方不管 而且性能低
def _load_data(self):
repo_dirs = Repo.probe_valid_repos_in_dir(
Path(config_var.get()["global"]["paths"]["repo"])
@@ -154,7 +165,7 @@ class DashboardScreen(Screen):
for r in self.repos:
self.repolink[str(id(r))] = r # 用于规避 ctype id 对象还原
list_item = ListItem(
Label(r.prompt),
*[Label(line) for line in r.prompt.splitlines()],
Button(
f"开始学习",
flat=True,

13
src/heurams/interface/screens/favmgr.py
View File

@@ -4,6 +4,8 @@ import base64
from pathlib import Path
from typing import List, Optional
from textual import events, on
from textual.app import ComposeResult
from textual.containers import ScrollableContainer
from textual.screen import Screen
@@ -18,6 +20,7 @@ from textual.widgets import (
Static,
)
from textual import events, on
from heurams.context import config_var
from heurams.kernel.repolib import Repo
from heurams.services.favorite_service import FavoriteItem, favorite_manager
@@ -53,7 +56,9 @@ class FavoriteManagerScreen(Screen):
def compose(self) -> ComposeResult:
"""组合界面组件"""
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer(id="favorites-container"):
if not self.favorites:
yield Label("暂无收藏", classes="empty-label")
@@ -63,6 +68,12 @@ class FavoriteManagerScreen(Screen):
yield ListView(id="favorites-list")
yield Footer()
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
def on_mount(self) -> None:
"""挂载后填充列表"""
if self.favorites:

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

@@ -10,6 +10,8 @@ from textual.reactive import reactive
from textual.screen import Screen
from textual.widgets import Button, Footer, Header, Label, Static
from textual import events, on
import heurams.kernel.particles as pt
import heurams.kernel.puzzles as pz
from heurams.context import config_var, rootdir
@@ -17,10 +19,13 @@ from heurams.kernel.reactor import *
from heurams.services.favorite_service import favorite_manager
from heurams.services.logger import get_logger
import pickle
from .. import shim
logger = get_logger(__name__)
class MemScreen(Screen):
BINDINGS = [
("q", "go_back", "返回"),
@@ -33,7 +38,8 @@ class MemScreen(Screen):
("z", "block_prompt"),
]
CSS_PATH = rootdir / 'interface' / 'css' / 'screens' / 'memoqueue.tcss'
SUB_TITLE = "学习中"
CSS_PATH = rootdir / "interface" / "css" / "screens" / "memoqueue.tcss"
if config_var.get()["interface"]["global"]["quick_pass"]:
BINDINGS.append(("k", "quick_pass", "正确应答"))
@@ -55,8 +61,15 @@ class MemScreen(Screen):
self.update_state()
self.expander: Expander
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer():
yield Label(self._get_progress_text(), id="head_stat")
yield ScrollableContainer(id="puzzle_container")
@@ -83,7 +96,7 @@ class MemScreen(Screen):
return Static(f"无法生成谜题 {e}")
def _get_progress_text(self):
s = f"阶段: {self.procession.phase.name}\n"
s = f"阶段: {self.procession.route.name}\n"
# 收藏状态
if self.repo is not None:
fav_status = "已收藏" if self._is_current_atom_favorited() else "未收藏"
@@ -98,7 +111,7 @@ class MemScreen(Screen):
def mount_puzzle(self):
"""挂载当前谜题组件"""
if self.procession.phase == RouterState.FINISHED:
if self.procession.route == RouterState.FINISHED:
self.mount_finished_widget()
return
container = self.query_one("#puzzle_container")
@@ -144,7 +157,7 @@ class MemScreen(Screen):
if new_rating == -1: # 安全值
return
self.update_state()
if self.procession.phase == RouterState.FINISHED:
if self.procession.route == RouterState.FINISHED:
rating = -1
return
self.expander.report(new_rating)

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

@@ -5,6 +5,8 @@ from textual.containers import Grid, ScrollableContainer
from textual.screen import ModalScreen
from textual.widgets import Button, Footer, Header, Label, ListItem, ListView, Static
from textual import events, on
from heurams.context import *
from heurams.services.logger import get_logger

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

@@ -8,6 +8,8 @@ from textual.screen import Screen
from textual.widgets import Button, Footer, Header, Label, ProgressBar, Static
from textual.worker import get_current_worker
from textual import events, on
import heurams.kernel.particles as pt
import heurams.services.hasher as hasher
from heurams.context import *
@@ -80,6 +82,12 @@ class PrecachingScreen(Screen):
continue
return total
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
def _update_cache_stats(self) -> None:
"""更新缓存统计信息"""
total_size = 0
@@ -103,7 +111,9 @@ class PrecachingScreen(Screen):
self.cache_stats["cache_rate"] = cache_rate
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer(id="precache_container"):
yield Label("[b]音频预缓存[/b]", classes="title-label")
with Container():

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

@@ -17,6 +17,8 @@ from textual.widgets import (
)
from textual.lazy import Reveal, Lazy
from textual import events, on
import heurams.kernel.particles as pt
import heurams.services.hasher as hasher
from heurams.context import *
@@ -34,7 +36,7 @@ class PreparationScreen(Screen):
BINDINGS = [
("q", "go_back", "返回"),
("p", "precache", "预缓存音频"),
("p", "precache", "缓存"),
("d", "toggle_dark", ""),
("0,1,2,3", "app.push_screen('about')", ""),
]
@@ -46,8 +48,16 @@ class PreparationScreen(Screen):
self.repo = repo
self.load_data()
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer(id="main_container"):
yield Markdown(
f"**准备就绪**: `{self.repo.manifest['title']}`\n", id="title"
@@ -72,7 +82,7 @@ class PreparationScreen(Screen):
classes="btn",
),
Button(
"预缓存音频",
"管理缓存",
id="precache_button",
variant="success",
classes="btn",

31
src/heurams/interface/screens/setting.py
View File

@@ -23,6 +23,8 @@ from textual.widgets import (
)
from textual.layouts import horizontal
from textual import events, on
import heurams.kernel.particles as pt
import heurams.services.timer as timer
import heurams.services.version as version
@@ -54,9 +56,16 @@ class SettingScreen(Screen):
) -> None:
super().__init__(name, id, classes)
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
def compose(self) -> ComposeResult:
"""组合界面组件"""
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer():
yield Label("[b]设置页面[/b]")
for i in config_var.get():
@@ -65,7 +74,7 @@ class SettingScreen(Screen):
a = self._get_subcfg(f"{i}")
if a:
yield Collapsible(
*a, title=i + f'\n{config_var.get().get(f"_{i}_desc", "")}'
*a, title=i + f'\n[d]{config_var.get().get(f"_{i}_desc", "")}[/d]'
)
yield Label(
"退出页面时, 所作的更改会立即保存, 但仍建议重启软件以确保新的配置得到应用",
@@ -85,7 +94,7 @@ class SettingScreen(Screen):
if a:
lst.append(
Collapsible(
*a, title=i + f'\n{parent.get(f"_{i}_desc", "")}'
*a, title=i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'
)
)
return lst
@@ -101,17 +110,17 @@ class SettingScreen(Screen):
if a:
lst.append(
Collapsible(
*a, title=i + f'\n{parent.get(f"_{i}_desc", "")}'
*a, title=i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'
)
)
elif f"_{i}_candidate" in parent: # 选择框模式
if isinstance(parent[f"_{i}_candidate"], dict):
lst.append(
Horizontal(
Label(i + f'\n{parent.get(f"_{i}_desc", "")}'),
Label(i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'),
Select(
(
(f"{j} ({k})", j)
(f"{j}\n[d]{k}[/d]", j)
for j, k in parent[f"_{i}_candidate"].items()
),
prompt=f'{parent.get(f"{i}", "")}',
@@ -122,7 +131,7 @@ class SettingScreen(Screen):
elif isinstance(parent[f"_{i}_candidate"], list):
lst.append(
Horizontal(
Label(i + f'\n{parent.get(f"_{i}_desc", "")}'),
Label(i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'),
Select(
((j, j) for j in parent[f"_{i}_candidate"]),
prompt=f'{parent.get(f"{i}", "")}',
@@ -134,7 +143,7 @@ class SettingScreen(Screen):
if isinstance(parent[i], float):
lst.append(
Horizontal(
Label(i + f'\n{parent.get(f"_{i}_desc", "")}'),
Label(i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'),
Input(
value=str(parent[i]),
placeholder="要求一个浮点数",
@@ -146,7 +155,7 @@ class SettingScreen(Screen):
elif isinstance(parent[i], str):
lst.append(
Horizontal(
Label(i + f'\n{parent.get(f"_{i}_desc", "")}'),
Label(i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'),
Input(
value=parent[i],
placeholder="要求一个字符串",
@@ -158,7 +167,7 @@ class SettingScreen(Screen):
elif isinstance(parent[i], bool):
lst.append(
Horizontal(
Label(i + f'\n{parent.get(f"_{i}_desc", "")}'),
Label(i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'),
Switch(
value=parent[i], id=domize(f"{parent_epath}.{i}")
),
@@ -167,7 +176,7 @@ class SettingScreen(Screen):
elif isinstance(parent[i], int):
lst.append(
Horizontal(
Label(i + f'\n{parent.get(f"_{i}_desc", "")}'),
Label(i + f'\n[d]{parent.get(f"_{i}_desc", "")}[/d]'),
Input(
value=str(parent[i]),
placeholder="要求一个整数",

12
src/heurams/interface/screens/synctool.py
View File

@@ -9,6 +9,8 @@ from textual.screen import Screen
from textual.widgets import Button, Footer, Header, Label, ProgressBar, Static
from textual.worker import get_current_worker
from textual import events, on
import heurams.kernel.particles as pt
import heurams.services.hasher as hasher
from heurams.context import *
@@ -26,8 +28,16 @@ class SyncScreen(Screen):
self.log_messages = []
self.max_log_lines = 50
@on(events.ScreenResume)
def post_active(self, event):
from heurams.interface import shim
shim.set_term_title(f"{self.app.TITLE} - {self.SUB_TITLE}")
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
if config_var.get()['interface']['global']['show_header']:
yield Header(show_clock=config_var.get()['interface']['global']['clock_on_header'])
with ScrollableContainer(id="sync_container"):
# 标题和连接状态
yield Static("同步工具", classes="title")

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

@@ -2,6 +2,8 @@
import heurams.interface.widgets as pzw
import heurams.kernel.puzzles as pz
import platform, os, sys
from heurams.context import config_var
puzzle2widget = {
pz.RecognitionPuzzle: pzw.Recognition,
@@ -9,3 +11,13 @@ puzzle2widget = {
pz.MCQPuzzle: pzw.MCQPuzzle,
pz.BasePuzzle: pzw.BasePuzzleWidget,
}
def set_term_title(title):
if not config_var.get()['interface']['global']['change_window_title']:
return
system = platform.system()
if system == "Windows":
os.system(f"title {title}")
else: # Linux, Mac, etc.
os.write(2, f"\033]2;{title}\007".encode("utf-8"))

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

@@ -1271,7 +1271,7 @@ class SM:
self.ofm = OFM(self)
def _find_index_to_insert(self, item, r=None):
"""Binary search to find insertion index for sorted queue."""
"""Binary search to find insertion index for sorted procession."""
if r is None:
r = list(range(len(self.q)))
@@ -1290,7 +1290,7 @@ class SM:
return self._find_index_to_insert(item, r[i:])
def add_item(self, value):
"""Add a new item to the queue."""
"""Add a new item to the procession."""
item = Item(self, value)
index = self._find_index_to_insert(item)
self.q.insert(index, item)
@@ -1330,7 +1330,7 @@ class SM:
item.answer(grade, now)
def discard(self, item):
"""Remove item from queue."""
"""Remove item from procession."""
if item in self.q:
self.q.remove(item)

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

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

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

@@ -6,7 +6,7 @@ from .nucleon import Nucleon
orbital_placeholder = {
"schedule": ["quick_review", "recognition", "final_review"],
"phases": {
"routes": {
"quick_review": [
["FillBlank", 1.0],
["SelectMeaning", 0.5],

12
src/heurams/kernel/reactor/README.md
View File

@@ -83,16 +83,16 @@ Router 的 __repr__ 定义了此对象"官方的显示"用作直观的调试.\
- current_atom: 当前记忆原子的引用
- atoms: 队列中所有原子列表
- cursor: 指针, 是当前原子在 atoms 列表中的索引
- phase: "阶段属性"
- route: "阶段属性"
> 注意区分 "Router" 和 "Phase", 其中 "Phase" 表示 "Router State".
> 注意区分 "Router" 和 "Route", 其中 "Route" 表示 "Router State".
- name\_: 阶段的命名
- state: 当前状态属性
### 初始化
接受一个 atoms 列表与 phase_state (RouterState Enum 类型)对象
接受一个 atoms 列表与 route_state (RouterState Enum 类型)对象
### 直接输出呈现形式
@@ -100,12 +100,12 @@ Router 的 __repr__ 定义了此对象"官方的显示"用作直观的调试.\
与 Router 不同, Procession 显示队列会对过长的 atom.ident 进行缩略(末尾 `>` 符号)
```text
| Type | Name | State | Progress | Queue | Current Atom |
| Type | Name | State | Progress | Procession | Current Atom |
| :--------- | :----- | :----- | :------- | :--------------------- | :---------------------------- |
| Procession | 新记忆 | active | 1 / 2 | ['秦孝公>', '君臣固>'] | 秦孝公据崤函之固, 拥雍州之地, |
```
| Type | Name | State | Progress | Queue | Current Atom |
| Type | Name | State | Progress | Procession | Current Atom |
| :--------- | :----- | :----- | :------- | :--------------------- | :---------------------------- |
| Procession | 新记忆 | active | 1 / 2 | ['秦孝公>', '君臣固>'] | 秦孝公据崤函之固, 拥雍州之地, |
@@ -166,7 +166,7 @@ Router 的 __repr__ 定义了此对象"官方的显示"用作直观的调试.\
### 初始化
接受 atom 对象和 phase 参数
接受 atom 对象和 route 参数
### 方法

16
src/heurams/kernel/reactor/expander.py
View File

@@ -16,13 +16,13 @@ logger = get_logger(__name__)
class Expander(Machine):
"""单原子调度展开器"""
def __init__(self, atom: pt.Atom, phase=RouterState.RECOGNITION):
self.phase = phase
def __init__(self, atom: pt.Atom, route=RouterState.RECOGNITION):
self.route = route
self.cursor = 0
self.atom = atom
self.current_puzzle_inf: dict
# phase 为 RouterState 枚举实例, 需要获取其value
phase_value = phase.value
# route 为 RouterState 枚举实例, 需要获取其value
route_value = route.value
states = [
{"name": ExpanderState.EXAMMODE.value},
{"name": ExpanderState.RETRONLY.value},
@@ -35,7 +35,7 @@ class Expander(Machine):
"dest": ExpanderState.RETRONLY.value,
},
]
if phase == RouterState.FINISHED:
if route == RouterState.FINISHED:
Machine.__init__(
self,
states=states,
@@ -43,7 +43,7 @@ class Expander(Machine):
initial=ExpanderState.EXAMMODE.value,
)
return
orbital_schedule = atom.registry["orbital"]["phases"][phase_value] # type: ignore
orbital_schedule = atom.registry["orbital"]["routes"][route_value] # type: ignore
orbital_puzzles = atom.registry["nucleon"]["puzzles"]
self.puzzles_inf = list()
self.min_ratings = []
@@ -122,7 +122,9 @@ class Expander(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)),
"Procession": 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
}
]

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

@@ -13,11 +13,11 @@ logger = get_logger(__name__)
class Procession(Machine):
"""队列: 标识单次记忆流程"""
def __init__(self, atoms: list, phase_state: RouterState, name_: str = ""):
def __init__(self, atoms: list, route_state: RouterState, name_: str = ""):
logger.debug(
"Procession.__init__: 原子数量=%d, phase=%s, name='%s'",
"Procession.__init__: 原子数量=%d, route=%s, name='%s'",
len(atoms),
phase_state.value,
route_state.value,
name_,
)
self.current_atom: pt.Atom | None
@@ -25,7 +25,7 @@ class Procession(Machine):
self.current_atom = atoms[0] if atoms else None
self.cursor = 0
self.name_ = name_
self.phase = phase_state
self.route = route_state
states = [
{"name": ProcessionState.ACTIVE.value, "on_enter": "on_active"},
@@ -114,7 +114,7 @@ class Procession(Machine):
return empty
def get_expander(self):
return Expander(atom=self.current_atom, phase=self.phase) # type: ignore
return Expander(atom=self.current_atom, route=self.route) # type: ignore
def __repr__(self, style="pipe", ends="\n"):
from heurams.services.textproc import truncate
@@ -125,7 +125,7 @@ class Procession(Machine):
"Name": self.name_,
"State": self.state,
"Progress": f"{self.cursor + 1} / {len(self.atoms)}",
"Queue": list(map(lambda f: truncate(f.ident), self.atoms)),
"Procession": list(map(lambda f: truncate(f.ident), self.atoms)),
"Current Atom": self.current_atom.ident, # type: ignore
}
]

17
src/heurams/kernel/reactor/router.py
View File

@@ -11,10 +11,10 @@ logger = get_logger(__name__)
class Router(Machine):
"""全局调度阶段管理器"""
"""全局调度阶段路由器"""
def __init__(self, atoms: list[pt.Atom]) -> None:
logger.debug("Router.__init__: 原子数量=%d", len(atoms))
logger.debug(f"Router.__init__: 原子数量={len(atoms)}")
self.atoms = atoms
new_atoms = list()
@@ -26,9 +26,10 @@ class Router(Machine):
else:
old_atoms.append(i)
logger.debug("新原子数量=%d, 旧原子数量=%d", len(new_atoms), len(old_atoms))
logger.debug(f"新原子数量={len(new_atoms)}, 旧原子数量={len(old_atoms)}")
self.processions = list()
"""路由中的所有队列"""
# TODO: 改进为基于配置文件的可选复习阶段
if len(old_atoms):
self.processions.append(
@@ -116,15 +117,15 @@ class Router(Machine):
for i in self.processions:
i: Procession
if i.state != ProcessionState.FINISHED.value:
# if i.phase == RouterState.UNSURE: 此判断是不必要的 因为没有这种 Procession
if i.phase == RouterState.QUICK_REVIEW:
# if i.route == RouterState.UNSURE: 此判断是不必要的 因为没有这种 Procession
if i.route == RouterState.QUICK_REVIEW:
self.to_quick_review()
elif i.phase == RouterState.RECOGNITION:
elif i.route == RouterState.RECOGNITION:
self.to_recognition()
elif i.phase == RouterState.FINAL_REVIEW:
elif i.route == RouterState.FINAL_REVIEW:
self.to_final_review()
logger.debug("找到未完成的 Procession: phase=%s", i.phase)
logger.debug("找到未完成的 Procession: route=%s", i.route)
return i
# 所有Procession都已完成

1
src/heurams/services/session.py
View File

@@ -1 +0,0 @@
"""会话记录模块"""

3
src/heurams/unifront/session.py Normal file
View File

@@ -0,0 +1,3 @@
"""会话模块"""
class Session:
pass