ajax/HeurAMS
0
1
Fork 0
Code Pull Requests Releases 13 Activity

基本对象系统移植

Browse Source
This commit is contained in:
Wang Zhiyu
2025-10-16 00:04:53 +08:00
parent aa99aa7686
commit 18df7380f5
17 changed files with 511 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files

278
src/heurams/kernel/particles.py
View File

@@ -1,278 +0,0 @@
#!/usr/bin/env python3
import pathlib
import toml
import time
import heurams.services.timer as timer
from typing import List
class Electron:
"""电子: 记忆分析元数据及算法"""
algorithm = "SM-2" # 暂时使用 SM-2 算法进行记忆拟合, 考虑 SM-15 替代
def __init__(self, content: str, metadata: dict):
self.content = content
self.metadata = metadata
if metadata == {}:
# print("NULL")
self._default_init()
def _default_init(self):
defaults = {
'efactor': 2.5, # 易度系数, 越大越简单, 最大为5
'real_rept': 0, # (实际)重复次数
'rept': 0, # (有效)重复次数
'interval': 0, # 最佳间隔
'last_date': 0, # 上一次复习的时间戳
'next_date': 0, # 将要复习的时间戳
'is_activated': 0, # 激活状态
# *NOTE: 此处"时间戳"是以天为单位的整数, 即 UNIX 时间戳除以一天的秒数取整
'last_modify': time.time() # 最后修改时间戳(此处是UNIX时间戳)
}
self.metadata = defaults
def activate(self):
self.metadata['is_activated'] = 1
self.metadata['last_modify'] = time.time()
def modify(self, var: str, value):
if var in self.metadata:
self.metadata[var] = value
self.metadata['last_modify'] = time.time()
else:
print(f"警告: '{var}' 非已知元数据字段")
def revisor(self, quality: int = 5, is_new_activation: bool = False):
"""SM-2 算法迭代决策机制实现
根据 quality(0 ~ 5) 进行参数迭代最佳间隔
quality 由主程序评估
Args:
quality (int): 记忆保留率量化参数
"""
print(f"REVISOR: {quality}, {is_new_activation}")
if quality == -1:
return -1
self.metadata['efactor'] = self.metadata['efactor'] + (
0.1 - (5 - quality) * (0.08 + (5 - quality) * 0.02)
)
self.metadata['efactor'] = max(1.3, self.metadata['efactor'])
if quality < 3:
# 若保留率低于 3，重置重复次数
self.metadata['rept'] = 0
self.metadata['interval'] = 0 # 设为0，以便下面重新计算 I(1)
else:
self.metadata['rept'] += 1
self.metadata['real_rept'] += 1
if is_new_activation: # 初次激活
self.metadata['rept'] = 0
self.metadata['efactor'] = 2.5
if self.metadata['rept'] == 0: # 刚被重置或初次激活后复习
self.metadata['interval'] = 1 # I(1)
elif self.metadata['rept'] == 1:
self.metadata['interval'] = 6 # I(2) 经验公式
else:
self.metadata['interval'] = round(
self.metadata['interval'] * self.metadata['efactor']
)
self.metadata['last_date'] = timer.get_daystamp()
self.metadata['next_date'] = timer.get_daystamp() + self.metadata['interval']
self.metadata['last_modify'] = time.time()
def __str__(self):
return (
f"记忆单元预览 \n"
f"内容: '{self.content}' \n"
f"易度系数: {self.metadata['efactor']:.2f} \n"
f"已经重复的次数: {self.metadata['rept']} \n"
f"下次间隔: {self.metadata['interval']} 天 \n"
f"下次复习日期时间戳: {self.metadata['next_date']}"
)
def __eq__(self, other):
if self.content == other.content:
return True
return False
def __hash__(self):
return hash(self.content)
def __getitem__(self, key):
if key == "content":
return self.content
if key in self.metadata:
return self.metadata[key]
else:
raise KeyError(f"Key '{key}' not found in metadata.")
def __setitem__(self, key, value):
if key == "content":
raise AttributeError("content 应为只读")
self.metadata[key] = value
self.metadata['last_modify'] = time.time()
def __iter__(self):
yield from self.metadata.keys()
def __len__(self):
return len(self.metadata)
@staticmethod
def placeholder():
return Electron("电子对象样例内容", {})
class Nucleon:
"""核子: 材料元数据"""
def __init__(self, content: str, data: dict):
self.metadata = data
self.content = content
def __getitem__(self, key):
if key == "content":
return self.content
if key in self.metadata:
return self.metadata[key]
else:
raise KeyError(f"Key '{key}' not found in metadata.")
def __iter__(self):
yield from self.metadata.keys()
def __len__(self):
return len(self.metadata)
def __hash__(self):
return hash(self.content)
@staticmethod
def placeholder():
return Nucleon("核子对象样例内容", {})
class NucleonUnion():
"""
替代原有 NucleonFile 类, 支持复杂逻辑
Attributes:
path (Path): 对应于 NucleonUnion 实例的文件路径。
name (str): 核联对象的显示名称，从文件名中派生。
nucleons (list): 内部核子对象的列表。
nucleons_dict (dict): 内部核子对象的字典，以核子内容作为键。
keydata (dict): 核子对象字典键名的翻译。
testdata (dict): 记忆测试项目的元数据。
Parameters:
path (Path): 包含核子数据的文件路径。
"""
def __init__(self, path: pathlib.Path):
self.path = path
self.name = path.name.replace(path.suffix, "")
with open(path, 'r') as f:
all = toml.load(f)
lst = list()
for i in all.keys():
if "attr" in i:
continue
if "data" in i:
continue
lst.append(Nucleon(i, all[i]))
self.keydata = all["keydata"]
self.testdata = all["testdata"]
self.nucleons: List[Nucleon] = lst
self.nucleons_dict = {i.content: i for i in lst}
def __len__(self):
return len(self.nucleons)
def linked_electron_union(self):
if (self.path.parent / '..' / 'electron' / self.path.name).exists():
return ElectronUnion(self.path.parent / '..' / 'electron' / self.path.name)
else:
return 0
def save(self):
with open(self.path, 'w') as f:
tmp = {i.content: i.metadata for i in self.nucleons}
toml.dump(tmp, f)
class ElectronUnion:
"""取代原有 ElectronFile 类, 以支持复杂逻辑"""
def __init__(self, path):
self.path = path
print(path)
self.name = path.name.replace(path.suffix, "")
with open(path, 'r') as f:
all = toml.load(f)
lst = list()
for i in all.keys():
if i != "total":
lst.append(Electron(i, all[i]))
self.total = all.get("total", {"last_date": 0})
self.electrons = lst
self.electrons_dict = {i.content: i for i in lst}
def sync(self):
"""同步 electrons_dict 中新增对到 electrons 中, 仅用于缺省初始化不存在映射时调用"""
self.electrons = self.electrons_dict.values()
def __len__(self):
return len(self.electrons)
def linked_nucleon_union(self):
return NucleonUnion(self.path.parent / '..' / 'nucleon' / self.path.name)
def save(self):
# print(1)
self.total["last_date"] = timer.get_daystamp()
with open(self.path, 'w') as f:
tmp = {i.content: i.metadata for i in self.electrons}
tmp["total"] = self.total
# print(tmp)
toml.dump(tmp, f)
class Atom:
@staticmethod
def placeholder():
return (Electron.placeholder(), Nucleon.placeholder(), {})
@staticmethod
def advanced_placeholder():
return (
Electron("两只黄鹤鸣翠柳", {}),
Nucleon(
"两只黄鹤鸣翠柳",
{
"note": [],
"translation": "臣子李密陈言：我因命运不好，小时候遭遇到了不幸",
"keyword_note": {
"险衅": "凶险祸患（这里指命运不好）",
"夙": "早时，这里指年幼的时候",
"闵": "通'悯'，指可忧患的事",
"凶": "不幸，指丧父"
}
}
),
{
"keydata": {
"note": "笔记",
"keyword_note": "关键词翻译",
"translation": "语句翻译"
},
"testdata": {
"additional_inf": ["translation", "note", "keyword_note"],
"fill_blank_test": ["translation"],
"draw_card_test": ["keyword_note"]
},
"is_new_activation": 0
}
)

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

@@ -0,0 +1,4 @@
from .electron import Electron
from .nucleon import Nucleon
from .unions import ElectronUnion, NucleonUnion
from .atom import Atom

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

@@ -0,0 +1,39 @@
from electron import Electron
from nucleon import Nucleon
class Atom:
@staticmethod
def placeholder():
return (Electron.placeholder(), Nucleon.placeholder(), {})
@staticmethod
def advanced_placeholder():
return (
Electron("两只黄鹤鸣翠柳", {}),
Nucleon(
"两只黄鹤鸣翠柳",
{
"note": [],
"translation": "臣子李密陈言：我因命运不好，小时候遭遇到了不幸",
"keyword_note": {
"险衅": "凶险祸患（这里指命运不好）",
"夙": "早时，这里指年幼的时候",
"闵": "通'悯'，指可忧患的事",
"凶": "不幸，指丧父"
}
}
),
{
"keydata": {
"note": "笔记",
"keyword_note": "关键词翻译",
"translation": "语句翻译"
},
"testdata": {
"additional_inf": ["translation", "note", "keyword_note"],
"fill_blank_test": ["translation"],
"draw_card_test": ["keyword_note"]
},
"is_new_activation": 0
}
)

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

@@ -1,8 +1,5 @@
#!/usr/bin/env python3
import pathlib
import toml
import time
import heurams.services.timer as timer
from heurams.context import config_var
class Electron:
"""电子: 记忆分析元数据及算法"""
@@ -25,18 +22,18 @@ class Electron:
'next_date': 0, # 将要复习的时间戳
'is_activated': 0, # 激活状态
# *NOTE: 此处"时间戳"是以天为单位的整数, 即 UNIX 时间戳除以一天的秒数取整
'last_modify': time.time() # 最后修改时间戳(此处是UNIX时间戳)
'last_modify': timer.get_timestamp() # 最后修改时间戳(此处是UNIX时间戳)
}
self.metadata = defaults
def activate(self):
self.metadata['is_activated'] = 1
self.metadata['last_modify'] = time.time()
self.metadata['last_modify'] = timer.get_timestamp()
def modify(self, var: str, value):
if var in self.metadata:
self.metadata[var] = value
self.metadata['last_modify'] = time.time()
self.metadata['last_modify'] = timer.get_timestamp()
else:
print(f"警告: '{var}' 非已知元数据字段")
@@ -81,7 +78,7 @@ class Electron:
self.metadata['last_date'] = timer.get_daystamp()
self.metadata['next_date'] = timer.get_daystamp() + self.metadata['interval']
self.metadata['last_modify'] = time.time()
self.metadata['last_modify'] = timer.get_timestamp()
def __str__(self):
return (
@@ -113,7 +110,7 @@ class Electron:
if key == "content":
raise AttributeError("content 应为只读")
self.metadata[key] = value
self.metadata['last_modify'] = time.time()
self.metadata['last_modify'] = timer.get_timestamp()
def __iter__(self):
yield from self.metadata.keys()

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

@@ -0,0 +1,27 @@
class Nucleon:
"""核子: 材料元数据"""
def __init__(self, content: str, data: dict):
self.metadata = data
self.content = content
def __getitem__(self, key):
if key == "content":
return self.content
if key in self.metadata:
return self.metadata[key]
else:
raise KeyError(f"Key '{key}' not found in metadata.")
def __iter__(self):
yield from self.metadata.keys()
def __len__(self):
return len(self.metadata)
def __hash__(self):
return hash(self.content)
@staticmethod
def placeholder():
return Nucleon("核子对象样例内容", {})

90
src/heurams/kernel/particles/unions.py
View File

@@ -0,0 +1,90 @@
from electron import Electron
from nucleon import Nucleon
import pathlib
import toml
from typing import List
import heurams.services.timer as timer
class NucleonUnion():
"""
替代原有 NucleonFile 类, 支持复杂逻辑
Attributes:
path (Path): 对应于 NucleonUnion 实例的文件路径。
name (str): 核联对象的显示名称，从文件名中派生。
nucleons (list): 内部核子对象的列表。
nucleons_dict (dict): 内部核子对象的字典，以核子内容作为键。
keydata (dict): 核子对象字典键名的翻译。
testdata (dict): 记忆测试项目的元数据。
Parameters:
path (Path): 包含核子数据的文件路径。
"""
def __init__(self, path: pathlib.Path):
self.path = path
self.name = path.name.replace(path.suffix, "")
with open(path, 'r') as f:
all = toml.load(f)
lst = list()
for i in all.keys():
if "attr" in i:
continue
if "data" in i:
continue
lst.append(Nucleon(i, all[i]))
self.keydata = all["keydata"]
self.testdata = all["testdata"]
self.nucleons: List[Nucleon] = lst
self.nucleons_dict = {i.content: i for i in lst}
def __len__(self):
return len(self.nucleons)
def linked_electron_union(self):
if (self.path.parent / '..' / 'electron' / self.path.name).exists():
return ElectronUnion(self.path.parent / '..' / 'electron' / self.path.name)
else:
return 0
def save(self):
with open(self.path, 'w') as f:
tmp = {i.content: i.metadata for i in self.nucleons}
toml.dump(tmp, f)
class ElectronUnion:
"""取代原有 ElectronFile 类, 以支持复杂逻辑"""
def __init__(self, path):
self.path = path
print(path)
self.name = path.name.replace(path.suffix, "")
with open(path, 'r') as f:
all = toml.load(f)
lst = list()
for i in all.keys():
if i != "total":
lst.append(Electron(i, all[i]))
self.total = all.get("total", {"last_date": 0})
self.electrons = lst
self.electrons_dict = {i.content: i for i in lst}
def sync(self):
"""同步 electrons_dict 中新增对到 electrons 中, 仅用于缺省初始化不存在映射时调用"""
self.electrons = self.electrons_dict.values()
def __len__(self):
return len(self.electrons)
def linked_nucleon_union(self):
return NucleonUnion(self.path.parent / '..' / 'nucleon' / self.path.name)
def save(self):
# print(1)
self.total["last_date"] = timer.get_daystamp()
with open(self.path, 'w') as f:
tmp = {i.content: i.metadata for i in self.electrons}
tmp["total"] = self.total
# print(tmp)
toml.dump(tmp, f)

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

@@ -0,0 +1,2 @@
class BasePuzzle:
pass

37
src/heurams/kernel/puzzles/cloze.py Normal file
View File

@@ -0,0 +1,37 @@
from base import BasePuzzle
import random
class ClozePuzzle(BasePuzzle):
"""填空题谜题生成器
Args:
text: 原始字符串(需要 "/" 分割句子, 末尾应有 "/")
min_denominator: 最小概率倒数(如占所有可生成填空数的 1/7 中的 7, 若期望值小于 1, 则取 1)
"""
def __init__(self, text: str, min_denominator: int):
self.text = text
self.min_denominator = min_denominator
self.wording = "填空题 - 尚未刷新谜题"
self.answer = ["填空题 - 尚未刷新谜题"]
def refresh(self): # 刷新谜题
placeholder = "___SLASH___"
tmp_text = self.text.replace("/", placeholder)
words = tmp_text.split(placeholder)
if not words:
return
words = [word for word in words if word]
num_blanks = min(max(1, len(words) // self.min_denominator), len(words))
indices_to_blank = random.sample(range(len(words)), num_blanks)
indices_to_blank.sort()
blanked_words = list(words)
answer = list()
for index in indices_to_blank:
blanked_words[index] = "__" * len(words[index])
answer.append(words[index])
self.answer = answer
self.wording = "".join(blanked_words)
def __str__(self):
return f"{self.wording}\n{str(self.answer)}"

69
src/heurams/kernel/puzzles/selection.py
View File

@@ -0,0 +1,69 @@
from base import BasePuzzle
import random
class MCQPuzzle(BasePuzzle):
"""选择题谜题生成器
Args:
mapping: 正确选项映射 {问题: 答案}
jammer: 干扰项列表
max_riddles_num: 最大生成谜题数 (默认2个)
prefix: 问题前缀
"""
def __init__(
self,
mapping: dict,
jammer: list,
max_riddles_num: int = 2,
prefix: str = ""
):
self.prefix = prefix
self.mapping = mapping
self.jammer = list(set(jammer + list(mapping.values())))
while len(self.jammer) < 4:
self.jammer.append(" ")
self.max_riddles_num = max(1, min(max_riddles_num, 5))
self.wording = "选择题 - 尚未刷新谜题"
self.answer = ["选择题 - 尚未刷新谜题"]
self.options = []
def refresh(self):
"""刷新谜题，根据题目数量生成适当数量的谜题"""
if not self.mapping:
self.wording = "无可用题目"
self.answer = ["无答案"]
self.options = []
return
num_questions = min(self.max_riddles_num, len(self.mapping))
questions = random.sample(list(self.mapping.items()), num_questions)
puzzles = []
answers = []
all_options = []
for question, correct_answer in questions:
options = [correct_answer]
available_jammers = [
j for j in self.jammer if j != correct_answer
]
if len(available_jammers) >= 3:
selected_jammers = random.sample(available_jammers, 3)
else:
selected_jammers = random.choices(available_jammers, k=3)
options.extend(selected_jammers)
random.shuffle(options)
puzzles.append(question)
answers.append(correct_answer)
all_options.append(options)
question_texts = []
for i, puzzle in enumerate(puzzles):
question_texts.append(f"{self.prefix}:\n {i+1}. {puzzle}")
self.wording = question_texts
self.answer = answers
self.options = all_options
def __str__(self):
return f"{self.wording}\n正确答案: {', '.join(self.answer)}"

0
src/heurams/kernel/puzzles/blank.py → src/heurams/kernel/reactor/__init__.py
View File

1
src/heurams/kernel/reactor/apparatus.py Normal file
View File

@@ -0,0 +1 @@
# 单个原子处理器

1
src/heurams/kernel/reactor/atommgr.py Normal file
View File

@@ -0,0 +1 @@
# 原子的生命周期管理器

35
src/heurams/kernel/reactor/glimpse.py Normal file
View File

@@ -0,0 +1,35 @@
# 轻量状态查看器
import heurams.kernel.particles as pt
class Glimpse():
"""轻量级只读, 用于状态指示"""
def __init__(self, nucleon_union: pt.NucleonUnion):
self.name = nucleon_union.name
self.nuc_u = nucleon_union
self.elt_u = self.nuc_u.linked_electron_union()
self.lastest_date = -1
self.next_date = 0x3f3f3f3f
self.avg_efactor = 0
self.total_num = 0
self.activated_num = 0
self.is_initialized = 0
if self.elt_u != 0:
self.is_initialized = 1
self.total_num = len(self.elt_u.electrons)
for i in self.elt_u.electrons:
if i['next_date'] > 0:
self.next_date = min(self.next_date, i['next_date'])
self.lastest_date = max(self.lastest_date, i['last_date'])
if i['is_activated']:
self.avg_efactor += i['efactor']
self.activated_num += 1
if self.next_date == 0x3f3f3f3f:
self.next_date = -1
self.is_initialized = 0
if self.activated_num == 0:
return
self.avg_efactor = round(self.avg_efactor / self.activated_num, 2)
if self.next_date == 0x3f3f3f3f:
self.next_date = -1
return

0
src/heurams/kernel/puzzles/factory.py → src/heurams/kernel/reactor/persistence.py
View File

198
src/heurams/kernel/reactor/r.py Normal file
View File

@@ -0,0 +1,198 @@
#!/usr/bin/env python3
import typing
import particles as pt
import pathlib
import auxiliary as aux
import compositions as comps
import random
#from pprint import pprint as print # debug
class Glimpse():
"""轻量级只读, 用于状态指示"""
def __init__(self, nucleon_union: pt.NucleonUnion):
self.name = nucleon_union.name
self.nuc_u = nucleon_union
self.elt_u = self.nuc_u.linked_electron_union()
self.lastest_date = -1
self.next_date = 0x3f3f3f3f
self.avg_efactor = 0
self.total_num = 0
self.activated_num = 0
self.is_initialized = 0
if self.elt_u != 0:
self.is_initialized = 1
self.total_num = len(self.elt_u.electrons)
for i in self.elt_u.electrons:
if i['next_date'] > 0:
self.next_date = min(self.next_date, i['next_date'])
self.lastest_date = max(self.lastest_date, i['last_date'])
if i['is_activated']:
self.avg_efactor += i['efactor']
self.activated_num += 1
if self.next_date == 0x3f3f3f3f:
self.next_date = -1
self.is_initialized = 0
if self.activated_num == 0:
return
self.avg_efactor = round(self.avg_efactor / self.activated_num, 2)
if self.next_date == 0x3f3f3f3f:
self.next_date = -1
return
class Apparatus():
"""反应器对象, 决策一个原子的不同记忆方式, 并反馈到布局"""
def __init__(self, screen, reactor, atom, is_review = 0):
self.electron: pt.Electron = atom[0]
self.nucleon: pt.Nucleon = atom[1]
self.positron: dict = atom[2]
self.testdata = self.positron["testdata"]
self.procession: typing.List[comps.Composition] = list()
if self.positron["is_new_activation"] == 1:
self.positron["is_new_activation"] = 0
self.procession.append(comps.registry["recognition"](screen, reactor, atom))
return
for i in self.positron["testdata"].keys():
if i == "additional_inf":
continue
if i == "fill_blank_test": # 加深
self.procession.append(comps.registry[i](screen, reactor, atom))
# self.procession.append(comps.registry[i](screen, reactor, atom))
self.procession.append(comps.registry[i](screen, reactor, atom))
# self.procession.reverse()
random.shuffle(self.procession)
if self.positron["is_new_activation"] == 0:
self.procession.append(comps.registry['recognition'](screen, reactor, atom))
if is_review == 1:
self.procession = [self.procession[-2], self.procession[-1]]
def iterator(self):
yield from self.procession
class Reactor():
"""反应堆对象, 处理和分配一次文件记忆流程的资源与策略"""
def __init__(self, nucleon_file: pt.NucleonUnion, electron_file: pt.ElectronUnion, screen, tasked_num):
# 导入原子对象
self.stage = 0
self.nucleon_file = nucleon_file
self.electron_file = electron_file
self.tasked_num = tasked_num
self.atoms_new = list()
self.atoms_review = list()
counter = self.tasked_num
self.screen = screen
self.electron_dict = electron_file.electrons_dict
self.quality_dict = {}
def electron_dict_get_fallback(key) -> pt.Electron:
value = self.electron_dict.get(key)
# 如果值不存在，则设置默认值
if value is None:
value = pt.Electron(key, {}) # 获取默认值
self.electron_dict[key] = value # 将默认值存入字典
electron_file.sync()
return value # 返回获取的值（可能是默认值）
for nucleon in nucleon_file.nucleons:
# atom = (Electron, Nucleon, Positron) 即 (记忆元数据, 内容元数据, 运行时数据)
atom = (electron_dict_get_fallback(nucleon.content), nucleon, {}) # 使用 "Positron" 代称 atom[2]
atom[2]["testdata"] = nucleon_file.testdata
atom[2]["keydata"] = nucleon_file.keydata
if atom[0]["is_activated"] == 0:
if counter > 0:
atom[2]["is_new_activation"] = 1
atom[0]["is_activated"] = 1
self.atoms_new.append(atom)
counter -= 1
else:
atom[2]["is_new_activation"] = 0
if int(atom[0]["next_date"]) <= aux.get_daystamp():
atom[0]["last_date"] = aux.get_daystamp()
self.atoms_review.append(atom)
# 设置运行时
self.index: int
self.procession: list
self.failed: list
self.round_title: str
self.current_atom: typing.Tuple[pt.Electron, pt.Nucleon, dict]
self.round_set = 0
self.current_atom = pt.Atom.placeholder()
#print(self.atoms_new)
def set_round(self, title, procession):
self.round_set = 1
self.round_title = title
self.procession = procession
self.failed = list()
self.index = -1
def set_round_templated(self, stage):
titles = {
1: "复习模式",
2: "新记忆模式",
3: "总复习模式"
}
processions = {
1: self.atoms_review,
2: self.atoms_new,
3: (self.atoms_new + self.atoms_review)
}
self.stage = stage
ret = 1
if stage == 1 and len(processions[1]) == 0:
stage = 2
ret = 2
if stage == 1 and len(processions[2]) == 0:
stage = 3
ret = 3
self.set_round(title=titles[stage], procession=processions[stage])
return ret
def forward(self, step = 1):
"""
返回值规则:
1: 重定向至 failed
-1: 此轮已完成
0: 下一个记忆单元
"""
if self.index + step >= len(self.procession):
if len(self.failed) > 0:
self.procession = self.failed
self.index = -1
self.forward(step)
if "- 额外复习" not in self.round_title:
self.round_title += " - 额外复习"
self.failed = list()
return 1 # 自动重定向到 failed
else:
self.round_set = 0
return -1 # 此轮已完成
self.index += step
self.current_atom = self.procession[self.index]
self.current_appar = Apparatus(self.screen, self, self.current_atom, (self.stage == 1)).iterator()
return 0
def save(self):
self._deploy_report()
print("Progress saved")
# self.nucleon_file.save()
if self.electron_file.total["last_date"] < aux.get_daystamp():
self.electron_file.save()
def _deploy_report(self):
"部署所有 _report"
for e, q in self.quality_dict.items():
if q == -1:
e.revisor(5, True)
continue
e.revisor(q)
def report(self, atom, quality):
"向反应器和最低质量记录汇报"
if atom in self.atoms_new:
self.quality_dict[atom[0]] = -1
print(self.quality_dict)
return
self.quality_dict[atom[0]] = min(quality, self.quality_dict.get(atom[0], 5))
if quality <= 3:
self.failed.append(atom)
print(self.quality_dict)

1
src/heurams/kernel/reactor/reactor.py Normal file
View File

@@ -0,0 +1 @@
# 核心流程状态机

1
src/heurams/kernel/reactor/rounds.py Normal file
View File

@@ -0,0 +1 @@
# 轮次编排