open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2026-02-05 14:05:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

add resource edge upload

Browse Source
This commit is contained in:
Xuwznln
2025-06-16 11:51:02 +08:00
parent a303bd7c5b
commit 57cb120c8c
7 changed files with 73 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
unilabos/app/backend.py
View File

@@ -8,6 +8,7 @@ def start_backend(
backend: str,
devices_config: dict = {},
resources_config: list = [],
resources_edge_config: list = [],
graph=None,
controllers_config: dict = {},
bridges=[],
@@ -31,7 +32,7 @@ def start_backend(
backend_thread = threading.Thread(
target=main if not without_host else slave,
args=(devices_config, resources_config, graph, controllers_config, bridges, visual, resources_mesh_config),
args=(devices_config, resources_config, resources_edge_config, graph, controllers_config, bridges, visual, resources_mesh_config),
name="backend_thread",
daemon=True,
)

16
unilabos/app/main.py
View File

@@ -10,7 +10,7 @@ from copy import deepcopy
import yaml
from unilabos.resources.graphio import tree_to_list
from unilabos.resources.graphio import tree_to_list, modify_to_backend_format
# 首先添加项目根目录到路径
current_dir = os.path.dirname(os.path.abspath(__file__))
@@ -136,15 +136,16 @@ def main():
# 注册表
build_registry(args_dict["registry_path"])
resource_edge_info = []
devices_and_resources = None
if args_dict["graph"] is not None:
import unilabos.resources.graphio as graph_res
graph_res.physical_setup_graph = (
read_node_link_json(args_dict["graph"])
if args_dict["graph"].endswith(".json")
else read_graphml(args_dict["graph"])
)
if args_dict["graph"].endswith(".json"):
graph, data = read_node_link_json(args_dict["graph"])
else:
graph, data = read_graphml(args_dict["graph"])
graph_res.physical_setup_graph = graph
resource_edge_info = modify_to_backend_format(data["links"])
devices_and_resources = dict_from_graph(graph_res.physical_setup_graph)
# args_dict["resources_config"] = initialize_resources(list(deepcopy(devices_and_resources).values()))
args_dict["resources_config"] = list(devices_and_resources.values())
@@ -185,6 +186,7 @@ def main():
signal.signal(signal.SIGTERM, _exit)
mqtt_client.start()
args_dict["resources_mesh_config"] = {}
args_dict["resources_edge_config"] = resource_edge_info
# web visiualize 2D
if args_dict["visual"] != "disable":
enable_rviz = args_dict["visual"] == "rviz"

20
unilabos/app/web/client.py
View File

@@ -30,7 +30,25 @@ class HTTPClient:
self.auth = MQConfig.lab_id
info(f"HTTPClient 初始化完成: remote_addr={self.remote_addr}")
def resource_add(self, resources: List[Dict[str, Any]], database_process_later:bool) -> requests.Response:
def resource_edge_add(self, resources: List[Dict[str, Any]], database_process_later: bool) -> requests.Response:
"""
添加资源
Args:
resources: 要添加的资源列表
database_process_later: 后台处理资源
Returns:
Response: API响应对象
"""
response = requests.post(
f"{self.remote_addr}/lab/resource/edge/batch_create/?database_process_later={1 if database_process_later else 0}",
json=resources,
headers={"Authorization": f"lab {self.auth}"},
timeout=5,
)
return response
def resource_add(self, resources: List[Dict[str, Any]], database_process_later: bool) -> requests.Response:
"""
添加资源

1
unilabos/app/web/pages.py
View File

@@ -16,7 +16,6 @@ from jinja2 import Environment, FileSystemLoader
from unilabos.config.config import BasicConfig
from unilabos.registry.registry import lab_registry
from unilabos.app.mq import mqtt_client
from unilabos.ros.msgs.message_converter import msg_converter_manager
from unilabos.utils.log import error
from unilabos.utils.type_check import TypeEncoder

29
unilabos/resources/graphio.py
View File

@@ -1,7 +1,7 @@
import importlib
import inspect
import json
from typing import Union
from typing import Union, Any
import numpy as np
import networkx as nx
from unilabos_msgs.msg import Resource
@@ -84,6 +84,8 @@ def canonicalize_links_ports(data: dict) -> dict:
# 第一遍处理：将字符串类型的port转换为字典格式
for link in data.get("links", []):
port = link.get("port")
if link["type"] == "physical":
link["type"] = "fluid"
if isinstance(port, int):
port = str(port)
if isinstance(port, str):
@@ -157,7 +159,28 @@ def read_node_link_json(json_file):
physical_setup_graph = nx.node_link_graph(data, multigraph=False) # edges="links" 3.6 warning
handle_communications(physical_setup_graph)
return physical_setup_graph
return physical_setup_graph, data
def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]:
for edge in data:
port = edge.pop("port", {})
source = edge["source"]
target = edge["target"]
if source in port:
edge["sourceHandle"] = port[source]
elif "source_port" in edge:
edge["sourceHandle"] = edge.pop("source_port")
if target in port:
edge["targetHandle"] = port[target]
elif "target_port" in edge:
edge["targetHandle"] = edge.pop("target_port")
if "id" not in edge:
edge["id"] = f"link_generated_{source}_{target}"
for key in ["source_port", "target_port"]:
if key in edge:
edge.pop(key)
return data
def read_graphml(graphml_file):
@@ -182,7 +205,7 @@ def read_graphml(graphml_file):
physical_setup_graph = nx.node_link_graph(data, edges="links", multigraph=False) # edges="links" 3.6 warning
handle_communications(physical_setup_graph)
return physical_setup_graph
return physical_setup_graph, data
def dict_from_graph(graph: nx.Graph) -> dict:

3
unilabos/ros/main_slave_run.py
View File

@@ -45,6 +45,7 @@ def exit() -> None:
def main(
devices_config: Dict[str, Any] = {},
resources_config: list=[],
resources_edge_config: list=[],
graph: Optional[Dict[str, Any]] = None,
controllers_config: Dict[str, Any] = {},
bridges: List[Any] = [],
@@ -62,6 +63,7 @@ def main(
"host_node",
devices_config,
resources_config,
resources_edge_config,
graph,
controllers_config,
bridges,
@@ -97,6 +99,7 @@ def main(
def slave(
devices_config: Dict[str, Any] = {},
resources_config=[],
resources_edge_config=[],
graph: Optional[Dict[str, Any]] = None,
controllers_config: Dict[str, Any] = {},
bridges: List[Any] = [],

17
unilabos/ros/nodes/presets/host_node.py
View File

@@ -58,6 +58,7 @@ class HostNode(BaseROS2DeviceNode):
device_id: str,
devices_config: Dict[str, Any],
resources_config: list,
resources_edge_config: list[dict],
physical_setup_graph: Optional[Dict[str, Any]] = None,
controllers_config: Optional[Dict[str, Any]] = None,
bridges: Optional[List[Any]] = None,
@@ -96,6 +97,7 @@ class HostNode(BaseROS2DeviceNode):
self.server_latest_timestamp = 0.0 #
self.devices_config = devices_config
self.resources_config = resources_config
self.resources_edge_config = resources_edge_config
self.physical_setup_graph = physical_setup_graph
if controllers_config is None:
controllers_config = {}
@@ -203,12 +205,19 @@ class HostNode(BaseROS2DeviceNode):
try:
for bridge in self.bridges:
if hasattr(bridge, "resource_add"):
from unilabos.app.web.client import HTTPClient
client: HTTPClient = bridge
resource_start_time = time.time()
resource_add_res = bridge.resource_add(add_schema(resource_with_parent_name), True)
resource_add_res = client.resource_add(add_schema(resource_with_parent_name), True)
resource_end_time = time.time()
self.lab_logger().info(
f"[Host Node-Resource] 物料上传 {round(resource_end_time - resource_start_time, 5) * 1000} ms"
)
resource_add_res = client.resource_edge_add(self.resources_edge_config, True)
resource_edge_end_time = time.time()
self.lab_logger().info(
f"[Host Node-Resource] 物料关系上传 {round(resource_edge_end_time - resource_end_time, 5) * 1000} ms"
)
except Exception as ex:
self.lab_logger().error("[Host Node-Resource] 添加物料出错！")
self.lab_logger().error(traceback.format_exc())
@@ -757,8 +766,10 @@ class HostNode(BaseROS2DeviceNode):
self.lab_logger().info(f"[Host Node-Resource] Add request received: {len(resources)} resources")
success = False
if len(self.bridges) > 0:
r = self.bridges[-1].resource_add(add_schema(resources))
if len(self.bridges) > 0: # 边的提交待定
from unilabos.app.web.client import HTTPClient
client: HTTPClient = self.bridges[-1]
r = client.resource_add(add_schema(resources), False)
success = bool(r)
response.success = success