feat: 进一步改进
This commit is contained in:
@@ -2,55 +2,55 @@ from heurams.services.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class Nucleon:
|
||||
"""原子核: 材料元数据"""
|
||||
"""原子核: 带有运行时隔离的半只读材料元数据容器
|
||||
"""
|
||||
|
||||
def __init__(self, ident: str, payload: dict, metadata: dict = {}):
|
||||
"""初始化原子核 (记忆内容)
|
||||
|
||||
Args:
|
||||
ident: 唯一标识符
|
||||
payload: 记忆内容信息
|
||||
metadata: 可选元数据信息
|
||||
"""
|
||||
logger.debug(
|
||||
"创建 Nucleon 实例, ident: '%s', payload keys: %s, metadata keys: %s",
|
||||
ident,
|
||||
list(payload.keys()) if payload else [],
|
||||
list(metadata.keys()) if metadata else [],
|
||||
)
|
||||
self.metadata = metadata
|
||||
self.payload = payload
|
||||
def __init__(self, ident, payload, common):
|
||||
self.ident = ident
|
||||
logger.debug("Nucleon 初始化完成")
|
||||
self.payload = payload
|
||||
self.common = common
|
||||
self.rtlayer = dict() # 运行时层
|
||||
|
||||
def __getitem__(self, key):
|
||||
logger.debug("Nucleon.__getitem__: key='%s'", key)
|
||||
if key == "ident":
|
||||
logger.debug("返回 ident: '%s'", self.ident)
|
||||
return self.ident
|
||||
if key in self.payload:
|
||||
value = self.payload[key]
|
||||
logger.debug(
|
||||
"返回 payload['%s'], value type: %s", key, type(value).__name__
|
||||
)
|
||||
return value
|
||||
merged = self.rtlayer | self.payload | self.common
|
||||
return merged[key]
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
if key == "ident":
|
||||
raise AttributeError("ident 应为只读")
|
||||
else:
|
||||
logger.error("键 '%s' 未在 payload 中找到", key)
|
||||
raise KeyError(f"Key '{key}' not found in payload.")
|
||||
self.rtlayer[key] = value
|
||||
|
||||
def __delitem__(self, key):
|
||||
raise AttributeError("Nucleon 包含的数据被设计为无法删除")
|
||||
|
||||
def __iter__(self):
|
||||
yield from self.payload.keys()
|
||||
merged = self.rtlayer | self.payload | self.common
|
||||
return iter(merged)
|
||||
|
||||
def __contains__(self, key):
|
||||
return key in (self.rtlayer | self.payload | self.common)
|
||||
|
||||
def get(self, key, default=None):
|
||||
if key in self:
|
||||
return self[key]
|
||||
return default
|
||||
|
||||
def __len__(self):
|
||||
return len(self.payload)
|
||||
return len(self.rtlayer | self.payload | self.common)
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.ident)
|
||||
def __repr__(self):
|
||||
return f"""RUNTIME:{repr(self.rtlayer)}
|
||||
PAYLOAD:{repr(self.payload)}
|
||||
COMMON:{repr(self.common)}"""
|
||||
|
||||
@staticmethod
|
||||
def placeholder():
|
||||
"""生成一个占位原子核"""
|
||||
logger.debug("创建 Nucleon 占位符")
|
||||
return Nucleon("核子对象样例内容", {})
|
||||
def create_on_nucleonic_data(nucleonic_data: tuple):
|
||||
_data = nucleonic_data
|
||||
payload = _data[1][0]
|
||||
common = _data[1][1]
|
||||
ident = _data[0] #TODO:实现eval
|
||||
return Nucleon(ident, payload, common)
|
||||
Reference in New Issue
Block a user