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

Merge branch 'device-registry-port' of github.com:KCFeng425/Uni-Lab-OS into device-registry-port

Browse Source
This commit is contained in:
KCFeng425
2025-06-17 16:57:28 +08:00
parent f6f9244ff1 ac88c59b50
commit cbe04a0b09
10 changed files with 70 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
unilabos/app/main.py
View File

@@ -22,6 +22,21 @@ from unilabos.config.config import load_config, BasicConfig, _update_config_from
from unilabos.utils.banner_print import print_status, print_unilab_banner
def load_config_from_file(config_path):
if config_path is None:
config_path = os.environ.get("UNILABOS.BASICCONFIG.CONFIG_PATH", None)
if config_path:
if not os.path.exists(config_path):
print_status(f"配置文件 {config_path} 不存在", "error")
elif not config_path.endswith(".py"):
print_status(f"配置文件 {config_path} 不是Python文件，必须以.py结尾", "error")
else:
load_config(config_path)
else:
print_status(f"启动 UniLab-OS时，配置文件参数未正确传入 --config '{config_path}' 尝试本地配置...", "warning")
load_config(config_path)
def parse_args():
"""解析命令行参数"""
parser = argparse.ArgumentParser(description="Start Uni-Lab Edge server.")
@@ -58,6 +73,11 @@ def parse_args():
action="store_true",
help="Slave模式下跳过等待host服务",
)
parser.add_argument(
"--upload_registry",
action="store_true",
help="启动unilab时同时报送注册表信息",
)
parser.add_argument(
"--config",
type=str,
@@ -97,22 +117,12 @@ def main():
# 加载配置文件，优先加载config，然后从env读取
config_path = args_dict.get("config")
if config_path is None:
config_path = os.environ.get("UNILABOS.BASICCONFIG.CONFIG_PATH", None)
if config_path:
if not os.path.exists(config_path):
print_status(f"配置文件 {config_path} 不存在", "error")
elif not config_path.endswith(".py"):
print_status(f"配置文件 {config_path} 不是Python文件，必须以.py结尾", "error")
else:
load_config(config_path)
else:
print_status(f"启动 UniLab-OS时，配置文件参数未正确传入 --config '{config_path}' 尝试本地配置...", "warning")
load_config(config_path)
load_config_from_file(config_path)
# 设置BasicConfig参数
BasicConfig.is_host_mode = not args_dict.get("without_host", False)
BasicConfig.slave_no_host = args_dict.get("slave_no_host", False)
BasicConfig.upload_registry = args_dict.get("upload_registry", False)
machine_name = os.popen("hostname").read().strip()
machine_name = "".join([c if c.isalnum() or c == "_" else "_" for c in machine_name])
BasicConfig.machine_name = machine_name

5
unilabos/app/mq.py
View File

@@ -172,13 +172,14 @@ class MQTTClient:
jobdata = {"job_id": job_id, "data": feedback_data, "status": status, "return_info": return_info}
self.client.publish(f"labs/{MQConfig.lab_id}/job/list/", json.dumps(jobdata), qos=2)
def publish_registry(self, device_id: str, device_info: dict):
def publish_registry(self, device_id: str, device_info: dict, print_debug: bool = True):
if self.mqtt_disable:
return
address = f"labs/{MQConfig.lab_id}/registry/"
registry_data = json.dumps({device_id: device_info}, ensure_ascii=False, cls=TypeEncoder)
self.client.publish(address, registry_data, qos=2)
logger.debug(f"Registry data published: address: {address}, {registry_data}")
if print_debug:
logger.debug(f"Registry data published: address: {address}, {registry_data}")
def publish_actions(self, action_id: str, action_info: dict):
if self.mqtt_disable:

10
unilabos/app/register.py
View File

@@ -15,15 +15,15 @@ def register_devices_and_resources(mqtt_client, lab_registry):
# 注册设备信息
for device_info in lab_registry.obtain_registry_device_info():
mqtt_client.publish_registry(device_info["id"], device_info)
mqtt_client.publish_registry(device_info["id"], device_info, False)
logger.debug(f"[UniLab Register] 注册设备: {device_info['id']}")
# 注册资源信息
for resource_info in lab_registry.obtain_registry_resource_info():
mqtt_client.publish_registry(resource_info["id"], resource_info)
mqtt_client.publish_registry(resource_info["id"], resource_info, False)
logger.debug(f"[UniLab Register] 注册资源: {resource_info['id']}")
time.sleep(20)
time.sleep(10)
logger.info("[UniLab Register] 设备和资源注册完成.")
@@ -53,6 +53,10 @@ def main():
load_config_from_file(args.config)
from unilabos.app.mq import mqtt_client
# 连接mqtt
mqtt_client.start()
from unilabos.registry.registry import lab_registry
# 注册设备和资源

3
unilabos/config/config.py
View File

@@ -10,8 +10,9 @@ from unilabos.utils import logger
class BasicConfig:
ENV = "pro" # 'test'
config_path = ""
is_host_mode = True # 从registry.py移动过来
is_host_mode = True
slave_no_host = False # 是否跳过rclient.wait_for_service()
upload_registry = False
machine_name = "undefined"
vis_2d_enable = False

1
unilabos/registry/devices/virtual_device.yaml
View File

@@ -218,6 +218,7 @@ virtual_multiway_valve:
- handler_key: transferpump
label: transferpump
data_type: fluid
side: NORTH
io_type: target
data_source: handle
data_key: fluid_in

3
unilabos/resources/graphio.py
View File

@@ -175,8 +175,7 @@ def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]
edge["targetHandle"] = port[target]
elif "target_port" in edge:
edge["targetHandle"] = edge.pop("target_port")
if "id" not in edge:
edge["id"] = f"reactflow__edge-{edge['sourceHandle']}-{edge['targetHandle']}"
edge["id"] = f"reactflow__edge-{source}-{edge['sourceHandle']}-{target}-{edge['targetHandle']}"
for key in ["source_port", "target_port"]:
if key in edge:
edge.pop(key)

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

@@ -22,6 +22,7 @@ from unilabos_msgs.srv import (
) # type: ignore
from unique_identifier_msgs.msg import UUID
from unilabos.config.config import BasicConfig
from unilabos.registry.registry import lab_registry
from unilabos.resources.graphio import initialize_resource
from unilabos.resources.registry import add_schema
@@ -146,13 +147,15 @@ class HostNode(BaseROS2DeviceNode):
self.device_status = {} # 用来存储设备状态
self.device_status_timestamps = {} # 用来存储设备状态最后更新时间
if BasicConfig.upload_registry:
from unilabos.app.mq import mqtt_client
from unilabos.app.mq import mqtt_client
for device_info in lab_registry.obtain_registry_device_info():
mqtt_client.publish_registry(device_info["id"], device_info)
for resource_info in lab_registry.obtain_registry_resource_info():
mqtt_client.publish_registry(resource_info["id"], resource_info)
for device_info in lab_registry.obtain_registry_device_info():
mqtt_client.publish_registry(device_info["id"], device_info)
for resource_info in lab_registry.obtain_registry_resource_info():
mqtt_client.publish_registry(resource_info["id"], resource_info)
else:
self.lab_logger().warning("本次启动注册表不报送云端，如果您需要联网调试，请使用unilab-register命令进行单独报送，或者在启动命令增加--upload_registry")
time.sleep(1) # 等待MQTT连接稳定
# 首次发现网络中的设备
self._discover_devices()
@@ -195,18 +198,18 @@ class HostNode(BaseROS2DeviceNode):
resource_ids_to_instance = {i["id"]: i for i in resources_config}
resource_name_to_with_parent_name = {}
for res in resources_config:
if res.get("parent") and res.get("type") == "device" and res.get("class"):
parent_id = res.get("parent")
parent_res = resource_ids_to_instance[parent_id]
if parent_res.get("type") == "device" and parent_res.get("class"):
resource_with_parent_name.append(copy.deepcopy(res))
resource_name_to_with_parent_name[resource_with_parent_name[-1]["id"]] = f"{parent_res['id']}/{res['id']}"
resource_with_parent_name[-1]["id"] = f"{parent_res['id']}/{res['id']}"
continue
# if res.get("parent") and res.get("type") == "device" and res.get("class"):
# parent_id = res.get("parent")
# parent_res = resource_ids_to_instance[parent_id]
# if parent_res.get("type") == "device" and parent_res.get("class"):
# resource_with_parent_name.append(copy.deepcopy(res))
# resource_name_to_with_parent_name[resource_with_parent_name[-1]["id"]] = f"{parent_res['id']}/{res['id']}"
# resource_with_parent_name[-1]["id"] = f"{parent_res['id']}/{res['id']}"
# continue
resource_with_parent_name.append(copy.deepcopy(res))
for edge in self.resources_edge_config:
edge["source"] = resource_name_to_with_parent_name.get(edge.get("source"), edge.get("source"))
edge["target"] = resource_name_to_with_parent_name.get(edge.get("target"), edge.get("target"))
# for edge in self.resources_edge_config:
# edge["source"] = resource_name_to_with_parent_name.get(edge.get("source"), edge.get("source"))
# edge["target"] = resource_name_to_with_parent_name.get(edge.get("target"), edge.get("target"))
try:
for bridge in self.bridges:
if hasattr(bridge, "resource_add"):

3
unilabos/ros/nodes/presets/protocol_node.py
View File

@@ -110,7 +110,8 @@ class ROS2ProtocolNode(BaseROS2DeviceNode):
def initialize_device(self, device_id, device_config):
"""初始化设备并创建相应的动作客户端"""
device_id_abs = f"{self.device_id}/{device_id}"
# device_id_abs = f"{self.device_id}/{device_id}"
device_id_abs = f"{device_id}"
self.lab_logger().info(f"初始化子设备: {device_id_abs}")
d = self.sub_devices[device_id] = initialize_device_from_dict(device_id_abs, device_config)