mirror of
https://github.com/deepmodeling/Uni-Lab-OS
synced 2026-04-23 10:55:27 +00:00
ccbf5378dd
update aksk desc print res query logs Fix skills exec error with action type Update Skills Update Skills addr Change uni-lab. to leap-lab. Support unit in pylabrobot Support async func. change to leap-lab backend. Support feedback interval. Reduce cocurrent lags. fix create_resource_with_slot update unilabos_formulation & batch-submit-exp scale multi exec thread up to 48 update handle creation api fit cocurrent gap add running status debounce allow non @topic_config support update skill add placeholder keys always free 提交实验技能 disable samples correct sample demo ret value 新增试剂reagent update registry 新增manual_confirm add workstation creation skill add virtual_sample_demo 样品追踪测试设备 add external devices param fix registry upload missing type fast registry load minor fix on skill & registry stripe ros2 schema desc add create-device-skill new registry system backwards to yaml remove not exist resource new registry sys exp. support with add device correct raise create resource error ret info fix revert ret info fix fix prcxi check add create_resource schema re signal host ready event add websocket connection timeout and improve reconnection logic add open_timeout parameter to websocket connection add TimeoutError and InvalidStatus exception handling implement exponential backoff for reconnection attempts simplify reconnection logic flow
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
import json
|
||
|
||
from unilabos.utils.type_check import TypeEncoder, json_default
|
||
|
||
try:
|
||
import orjson
|
||
|
||
def fast_dumps(obj, **kwargs) -> bytes:
|
||
"""JSON 序列化为 bytes，优先使用 orjson。"""
|
||
return orjson.dumps(obj, option=orjson.OPT_NON_STR_KEYS, default=json_default)
|
||
|
||
def fast_dumps_pretty(obj, **kwargs) -> bytes:
|
||
"""JSON 序列化为 bytes（带缩进），优先使用 orjson。"""
|
||
return orjson.dumps(
|
||
obj,
|
||
option=orjson.OPT_NON_STR_KEYS | orjson.OPT_INDENT_2,
|
||
default=json_default,
|
||
)
|
||
|
||
def fast_loads(data) -> dict:
|
||
"""JSON 反序列化，优先使用 orjson。接受 str / bytes。"""
|
||
return orjson.loads(data)
|
||
|
||
def fast_dumps_str(obj, **kwargs) -> str:
|
||
"""JSON 序列化为 str，优先使用 orjson。用于需要 str 而非 bytes 的场景（如 ROS msg）。"""
|
||
return orjson.dumps(obj, option=orjson.OPT_NON_STR_KEYS, default=json_default).decode("utf-8")
|
||
|
||
def normalize_json(info: dict) -> dict:
|
||
"""经 JSON 序列化/反序列化一轮来清理非标准类型。"""
|
||
return orjson.loads(orjson.dumps(info, default=json_default))
|
||
|
||
except ImportError:
|
||
|
||
def fast_dumps(obj, **kwargs) -> bytes: # type: ignore[misc]
|
||
return json.dumps(obj, ensure_ascii=False, cls=TypeEncoder).encode("utf-8")
|
||
|
||
def fast_dumps_pretty(obj, **kwargs) -> bytes: # type: ignore[misc]
|
||
return json.dumps(obj, indent=2, ensure_ascii=False, cls=TypeEncoder).encode("utf-8")
|
||
|
||
def fast_loads(data) -> dict: # type: ignore[misc]
|
||
if isinstance(data, bytes):
|
||
data = data.decode("utf-8")
|
||
return json.loads(data)
|
||
|
||
def fast_dumps_str(obj, **kwargs) -> str: # type: ignore[misc]
|
||
return json.dumps(obj, ensure_ascii=False, cls=TypeEncoder)
|
||
|
||
def normalize_json(info: dict) -> dict: # type: ignore[misc]
|
||
return json.loads(json.dumps(info, ensure_ascii=False, cls=TypeEncoder))
|
||
|
||
|
||
# 辅助函数：将UUID数组转换为字符串
|
||
def uuid_to_str(uuid_array) -> str:
|
||
"""将UUID字节数组转换为十六进制字符串"""
|
||
return "".join(format(byte, "02x") for byte in uuid_array)
|