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

feat: 代码格式化, 改进仪表盘, 新增多CSS支持

Browse Source
This commit is contained in:
Wang Zhiyu
2026-04-20 16:30:04 +08:00
parent 845a505ca1
commit 65fbdec0a9
43 changed files with 551 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/heurams/kernel/algorithms/base.py
View File

@@ -8,7 +8,8 @@ logger = get_logger(__name__)
class BaseAlgorithm:
algo_name = "BaseAlgorithm"
desc = '算法基类'
desc = "算法基类"
class AlgodataDict(TypedDict):
real_rept: int
rept: int

11
src/heurams/kernel/algorithms/nsp0.py
View File

@@ -10,7 +10,8 @@ logger = get_logger(__name__)
class NSP0Algorithm(BaseAlgorithm):
algo_name = "NSP-0"
desc = '快速筛选用特殊调度器'
desc = "快速筛选用特殊调度器"
class AlgodataDict(TypedDict):
real_rept: int
rept: int
@@ -23,7 +24,7 @@ class NSP0Algorithm(BaseAlgorithm):
defaults = {
"real_rept": 0,
'important': 0,
"important": 0,
"rept": 0,
"interval": 0,
"last_date": 0,
@@ -52,8 +53,10 @@ class NSP0Algorithm(BaseAlgorithm):
if feedback == -1:
logger.debug("feedback 为 -1, 跳过更新")
return
algodata[cls.algo_name]["interval"] = (1 if feedback <= 3 else float('inf'))
algodata[cls.algo_name]["important"] = (1 if feedback <= 3 else algodata[cls.algo_name]["important"])
algodata[cls.algo_name]["interval"] = 1 if feedback <= 3 else float("inf")
algodata[cls.algo_name]["important"] = (
1 if feedback <= 3 else algodata[cls.algo_name]["important"]
)
algodata[cls.algo_name]["last_date"] = timer.get_daystamp()
algodata[cls.algo_name]["next_date"] = (
timer.get_daystamp() + algodata[cls.algo_name]["interval"]

2
src/heurams/kernel/algorithms/sm15m.py
View File

@@ -27,7 +27,7 @@ from heurams.kernel.algorithms.sm15m_calc import (
# 全局状态文件路径
_GLOBAL_STATE_FILE = os.path.expanduser(
pathlib.Path(config_var.get()['global']["paths"]["data"])
pathlib.Path(config_var.get()["global"]["paths"]["data"])
/ "global"
/ "sm15m_global_state.json"
)

3
src/heurams/kernel/algorithms/sm2.py
View File

@@ -10,7 +10,8 @@ logger = get_logger(__name__)
class SM2Algorithm(BaseAlgorithm):
algo_name = "SM-2"
desc = '经典间隔重复算法'
desc = "经典间隔重复算法"
class AlgodataDict(TypedDict):
efactor: float
real_rept: int

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

@@ -32,7 +32,7 @@ class Atom:
default_runtime = {
"locked": False,
"min_rate": float('inf'),
"min_rate": float("inf"),
"new_activation": False,
}

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

@@ -24,6 +24,7 @@ class Electron:
algo_name = "SM-2"
self.algodata = algodata
self.ident = ident
self.algoname = algo_name
self.algo: algolib.BaseAlgorithm = algorithms[algo_name]
if not self.algo.check_integrity(self.algodata):
@@ -53,10 +54,10 @@ class Electron:
result = self.algo.is_due(self.algodata)
return result and self.is_activated()
def rept(self, real_rept = False):
def rept(self, real_rept=False):
if real_rept:
return self.algodata[self.algo.algo_name]['real_rept']
return self.algodata[self.algo.algo_name]['rept']
return self.algodata[self.algo.algo_name]["real_rept"]
return self.algodata[self.algo.algo_name]["rept"]
def is_activated(self):
result = self.algodata[self.algo.algo_name]["is_activated"]
@@ -112,7 +113,7 @@ class Electron:
return len(self.algodata[self.algo.algo_name])
@staticmethod
def create_on_electonic_data(electronic_data: tuple, algo_name: str = ""):
def from_data(electronic_data: tuple, algo_name: str = ""):
_data = electronic_data
ident = _data[0]
algodata = _data[1]

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

@@ -15,26 +15,26 @@ class Nucleon:
self.ident = ident
try:
data_safe = deepcopy((payload | common))
data_puz = deepcopy(data_safe['puzzles'])
data_safe['puzzles'] = {}
data_puz = deepcopy(data_safe["puzzles"])
data_safe["puzzles"] = {}
env = {
"payload": data_safe,
"default": config_var.get()['interface']["puzzles"],
"default": config_var.get()["interface"]["puzzles"],
"nucleon": data_safe,
}
self.evalizer = Evalizer(environment=env)
data_safe = self.evalizer(deepcopy(data_safe))
env = {
"payload": data_safe,
"default": config_var.get()['interface']["puzzles"],
"default": config_var.get()["interface"]["puzzles"],
"nucleon": data_safe,
}
self.evalizer = Evalizer(environment=env)
data_puz = self.evalizer(deepcopy(data_puz))
data_safe['puzzles'] = data_puz # type: ignore
self.data: dict = data_safe # type: ignore
data_safe["puzzles"] = data_puz # type: ignore
self.data: dict = data_safe # type: ignore
except Exception:
self.data = (payload | common)
self.data = payload | common
def __getitem__(self, key):
if isinstance(key, str):
@@ -71,7 +71,7 @@ class Nucleon:
return s
@staticmethod
def create_on_nucleonic_data(nucleonic_data: tuple):
def from_data(nucleonic_data: tuple):
_data = nucleonic_data
payload = _data[1][0]
common = _data[1][1]

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

@@ -75,7 +75,7 @@ class Fission(Machine):
self.current_puzzle_inf = self.puzzles_inf[0]
for i in range(len(self.puzzles_inf)):
self.min_ratings.append(float('inf'))
self.min_ratings.append(float("inf"))
Machine.__init__(
self,

37
src/heurams/kernel/repolib/repo.py
View File

@@ -13,10 +13,14 @@ from heurams.kernel.auxiliary.lict import Lict
class RepoManifest(TypedDict):
title: str
author: str
package: str
desc: str
class Repo:
"""只维护仓库本身
上层 API 请访问此对象下粒子对象列表"""
file_mapping = {
"schedule": "schedule.toml",
"payload": "payload.toml",
@@ -58,14 +62,15 @@ class Repo:
"algodata": self.algodata,
"source": self.source,
}
self.generate_particles_data()
def generate_particles_data(self):
self._generate_particles_data()
def _generate_particles_data(self):
"""生成上层的粒子对象组和 API 交互, 会在 init 后自动调用"""
self.nucleonic_data_lict = Lict(
initlist=list(map(self._nucleonic_proc, self.payload))
)
self.orbitic_data = self.schedule
self.data_length = len(self.nucleonic_data_lict)
self.ident_index = self.nucleonic_data_lict.keys()
for i in self.ident_index:
self.algodata.append_new((i, {}))
@@ -76,13 +81,6 @@ class Repo:
common = self.typedef["common"]
return (ident, (unit[1], common))
@staticmethod
def _merge(value):
def inner(x):
return (x, value)
return inner
def __len__(self):
return len(self.payload)
@@ -95,6 +93,7 @@ class Repo:
def persist_to_repodir(
self, save_list: list | None = None, source: Path | None = None
):
"""保存单元集数据到目录"""
if save_list == None:
save_list = self.default_save_list
if self.source != None and source == None:
@@ -116,11 +115,13 @@ class Repo:
else:
raise ValueError(f"不支持的文件类型: {filename}")
def export_to_single_dict(self):
def export_to_dict(self):
"""导出至单个字典"""
return self.database
@classmethod
def create_new_repo(cls, source=None):
"""创建新的空单元集"""
default_database = {
"schedule": {},
"payload": Lict([]),
@@ -132,7 +133,8 @@ class Repo:
return Repo(**default_database)
@classmethod
def create_from_repodir(cls, source: Path):
def from_repodir(cls, source: Path):
"""从目录创建单元集"""
database = {}
for keyname, filename in cls.file_mapping.items():
with open(source / filename, "r") as f:
@@ -153,21 +155,24 @@ class Repo:
return Repo(**database)
@classmethod
def create_from_single_dict(cls, dictdata, source: Path | None = None):
def from_dict(cls, dictdata, source: Path | None = None):
"""从单一字典创建单元集"""
database = dictdata
database["source"] = source
return Repo(**database)
@classmethod
def check_repodir(cls, source: Path):
"""检测单元集目录合法性"""
try:
cls.create_from_repodir(source)
return 1
cls.from_repodir(source)
return True
except:
return 0
return False
@classmethod
def probe_valid_repos_in_dir(cls, folder: Path):
"""返回一个合法的子目录 Path() 列表"""
lst = list()
for i in folder.iterdir():
if i.is_dir():