whut/AutoCalib-Workshop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 已添加win_ubuntu_bridge

Browse Source
This commit is contained in:
li-shihao-code
2026-03-06 17:16:50 +08:00
parent 1ea480eccd
commit 18d6500d04
21 changed files with 1673 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agv_calib_eu/scripts/agv_calib_chassis_pb2.py
+52
View File
@@ -0,0 +1,52 @@
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: agv_calib_chassis.proto
# Protobuf Python Version: 6.31.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
_runtime_version.Domain.PUBLIC,
6,
31,
1,
'',
'agv_calib_chassis.proto'
)
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x17\x61gv_calib_chassis.proto\x12\x17\x61gv.calibration.chassis\"\x07\n\x05\x45mpty\"4\n\x10StandardResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t\"\x96\x01\n\x15\x44iagnosticModeRequest\x12H\n\x0btarget_mode\x18\x01 \x01(\x0e\x32\x33.agv.calibration.chassis.DiagnosticModeRequest.Mode\"3\n\x04Mode\x12\x15\n\x11NORMAL_KINEMATICS\x10\x00\x12\x14\n\x10\x44IRECT_RAW_DRIVE\x10\x01\"\xfd\x02\n\rHardwareState\x12\x1d\n\x15hardware_timestamp_us\x18\x01 \x01(\x03\x12\x18\n\x10\x65ncoder_ticks_fl\x18\x02 \x01(\x03\x12\x18\n\x10\x65ncoder_ticks_fr\x18\x03 \x01(\x03\x12\x18\n\x10\x65ncoder_ticks_rl\x18\x04 \x01(\x03\x12\x18\n\x10\x65ncoder_ticks_rr\x18\x05 \x01(\x03\x12$\n\x1c\x61\x63tual_steer_angle_front_deg\x18\x06 \x01(\x01\x12#\n\x1b\x61\x63tual_steer_angle_rear_deg\x18\x07 \x01(\x01\x12\x16\n\x0e\x63urrent_fl_amp\x18\x08 \x01(\x01\x12\x16\n\x0e\x63urrent_fr_amp\x18\t \x01(\x01\x12\x16\n\x0e\x63urrent_rl_amp\x18\n \x01(\x01\x12\x16\n\x0e\x63urrent_rr_amp\x18\x0b \x01(\x01\x12\x1f\n\x17\x63urrent_steer_front_amp\x18\x0c \x01(\x01\x12\x19\n\x11\x64river_error_code\x18\r \x01(\r\"\x95\x01\n\x0fRawDriveRequest\x12\x14\n\x0ctest_case_id\x18\x01 \x01(\t\x12\x14\n\x0c\x66l_motor_rpm\x18\x02 \x01(\x01\x12\x14\n\x0c\x66r_motor_rpm\x18\x03 \x01(\x01\x12\x14\n\x0crl_motor_rpm\x18\x04 \x01(\x01\x12\x14\n\x0crr_motor_rpm\x18\x05 \x01(\x01\x12\x14\n\x0c\x64uration_sec\x18\x06 \x01(\x01\"\xec\x01\n\x0fRawSteerRequest\x12\x14\n\x0ctest_case_id\x18\x01 \x01(\t\x12\x1d\n\x15\x66ront_steer_angle_deg\x18\x02 \x01(\x01\x12\x1c\n\x14rear_steer_angle_deg\x18\x03 \x01(\x01\x12 \n\x13sweep_amplitude_deg\x18\x04 \x01(\x01H\x00\x88\x01\x01\x12\x1f\n\x12sweep_frequency_hz\x18\x05 \x01(\x01H\x01\x88\x01\x01\x12\x14\n\x0c\x64uration_sec\x18\x06 \x01(\x01\x42\x16\n\x14_sweep_amplitude_degB\x15\n\x13_sweep_frequency_hz\"\xdd\x04\n\x0fKinematicParams\x12\x1e\n\x11wheel_radius_fl_m\x18\x01 \x01(\x01H\x00\x88\x01\x01\x12\x1e\n\x11wheel_radius_fr_m\x18\x02 \x01(\x01H\x01\x88\x01\x01\x12\x1e\n\x11wheel_radius_rl_m\x18\x03 \x01(\x01H\x02\x88\x01\x01\x12\x1e\n\x11wheel_radius_rr_m\x18\x04 \x01(\x01H\x03\x88\x01\x01\x12(\n\x1bsteer_zero_offset_front_deg\x18\x05 \x01(\x01H\x04\x88\x01\x01\x12\'\n\x1asteer_zero_offset_rear_deg\x18\x06 \x01(\x01H\x05\x88\x01\x01\x12$\n\x17\x65\x66\x66\x65\x63tive_track_width_m\x18\x07 \x01(\x01H\x06\x88\x01\x01\x12#\n\x16\x65\x66\x66\x65\x63tive_wheel_base_m\x18\x08 \x01(\x01H\x07\x88\x01\x01\x12\x1b\n\x0eicr_offset_x_m\x18\t \x01(\x01H\x08\x88\x01\x01\x12\x1b\n\x0eicr_offset_y_m\x18\n \x01(\x01H\t\x88\x01\x01\x42\x14\n\x12_wheel_radius_fl_mB\x14\n\x12_wheel_radius_fr_mB\x14\n\x12_wheel_radius_rl_mB\x14\n\x12_wheel_radius_rr_mB\x1e\n\x1c_steer_zero_offset_front_degB\x1d\n\x1b_steer_zero_offset_rear_degB\x1a\n\x18_effective_track_width_mB\x19\n\x17_effective_wheel_base_mB\x11\n\x0f_icr_offset_x_mB\x11\n\x0f_icr_offset_y_m2\xa2\x05\n\x16\x41gvCalibChassisService\x12n\n\x11SetDiagnosticMode\x12..agv.calibration.chassis.DiagnosticModeRequest\x1a).agv.calibration.chassis.StandardResponse\x12\x63\n\x16HardwareEmergencyBrake\x12\x1e.agv.calibration.chassis.Empty\x1a).agv.calibration.chassis.StandardResponse\x12\x63\n\x17StreamHardwareTelemetry\x12\x1e.agv.calibration.chassis.Empty\x1a&.agv.calibration.chassis.HardwareState0\x01\x12m\n\x16\x45xecuteRawDriveCommand\x12(.agv.calibration.chassis.RawDriveRequest\x1a).agv.calibration.chassis.StandardResponse\x12m\n\x16\x45xecuteRawSteerCommand\x12(.agv.calibration.chassis.RawSteerRequest\x1a).agv.calibration.chassis.StandardResponse\x12p\n\x19\x43ommitKinematicParameters\x12(.agv.calibration.chassis.KinematicParams\x1a).agv.calibration.chassis.StandardResponseb\x06proto3')
_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'agv_calib_chassis_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
DESCRIPTOR._loaded_options = None
_globals['_EMPTY']._serialized_start=52
_globals['_EMPTY']._serialized_end=59
_globals['_STANDARDRESPONSE']._serialized_start=61
_globals['_STANDARDRESPONSE']._serialized_end=113
_globals['_DIAGNOSTICMODEREQUEST']._serialized_start=116
_globals['_DIAGNOSTICMODEREQUEST']._serialized_end=266
_globals['_DIAGNOSTICMODEREQUEST_MODE']._serialized_start=215
_globals['_DIAGNOSTICMODEREQUEST_MODE']._serialized_end=266
_globals['_HARDWARESTATE']._serialized_start=269
_globals['_HARDWARESTATE']._serialized_end=650
_globals['_RAWDRIVEREQUEST']._serialized_start=653
_globals['_RAWDRIVEREQUEST']._serialized_end=802
_globals['_RAWSTEERREQUEST']._serialized_start=805
_globals['_RAWSTEERREQUEST']._serialized_end=1041
_globals['_KINEMATICPARAMS']._serialized_start=1044
_globals['_KINEMATICPARAMS']._serialized_end=1649
_globals['_AGVCALIBCHASSISSERVICE']._serialized_start=1652
_globals['_AGVCALIBCHASSISSERVICE']._serialized_end=2326
# @@protoc_insertion_point(module_scope)
agv_calib_eu/scripts/agv_calib_chassis_pb2_grpc.py
+351
View File
@@ -0,0 +1,351 @@
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings
import agv_calib_chassis_pb2 as agv__calib__chassis__pb2
GRPC_GENERATED_VERSION = '1.78.0'
GRPC_VERSION = grpc.__version__
_version_not_supported = False
try:
from grpc._utilities import first_version_is_lower
_version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
_version_not_supported = True
if _version_not_supported:
raise RuntimeError(
f'The grpc package installed is at version {GRPC_VERSION},'
+ ' but the generated code in agv_calib_chassis_pb2_grpc.py depends on'
+ f' grpcio>={GRPC_GENERATED_VERSION}.'
+ f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+ f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
)
class AgvCalibChassisServiceStub(object):
"""=========================================================
核心服务：AGV 底盘底层硬件自诊与物理运动学标定代理服务
[部署端 Server]：Windows 车端 (只负责听口令、转电机、报裸数据)
[调用端 Client]：Linux 车间服务器 (负责发口令、看雷达真值、算误差)
=========================================================
"""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.SetDiagnosticMode = channel.unary_unary(
'/agv.calibration.chassis.AgvCalibChassisService/SetDiagnosticMode',
request_serializer=agv__calib__chassis__pb2.DiagnosticModeRequest.SerializeToString,
response_deserializer=agv__calib__chassis__pb2.StandardResponse.FromString,
_registered_method=True)
self.HardwareEmergencyBrake = channel.unary_unary(
'/agv.calibration.chassis.AgvCalibChassisService/HardwareEmergencyBrake',
request_serializer=agv__calib__chassis__pb2.Empty.SerializeToString,
response_deserializer=agv__calib__chassis__pb2.StandardResponse.FromString,
_registered_method=True)
self.StreamHardwareTelemetry = channel.unary_stream(
'/agv.calibration.chassis.AgvCalibChassisService/StreamHardwareTelemetry',
request_serializer=agv__calib__chassis__pb2.Empty.SerializeToString,
response_deserializer=agv__calib__chassis__pb2.HardwareState.FromString,
_registered_method=True)
self.ExecuteRawDriveCommand = channel.unary_unary(
'/agv.calibration.chassis.AgvCalibChassisService/ExecuteRawDriveCommand',
request_serializer=agv__calib__chassis__pb2.RawDriveRequest.SerializeToString,
response_deserializer=agv__calib__chassis__pb2.StandardResponse.FromString,
_registered_method=True)
self.ExecuteRawSteerCommand = channel.unary_unary(
'/agv.calibration.chassis.AgvCalibChassisService/ExecuteRawSteerCommand',
request_serializer=agv__calib__chassis__pb2.RawSteerRequest.SerializeToString,
response_deserializer=agv__calib__chassis__pb2.StandardResponse.FromString,
_registered_method=True)
self.CommitKinematicParameters = channel.unary_unary(
'/agv.calibration.chassis.AgvCalibChassisService/CommitKinematicParameters',
request_serializer=agv__calib__chassis__pb2.KinematicParams.SerializeToString,
response_deserializer=agv__calib__chassis__pb2.StandardResponse.FromString,
_registered_method=True)
class AgvCalibChassisServiceServicer(object):
"""=========================================================
核心服务：AGV 底盘底层硬件自诊与物理运动学标定代理服务
[部署端 Server]：Windows 车端 (只负责听口令、转电机、报裸数据)
[调用端 Client]：Linux 车间服务器 (负责发口令、看雷达真值、算误差)
=========================================================
"""
def SetDiagnosticMode(self, request, context):
"""---------------------------------------------------------
第一步：权限接管与安全熔断 (剥夺车端算法大脑)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：要求切断底盘运动学逆解，进入纯物理开环直驱模式
🚙 [Windows 返回 -> Linux]：返回接管是否成功的回执
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def HardwareEmergencyBrake(self, request, context):
"""💻 [Linux 发送 -> Windows]：无视一切状态立刻抱死电机的紧急急停指令
🚙 [Windows 返回 -> Linux]：返回急停执行状态
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def StreamHardwareTelemetry(self, request, context):
"""---------------------------------------------------------
第二步：打开体征监控水龙头 (数字孪生健康诊断)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：发送空请求，触发高频推流开关
🚙 [Windows 持续流式返回 -> Linux]：以 50Hz 频率持续不断地回传原始脉冲与电流
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ExecuteRawDriveCommand(self, request, context):
"""---------------------------------------------------------
第三步：原始物理开环考题下发 (逼迫底盘暴露机械缺陷)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：绕过算法，直接命令指定驱动轮以固定 RPM 盲跑
🚙 [Windows 返回 -> Linux]：返回电机是否已成功按给定 RPM 运转
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ExecuteRawSteerCommand(self, request, context):
"""💻 [Linux 发送 -> Windows]：直接对转向机构下发绝对物理角度 (测机械装歪的角度)
🚙 [Windows 返回 -> Linux]：返回舵机是否已开始执行角度指令
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def CommitKinematicParameters(self, request, context):
"""---------------------------------------------------------
第四步：物理本底参数定稿写值 (标定闭环结束)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：下发 Linux 结合外部真值算出的绝对物理修正系数
🚙 [Windows 返回 -> Linux]：将系数覆写到本地硬盘/驱动板后，返回成功回执
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_AgvCalibChassisServiceServicer_to_server(servicer, server):
rpc_method_handlers = {
'SetDiagnosticMode': grpc.unary_unary_rpc_method_handler(
servicer.SetDiagnosticMode,
request_deserializer=agv__calib__chassis__pb2.DiagnosticModeRequest.FromString,
response_serializer=agv__calib__chassis__pb2.StandardResponse.SerializeToString,
),
'HardwareEmergencyBrake': grpc.unary_unary_rpc_method_handler(
servicer.HardwareEmergencyBrake,
request_deserializer=agv__calib__chassis__pb2.Empty.FromString,
response_serializer=agv__calib__chassis__pb2.StandardResponse.SerializeToString,
),
'StreamHardwareTelemetry': grpc.unary_stream_rpc_method_handler(
servicer.StreamHardwareTelemetry,
request_deserializer=agv__calib__chassis__pb2.Empty.FromString,
response_serializer=agv__calib__chassis__pb2.HardwareState.SerializeToString,
),
'ExecuteRawDriveCommand': grpc.unary_unary_rpc_method_handler(
servicer.ExecuteRawDriveCommand,
request_deserializer=agv__calib__chassis__pb2.RawDriveRequest.FromString,
response_serializer=agv__calib__chassis__pb2.StandardResponse.SerializeToString,
),
'ExecuteRawSteerCommand': grpc.unary_unary_rpc_method_handler(
servicer.ExecuteRawSteerCommand,
request_deserializer=agv__calib__chassis__pb2.RawSteerRequest.FromString,
response_serializer=agv__calib__chassis__pb2.StandardResponse.SerializeToString,
),
'CommitKinematicParameters': grpc.unary_unary_rpc_method_handler(
servicer.CommitKinematicParameters,
request_deserializer=agv__calib__chassis__pb2.KinematicParams.FromString,
response_serializer=agv__calib__chassis__pb2.StandardResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'agv.calibration.chassis.AgvCalibChassisService', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
server.add_registered_method_handlers('agv.calibration.chassis.AgvCalibChassisService', rpc_method_handlers)
# This class is part of an EXPERIMENTAL API.
class AgvCalibChassisService(object):
"""=========================================================
核心服务：AGV 底盘底层硬件自诊与物理运动学标定代理服务
[部署端 Server]：Windows 车端 (只负责听口令、转电机、报裸数据)
[调用端 Client]：Linux 车间服务器 (负责发口令、看雷达真值、算误差)
=========================================================
"""
@staticmethod
def SetDiagnosticMode(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.chassis.AgvCalibChassisService/SetDiagnosticMode',
agv__calib__chassis__pb2.DiagnosticModeRequest.SerializeToString,
agv__calib__chassis__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def HardwareEmergencyBrake(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.chassis.AgvCalibChassisService/HardwareEmergencyBrake',
agv__calib__chassis__pb2.Empty.SerializeToString,
agv__calib__chassis__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def StreamHardwareTelemetry(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
'/agv.calibration.chassis.AgvCalibChassisService/StreamHardwareTelemetry',
agv__calib__chassis__pb2.Empty.SerializeToString,
agv__calib__chassis__pb2.HardwareState.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def ExecuteRawDriveCommand(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.chassis.AgvCalibChassisService/ExecuteRawDriveCommand',
agv__calib__chassis__pb2.RawDriveRequest.SerializeToString,
agv__calib__chassis__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def ExecuteRawSteerCommand(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.chassis.AgvCalibChassisService/ExecuteRawSteerCommand',
agv__calib__chassis__pb2.RawSteerRequest.SerializeToString,
agv__calib__chassis__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def CommitKinematicParameters(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.chassis.AgvCalibChassisService/CommitKinematicParameters',
agv__calib__chassis__pb2.KinematicParams.SerializeToString,
agv__calib__chassis__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
agv_calib_eu/scripts/agv_calib_control_pb2.py
+56
View File
@@ -0,0 +1,56 @@
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: agv_calib_control.proto
# Protobuf Python Version: 6.31.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
_runtime_version.Domain.PUBLIC,
6,
31,
1,
'',
'agv_calib_control.proto'
)
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x17\x61gv_calib_control.proto\x12\x17\x61gv.calibration.control\"\x07\n\x05\x45mpty\"4\n\x10StandardResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t\"\x8b\x01\n\x0bModeRequest\x12>\n\x0btarget_mode\x18\x01 \x01(\x0e\x32).agv.calibration.control.ModeRequest.Mode\"<\n\x04Mode\x12\x0f\n\x0bNORMAL_MODE\x10\x00\x12\x12\n\x0eOPEN_LOOP_MODE\x10\x01\x12\x0f\n\x0bTUNING_MODE\x10\x02\"p\n\x0fOpenLoopRequest\x12\x16\n\x0eleft_motor_cmd\x18\x01 \x01(\x01\x12\x17\n\x0fright_motor_cmd\x18\x02 \x01(\x01\x12\x16\n\x0esteering_angle\x18\x03 \x01(\x01\x12\x14\n\x0c\x64uration_sec\x18\x04 \x01(\x01\"h\n\x0fTrajectoryPoint\x12\x0b\n\x03x_m\x18\x01 \x01(\x01\x12\x0b\n\x03y_m\x18\x02 \x01(\x01\x12\x0f\n\x07yaw_rad\x18\x03 \x01(\x01\x12\x17\n\x0ftarget_speed_ms\x18\x04 \x01(\x01\x12\x11\n\tcurvature\x18\x05 \x01(\x01\"a\n\x11TrajectoryRequest\x12\x14\n\x0ctest_case_id\x18\x01 \x01(\t\x12\x36\n\x04path\x18\x02 \x03(\x0b\x32(.agv.calibration.control.TrajectoryPoint\"G\n\x13StepResponseRequest\x12\x1a\n\x12target_velocity_ms\x18\x01 \x01(\x01\x12\x14\n\x0c\x64uration_sec\x18\x02 \x01(\x01\"\xf9\x05\n\rControlParams\x12$\n\x17wheel_radius_left_ratio\x18\x01 \x01(\x01H\x00\x88\x01\x01\x12%\n\x18wheel_radius_right_ratio\x18\x02 \x01(\x01H\x01\x88\x01\x01\x12$\n\x17\x65\x66\x66\x65\x63tive_track_width_m\x18\x03 \x01(\x01H\x02\x88\x01\x01\x12%\n\x18steering_zero_offset_deg\x18\x04 \x01(\x01H\x03\x88\x01\x01\x12\x1b\n\x0epid_kp_lateral\x18\x05 \x01(\x01H\x04\x88\x01\x01\x12\x1b\n\x0epid_ki_lateral\x18\x06 \x01(\x01H\x05\x88\x01\x01\x12\x1b\n\x0epid_kd_lateral\x18\x07 \x01(\x01H\x06\x88\x01\x01\x12\x1b\n\x0epid_kp_heading\x18\x08 \x01(\x01H\x07\x88\x01\x01\x12\x1b\n\x0epid_ki_heading\x18\t \x01(\x01H\x08\x88\x01\x01\x12\x1b\n\x0epid_kd_heading\x18\n \x01(\x01H\t\x88\x01\x01\x12%\n\x18pure_pursuit_lookahead_m\x18\x0b \x01(\x01H\n\x88\x01\x01\x12!\n\x14mpc_weight_q_lateral\x18\x0c \x01(\x01H\x0b\x88\x01\x01\x12\"\n\x15mpc_weight_r_steering\x18\r \x01(\x01H\x0c\x88\x01\x01\x42\x1a\n\x18_wheel_radius_left_ratioB\x1b\n\x19_wheel_radius_right_ratioB\x1a\n\x18_effective_track_width_mB\x1b\n\x19_steering_zero_offset_degB\x11\n\x0f_pid_kp_lateralB\x11\n\x0f_pid_ki_lateralB\x11\n\x0f_pid_kd_lateralB\x11\n\x0f_pid_kp_headingB\x11\n\x0f_pid_ki_headingB\x11\n\x0f_pid_kd_headingB\x1b\n\x19_pure_pursuit_lookahead_mB\x17\n\x15_mpc_weight_q_lateralB\x18\n\x16_mpc_weight_r_steering\"\xad\x02\n\rTelemetryData\x12\x1d\n\x15hardware_timestamp_us\x18\x01 \x01(\x03\x12\x10\n\x08odom_x_m\x18\x02 \x01(\x01\x12\x10\n\x08odom_y_m\x18\x03 \x01(\x01\x12\x14\n\x0codom_yaw_rad\x18\x04 \x01(\x01\x12\x1e\n\x16\x66\x65\x65\x64\x62\x61\x63k_linear_vel_ms\x18\x05 \x01(\x01\x12!\n\x19\x66\x65\x65\x64\x62\x61\x63k_angular_vel_rads\x18\x06 \x01(\x01\x12\x1e\n\x16left_motor_current_amp\x18\x07 \x01(\x01\x12\x1f\n\x17right_motor_current_amp\x18\x08 \x01(\x01\x12\"\n\x1asteering_motor_current_amp\x18\t \x01(\x01\x12\x1b\n\x13\x63md_steering_output\x18\n \x01(\x01\x32\xd1\x06\n\x16\x41gvCalibControlService\x12\x61\n\x0eSetControlMode\x12$.agv.calibration.control.ModeRequest\x1a).agv.calibration.control.StandardResponse\x12Z\n\rEmergencyStop\x12\x1e.agv.calibration.control.Empty\x1a).agv.calibration.control.StandardResponse\x12i\n\x12\x45xecuteOpenLoopCmd\x12(.agv.calibration.control.OpenLoopRequest\x1a).agv.calibration.control.StandardResponse\x12m\n\x14\x46ollowTestTrajectory\x12*.agv.calibration.control.TrajectoryRequest\x1a).agv.calibration.control.StandardResponse\x12n\n\x13\x45xecuteStepResponse\x12,.agv.calibration.control.StepResponseRequest\x1a).agv.calibration.control.StandardResponse\x12k\n\x16InjectTuningParameters\x12&.agv.calibration.control.ControlParams\x1a).agv.calibration.control.StandardResponse\x12\x64\n\x17\x43ommitControlParameters\x12\x1e.agv.calibration.control.Empty\x1a).agv.calibration.control.StandardResponse\x12[\n\x0fStreamTelemetry\x12\x1e.agv.calibration.control.Empty\x1a&.agv.calibration.control.TelemetryData0\x01\x62\x06proto3')
_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'agv_calib_control_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
DESCRIPTOR._loaded_options = None
_globals['_EMPTY']._serialized_start=52
_globals['_EMPTY']._serialized_end=59
_globals['_STANDARDRESPONSE']._serialized_start=61
_globals['_STANDARDRESPONSE']._serialized_end=113
_globals['_MODEREQUEST']._serialized_start=116
_globals['_MODEREQUEST']._serialized_end=255
_globals['_MODEREQUEST_MODE']._serialized_start=195
_globals['_MODEREQUEST_MODE']._serialized_end=255
_globals['_OPENLOOPREQUEST']._serialized_start=257
_globals['_OPENLOOPREQUEST']._serialized_end=369
_globals['_TRAJECTORYPOINT']._serialized_start=371
_globals['_TRAJECTORYPOINT']._serialized_end=475
_globals['_TRAJECTORYREQUEST']._serialized_start=477
_globals['_TRAJECTORYREQUEST']._serialized_end=574
_globals['_STEPRESPONSEREQUEST']._serialized_start=576
_globals['_STEPRESPONSEREQUEST']._serialized_end=647
_globals['_CONTROLPARAMS']._serialized_start=650
_globals['_CONTROLPARAMS']._serialized_end=1411
_globals['_TELEMETRYDATA']._serialized_start=1414
_globals['_TELEMETRYDATA']._serialized_end=1715
_globals['_AGVCALIBCONTROLSERVICE']._serialized_start=1718
_globals['_AGVCALIBCONTROLSERVICE']._serialized_end=2567
# @@protoc_insertion_point(module_scope)
agv_calib_eu/scripts/agv_calib_control_pb2_grpc.py
+444
View File
@@ -0,0 +1,444 @@
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings
import agv_calib_control_pb2 as agv__calib__control__pb2
GRPC_GENERATED_VERSION = '1.78.0'
GRPC_VERSION = grpc.__version__
_version_not_supported = False
try:
from grpc._utilities import first_version_is_lower
_version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
_version_not_supported = True
if _version_not_supported:
raise RuntimeError(
f'The grpc package installed is at version {GRPC_VERSION},'
+ ' but the generated code in agv_calib_control_pb2_grpc.py depends on'
+ f' grpcio>={GRPC_GENERATED_VERSION}.'
+ f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+ f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
)
class AgvCalibControlServiceStub(object):
"""=========================================================
核心服务：AGV 运控大脑（PID/MPC）参数自动化寻优调教代理
[部署端 Server]：Windows车端 (满血保留自身算法，负责执行闭环追踪与高频汇报)
[调用端 Client]：Linux标定服务器 (上帝视角，负责发轨迹、看误差、AI打分与发新参数)
=========================================================
"""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.SetControlMode = channel.unary_unary(
'/agv.calibration.control.AgvCalibControlService/SetControlMode',
request_serializer=agv__calib__control__pb2.ModeRequest.SerializeToString,
response_deserializer=agv__calib__control__pb2.StandardResponse.FromString,
_registered_method=True)
self.EmergencyStop = channel.unary_unary(
'/agv.calibration.control.AgvCalibControlService/EmergencyStop',
request_serializer=agv__calib__control__pb2.Empty.SerializeToString,
response_deserializer=agv__calib__control__pb2.StandardResponse.FromString,
_registered_method=True)
self.ExecuteOpenLoopCmd = channel.unary_unary(
'/agv.calibration.control.AgvCalibControlService/ExecuteOpenLoopCmd',
request_serializer=agv__calib__control__pb2.OpenLoopRequest.SerializeToString,
response_deserializer=agv__calib__control__pb2.StandardResponse.FromString,
_registered_method=True)
self.FollowTestTrajectory = channel.unary_unary(
'/agv.calibration.control.AgvCalibControlService/FollowTestTrajectory',
request_serializer=agv__calib__control__pb2.TrajectoryRequest.SerializeToString,
response_deserializer=agv__calib__control__pb2.StandardResponse.FromString,
_registered_method=True)
self.ExecuteStepResponse = channel.unary_unary(
'/agv.calibration.control.AgvCalibControlService/ExecuteStepResponse',
request_serializer=agv__calib__control__pb2.StepResponseRequest.SerializeToString,
response_deserializer=agv__calib__control__pb2.StandardResponse.FromString,
_registered_method=True)
self.InjectTuningParameters = channel.unary_unary(
'/agv.calibration.control.AgvCalibControlService/InjectTuningParameters',
request_serializer=agv__calib__control__pb2.ControlParams.SerializeToString,
response_deserializer=agv__calib__control__pb2.StandardResponse.FromString,
_registered_method=True)
self.CommitControlParameters = channel.unary_unary(
'/agv.calibration.control.AgvCalibControlService/CommitControlParameters',
request_serializer=agv__calib__control__pb2.Empty.SerializeToString,
response_deserializer=agv__calib__control__pb2.StandardResponse.FromString,
_registered_method=True)
self.StreamTelemetry = channel.unary_stream(
'/agv.calibration.control.AgvCalibControlService/StreamTelemetry',
request_serializer=agv__calib__control__pb2.Empty.SerializeToString,
response_deserializer=agv__calib__control__pb2.TelemetryData.FromString,
_registered_method=True)
class AgvCalibControlServiceServicer(object):
"""=========================================================
核心服务：AGV 运控大脑（PID/MPC）参数自动化寻优调教代理
[部署端 Server]：Windows车端 (满血保留自身算法，负责执行闭环追踪与高频汇报)
[调用端 Client]：Linux标定服务器 (上帝视角，负责发轨迹、看误差、AI打分与发新参数)
=========================================================
"""
def SetControlMode(self, request, context):
"""---------------------------------------------------------
第一步：权限接管与生命周期安全管控
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：要求切断避障，但保留底层 PID/MPC 算法就绪
🚙 [Windows 返回 -> Linux]：回复模式切换成功，准备好接考题
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def EmergencyStop(self, request, context):
"""💻 [Linux 发送 -> Windows]：断网或飞车时的最高级别急停，无视一切直接刹车
🚙 [Windows 返回 -> Linux]：返回底层抱死结果
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ExecuteOpenLoopCmd(self, request, context):
"""---------------------------------------------------------
第二步：运动考题下发 (开环排雷 / 闭环寻优 / 波峰对齐)
---------------------------------------------------------
【场景A: 纯物理开环备用】
💻 [Linux 发送 -> Windows]：要求切断算法盲跑，多用于摸底或辅助验证
🚙 [Windows 返回 -> Linux]：确认已按指定 RPM/PWM 运转
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def FollowTestTrajectory(self, request, context):
"""【场景B: 算法闭环调优】
💻 [Linux 发送 -> Windows]：下发一条由几百个点组成的测试轨迹(如 S型贝塞尔曲线)
🚙 [Windows 返回 -> Linux]：收到轨迹后，车端立刻使用它自带的 PID/MPC 算法努力贴合轨迹跑圈
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ExecuteStepResponse(self, request, context):
"""【场景C: 波峰时序对齐】
💻 [Linux 发送 -> Windows]：下发极短促的阶跃加速指令，人为制造绝对速度波峰
🚙 [Windows 返回 -> Linux]：确认加速。(Linux 借此波峰算出网络的绝对 Time Offset)
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def InjectTuningParameters(self, request, context):
"""---------------------------------------------------------
第三步：运控参数 AI 寻优：动态热注入与最终固化
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：Linux 发现上一圈跑得差，AI算出了新的 PID/前瞻距离，要求立即热注入
🚙 [Windows 返回 -> Linux]：车端将新参数瞬间覆写进运行内存(不重启系统)，随时准备用新参数重跑
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def CommitControlParameters(self, request, context):
"""💻 [Linux 发送 -> Windows]：Linux 判定误差极小，调优结束，命令固化目前内存里的最高分参数
🚙 [Windows 返回 -> Linux]：车端将这组完美参数永久覆写进硬盘的 config.yaml 或系统注册表
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def StreamTelemetry(self, request, context):
"""---------------------------------------------------------
第四步：高频数字孪生体感上报 (50Hz)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：空包触发，命令车端开始疯狂推流
🚙 [Windows 持续流式返回 -> Linux]：以 50Hz 频率，持续上报自己的里程计坐标、速度和单调时间戳
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_AgvCalibControlServiceServicer_to_server(servicer, server):
rpc_method_handlers = {
'SetControlMode': grpc.unary_unary_rpc_method_handler(
servicer.SetControlMode,
request_deserializer=agv__calib__control__pb2.ModeRequest.FromString,
response_serializer=agv__calib__control__pb2.StandardResponse.SerializeToString,
),
'EmergencyStop': grpc.unary_unary_rpc_method_handler(
servicer.EmergencyStop,
request_deserializer=agv__calib__control__pb2.Empty.FromString,
response_serializer=agv__calib__control__pb2.StandardResponse.SerializeToString,
),
'ExecuteOpenLoopCmd': grpc.unary_unary_rpc_method_handler(
servicer.ExecuteOpenLoopCmd,
request_deserializer=agv__calib__control__pb2.OpenLoopRequest.FromString,
response_serializer=agv__calib__control__pb2.StandardResponse.SerializeToString,
),
'FollowTestTrajectory': grpc.unary_unary_rpc_method_handler(
servicer.FollowTestTrajectory,
request_deserializer=agv__calib__control__pb2.TrajectoryRequest.FromString,
response_serializer=agv__calib__control__pb2.StandardResponse.SerializeToString,
),
'ExecuteStepResponse': grpc.unary_unary_rpc_method_handler(
servicer.ExecuteStepResponse,
request_deserializer=agv__calib__control__pb2.StepResponseRequest.FromString,
response_serializer=agv__calib__control__pb2.StandardResponse.SerializeToString,
),
'InjectTuningParameters': grpc.unary_unary_rpc_method_handler(
servicer.InjectTuningParameters,
request_deserializer=agv__calib__control__pb2.ControlParams.FromString,
response_serializer=agv__calib__control__pb2.StandardResponse.SerializeToString,
),
'CommitControlParameters': grpc.unary_unary_rpc_method_handler(
servicer.CommitControlParameters,
request_deserializer=agv__calib__control__pb2.Empty.FromString,
response_serializer=agv__calib__control__pb2.StandardResponse.SerializeToString,
),
'StreamTelemetry': grpc.unary_stream_rpc_method_handler(
servicer.StreamTelemetry,
request_deserializer=agv__calib__control__pb2.Empty.FromString,
response_serializer=agv__calib__control__pb2.TelemetryData.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'agv.calibration.control.AgvCalibControlService', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
server.add_registered_method_handlers('agv.calibration.control.AgvCalibControlService', rpc_method_handlers)
# This class is part of an EXPERIMENTAL API.
class AgvCalibControlService(object):
"""=========================================================
核心服务：AGV 运控大脑（PID/MPC）参数自动化寻优调教代理
[部署端 Server]：Windows车端 (满血保留自身算法，负责执行闭环追踪与高频汇报)
[调用端 Client]：Linux标定服务器 (上帝视角，负责发轨迹、看误差、AI打分与发新参数)
=========================================================
"""
@staticmethod
def SetControlMode(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.control.AgvCalibControlService/SetControlMode',
agv__calib__control__pb2.ModeRequest.SerializeToString,
agv__calib__control__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def EmergencyStop(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.control.AgvCalibControlService/EmergencyStop',
agv__calib__control__pb2.Empty.SerializeToString,
agv__calib__control__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def ExecuteOpenLoopCmd(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.control.AgvCalibControlService/ExecuteOpenLoopCmd',
agv__calib__control__pb2.OpenLoopRequest.SerializeToString,
agv__calib__control__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def FollowTestTrajectory(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.control.AgvCalibControlService/FollowTestTrajectory',
agv__calib__control__pb2.TrajectoryRequest.SerializeToString,
agv__calib__control__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def ExecuteStepResponse(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.control.AgvCalibControlService/ExecuteStepResponse',
agv__calib__control__pb2.StepResponseRequest.SerializeToString,
agv__calib__control__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def InjectTuningParameters(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.control.AgvCalibControlService/InjectTuningParameters',
agv__calib__control__pb2.ControlParams.SerializeToString,
agv__calib__control__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def CommitControlParameters(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.control.AgvCalibControlService/CommitControlParameters',
agv__calib__control__pb2.Empty.SerializeToString,
agv__calib__control__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def StreamTelemetry(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
'/agv.calibration.control.AgvCalibControlService/StreamTelemetry',
agv__calib__control__pb2.Empty.SerializeToString,
agv__calib__control__pb2.TelemetryData.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
agv_calib_eu/scripts/agv_calib_sensor_pb2.py
+54
View File
@@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: agv_calib_sensor.proto
# Protobuf Python Version: 6.31.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
_runtime_version.Domain.PUBLIC,
6,
31,
1,
'',
'agv_calib_sensor.proto'
)
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x16\x61gv_calib_sensor.proto\x12\x16\x61gv.calibration.sensor\"4\n\x10StandardResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t\"b\n\x0bPoseRequest\x12\x12\n\ntarget_x_m\x18\x01 \x01(\x01\x12\x12\n\ntarget_y_m\x18\x02 \x01(\x01\x12\x16\n\x0etarget_yaw_deg\x18\x03 \x01(\x01\x12\x13\n\x0bis_relative\x18\x04 \x01(\x08\"$\n\x0e\x43\x61ptureRequest\x12\x12\n\nsensor_ids\x18\x01 \x03(\t\"W\n\x0f\x43\x61ptureResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x1c\n\x14\x63\x61pture_timestamp_us\x18\x02 \x01(\x03\x12\x15\n\rerror_message\x18\x03 \x01(\t\"C\n\x10\x44\x61taFetchRequest\x12\x1c\n\x14\x63\x61pture_timestamp_us\x18\x01 \x01(\x03\x12\x11\n\tsensor_id\x18\x02 \x01(\t\"J\n\tFileChunk\x12\x12\n\nchunk_data\x18\x01 \x01(\x0c\x12\x15\n\ris_last_chunk\x18\x02 \x01(\x08\x12\x12\n\nformat_ext\x18\x03 \x01(\t\"j\n\x10\x43\x61meraIntrinsics\x12\x11\n\tcamera_id\x18\x01 \x01(\t\x12\n\n\x02\x66x\x18\x02 \x01(\x01\x12\n\n\x02\x66y\x18\x03 \x01(\x01\x12\n\n\x02\x63x\x18\x04 \x01(\x01\x12\n\n\x02\x63y\x18\x05 \x01(\x01\x12\x13\n\x0b\x64ist_coeffs\x18\x06 \x03(\x01\"\xb0\x01\n\x10SensorExtrinsics\x12\x14\n\x0csource_frame\x18\x01 \x01(\t\x12\x14\n\x0ctarget_frame\x18\x02 \x01(\t\x12\x12\n\ntrans_x_mm\x18\x03 \x01(\x01\x12\x12\n\ntrans_y_mm\x18\x04 \x01(\x01\x12\x12\n\ntrans_z_mm\x18\x05 \x01(\x01\x12\x10\n\x08roll_deg\x18\x06 \x01(\x01\x12\x11\n\tpitch_deg\x18\x07 \x01(\x01\x12\x0f\n\x07yaw_deg\x18\x08 \x01(\x01\"\xb1\x01\n\x12\x43\x61librationPayload\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x44\n\x12updated_intrinsics\x18\x02 \x03(\x0b\x32(.agv.calibration.sensor.CameraIntrinsics\x12\x44\n\x12updated_extrinsics\x18\x03 \x03(\x0b\x32(.agv.calibration.sensor.SensorExtrinsics2\xa0\x04\n\x18SensorCalibrationService\x12\x66\n\x15MoveToObservationPose\x12#.agv.calibration.sensor.PoseRequest\x1a(.agv.calibration.sensor.StandardResponse\x12\x65\n\x12TriggerSyncCapture\x12&.agv.calibration.sensor.CaptureRequest\x1a\'.agv.calibration.sensor.CaptureResponse\x12^\n\rDownloadImage\x12(.agv.calibration.sensor.DataFetchRequest\x1a!.agv.calibration.sensor.FileChunk0\x01\x12\x63\n\x12\x44ownloadPointCloud\x12(.agv.calibration.sensor.DataFetchRequest\x1a!.agv.calibration.sensor.FileChunk0\x01\x12p\n\x18\x43ommitCalibrationResults\x12*.agv.calibration.sensor.CalibrationPayload\x1a(.agv.calibration.sensor.StandardResponseb\x06proto3')
_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'agv_calib_sensor_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
DESCRIPTOR._loaded_options = None
_globals['_STANDARDRESPONSE']._serialized_start=50
_globals['_STANDARDRESPONSE']._serialized_end=102
_globals['_POSEREQUEST']._serialized_start=104
_globals['_POSEREQUEST']._serialized_end=202
_globals['_CAPTUREREQUEST']._serialized_start=204
_globals['_CAPTUREREQUEST']._serialized_end=240
_globals['_CAPTURERESPONSE']._serialized_start=242
_globals['_CAPTURERESPONSE']._serialized_end=329
_globals['_DATAFETCHREQUEST']._serialized_start=331
_globals['_DATAFETCHREQUEST']._serialized_end=398
_globals['_FILECHUNK']._serialized_start=400
_globals['_FILECHUNK']._serialized_end=474
_globals['_CAMERAINTRINSICS']._serialized_start=476
_globals['_CAMERAINTRINSICS']._serialized_end=582
_globals['_SENSOREXTRINSICS']._serialized_start=585
_globals['_SENSOREXTRINSICS']._serialized_end=761
_globals['_CALIBRATIONPAYLOAD']._serialized_start=764
_globals['_CALIBRATIONPAYLOAD']._serialized_end=941
_globals['_SENSORCALIBRATIONSERVICE']._serialized_start=944
_globals['_SENSORCALIBRATIONSERVICE']._serialized_end=1488
# @@protoc_insertion_point(module_scope)
agv_calib_eu/scripts/agv_calib_sensor_pb2_grpc.py
+306
View File
@@ -0,0 +1,306 @@
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings
import agv_calib_sensor_pb2 as agv__calib__sensor__pb2
GRPC_GENERATED_VERSION = '1.78.0'
GRPC_VERSION = grpc.__version__
_version_not_supported = False
try:
from grpc._utilities import first_version_is_lower
_version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
_version_not_supported = True
if _version_not_supported:
raise RuntimeError(
f'The grpc package installed is at version {GRPC_VERSION},'
+ ' but the generated code in agv_calib_sensor_pb2_grpc.py depends on'
+ f' grpcio>={GRPC_GENERATED_VERSION}.'
+ f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+ f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
)
class SensorCalibrationServiceStub(object):
"""=========================================================
核心服务：多传感器自动化外参标定代理服务
[部署端 Server]：Windows车端 (充当带轮子的三脚架与文件下载服务器)
[调用端 Client]：Linux标定服务器 (掌控状态机、拉取大文件、算 Ceres 矩阵)
=========================================================
"""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.MoveToObservationPose = channel.unary_unary(
'/agv.calibration.sensor.SensorCalibrationService/MoveToObservationPose',
request_serializer=agv__calib__sensor__pb2.PoseRequest.SerializeToString,
response_deserializer=agv__calib__sensor__pb2.StandardResponse.FromString,
_registered_method=True)
self.TriggerSyncCapture = channel.unary_unary(
'/agv.calibration.sensor.SensorCalibrationService/TriggerSyncCapture',
request_serializer=agv__calib__sensor__pb2.CaptureRequest.SerializeToString,
response_deserializer=agv__calib__sensor__pb2.CaptureResponse.FromString,
_registered_method=True)
self.DownloadImage = channel.unary_stream(
'/agv.calibration.sensor.SensorCalibrationService/DownloadImage',
request_serializer=agv__calib__sensor__pb2.DataFetchRequest.SerializeToString,
response_deserializer=agv__calib__sensor__pb2.FileChunk.FromString,
_registered_method=True)
self.DownloadPointCloud = channel.unary_stream(
'/agv.calibration.sensor.SensorCalibrationService/DownloadPointCloud',
request_serializer=agv__calib__sensor__pb2.DataFetchRequest.SerializeToString,
response_deserializer=agv__calib__sensor__pb2.FileChunk.FromString,
_registered_method=True)
self.CommitCalibrationResults = channel.unary_unary(
'/agv.calibration.sensor.SensorCalibrationService/CommitCalibrationResults',
request_serializer=agv__calib__sensor__pb2.CalibrationPayload.SerializeToString,
response_deserializer=agv__calib__sensor__pb2.StandardResponse.FromString,
_registered_method=True)
class SensorCalibrationServiceServicer(object):
"""=========================================================
核心服务：多传感器自动化外参标定代理服务
[部署端 Server]：Windows车端 (充当带轮子的三脚架与文件下载服务器)
[调用端 Client]：Linux标定服务器 (掌控状态机、拉取大文件、算 Ceres 矩阵)
=========================================================
"""
def MoveToObservationPose(self, request, context):
"""---------------------------------------------------------
第一步：物理走位 (走)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：调度车辆开到指定的标定观测点，到达后【绝对刹车静止】
🚙 [Windows 返回 -> Linux]：物理到位抱死刹车后，返回成功回执
🚨 业务潜台词：Linux 收到回执后，必须在代码里强制 sleep(0.5s) 等待避震悬挂平息，冻结物理空间！
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def TriggerSyncCapture(self, request, context):
"""---------------------------------------------------------
第二步：防延迟同步锁存 (停与拍)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：命令车辆瞬间将底层相机的显存和雷达的点云冻结到后备内存池
🚙 [Windows 返回 -> Linux]：立刻锁存，并返回高精度硬件时间戳，作为后续拉取大文件的唯一“取件码”
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def DownloadImage(self, request, context):
"""---------------------------------------------------------
第三步：大文件流式下载 (传 —— 破解 Windows 网络延迟的绝杀)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：凭“取件码”请求下载巨大的图片/点云原文件
🚙 [Windows 持续流式返回 -> Linux]：将 5MB+ 的无损文件切成小块，像流水一样源源不断传回 Linux
🚨 业务潜台词：必须使用 stream 关键字！否则 gRPC 会因为单包超过 4MB 瞬间崩溃！
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def DownloadPointCloud(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def CommitCalibrationResults(self, request, context):
"""---------------------------------------------------------
第四步：标定闭环定稿 (写)
---------------------------------------------------------
💻 [Linux 发送 -> Windows]：Linux 攒够数据算完复杂的 4x4 外参矩阵后，下发给车端持久化保存
🚙 [Windows 返回 -> Linux]：车端收到后直接覆写 sensor_config.yaml 或注册表，返回成功
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_SensorCalibrationServiceServicer_to_server(servicer, server):
rpc_method_handlers = {
'MoveToObservationPose': grpc.unary_unary_rpc_method_handler(
servicer.MoveToObservationPose,
request_deserializer=agv__calib__sensor__pb2.PoseRequest.FromString,
response_serializer=agv__calib__sensor__pb2.StandardResponse.SerializeToString,
),
'TriggerSyncCapture': grpc.unary_unary_rpc_method_handler(
servicer.TriggerSyncCapture,
request_deserializer=agv__calib__sensor__pb2.CaptureRequest.FromString,
response_serializer=agv__calib__sensor__pb2.CaptureResponse.SerializeToString,
),
'DownloadImage': grpc.unary_stream_rpc_method_handler(
servicer.DownloadImage,
request_deserializer=agv__calib__sensor__pb2.DataFetchRequest.FromString,
response_serializer=agv__calib__sensor__pb2.FileChunk.SerializeToString,
),
'DownloadPointCloud': grpc.unary_stream_rpc_method_handler(
servicer.DownloadPointCloud,
request_deserializer=agv__calib__sensor__pb2.DataFetchRequest.FromString,
response_serializer=agv__calib__sensor__pb2.FileChunk.SerializeToString,
),
'CommitCalibrationResults': grpc.unary_unary_rpc_method_handler(
servicer.CommitCalibrationResults,
request_deserializer=agv__calib__sensor__pb2.CalibrationPayload.FromString,
response_serializer=agv__calib__sensor__pb2.StandardResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'agv.calibration.sensor.SensorCalibrationService', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
server.add_registered_method_handlers('agv.calibration.sensor.SensorCalibrationService', rpc_method_handlers)
# This class is part of an EXPERIMENTAL API.
class SensorCalibrationService(object):
"""=========================================================
核心服务：多传感器自动化外参标定代理服务
[部署端 Server]：Windows车端 (充当带轮子的三脚架与文件下载服务器)
[调用端 Client]：Linux标定服务器 (掌控状态机、拉取大文件、算 Ceres 矩阵)
=========================================================
"""
@staticmethod
def MoveToObservationPose(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.sensor.SensorCalibrationService/MoveToObservationPose',
agv__calib__sensor__pb2.PoseRequest.SerializeToString,
agv__calib__sensor__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def TriggerSyncCapture(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.sensor.SensorCalibrationService/TriggerSyncCapture',
agv__calib__sensor__pb2.CaptureRequest.SerializeToString,
agv__calib__sensor__pb2.CaptureResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def DownloadImage(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
'/agv.calibration.sensor.SensorCalibrationService/DownloadImage',
agv__calib__sensor__pb2.DataFetchRequest.SerializeToString,
agv__calib__sensor__pb2.FileChunk.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def DownloadPointCloud(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
'/agv.calibration.sensor.SensorCalibrationService/DownloadPointCloud',
agv__calib__sensor__pb2.DataFetchRequest.SerializeToString,
agv__calib__sensor__pb2.FileChunk.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def CommitCalibrationResults(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/agv.calibration.sensor.SensorCalibrationService/CommitCalibrationResults',
agv__calib__sensor__pb2.CalibrationPayload.SerializeToString,
agv__calib__sensor__pb2.StandardResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
agv_calib_eu/scripts/mock_windows_agv.py
+51
View File
@@ -0,0 +1,51 @@
import grpc
from concurrent import futures
import time
# 导入刚才生成的 3 个 Python 契约库
import agv_calib_chassis_pb2 as chassis_pb2, agv_calib_chassis_pb2_grpc as chassis_grpc
import agv_calib_control_pb2 as control_pb2, agv_calib_control_pb2_grpc as control_grpc
import agv_calib_sensor_pb2 as sensor_pb2, agv_calib_sensor_pb2_grpc as sensor_grpc
# 1. 伪装物理底盘
class FakeChassis(chassis_grpc.AgvCalibChassisServiceServicer):
def SetDiagnosticMode(self, request, context):
mode = "纯物理直驱" if request.target_mode == 1 else "正常算法"
print(f"\n[⚙️ 底盘硬件] 收到 Linux 夺权指令！切换至: {mode}模式")
return chassis_pb2.StandardResponse(success=True, message="底盘已交出控制权")
def StreamHardwareTelemetry(self, request, context):
print("[🌊 底盘硬件] 开始向 Linux 推送 50Hz 硬件裸数据流...")
while context.is_active():
# 疯狂发送假脉冲数据
yield chassis_pb2.HardwareState(hardware_timestamp_us=int(time.time()*1000000), encoder_ticks_fl=1024, current_fl_amp=2.5)
time.sleep(0.02) # 50Hz
# 2. 伪装运控大脑
class FakeControl(control_grpc.AgvCalibControlServiceServicer):
def InjectTuningParameters(self, request, context):
print(f"\n[🧠 运控大脑] 收到 AI 热注入参数！Kp 被修改为: {request.pid_kp_lateral}")
return control_pb2.StandardResponse(success=True, message="PID参数已瞬间写入内存")
# 3. 伪装传感器代理
class FakeSensor(sensor_grpc.SensorCalibrationServiceServicer):
def TriggerSyncCapture(self, request, context):
print(f"\n[📷 传感器代理] 收到冻结指令！要求拍摄: {request.sensor_ids}")
time.sleep(0.5) # 模拟硬件快门延迟
print("[📷 传感器代理] 咔嚓！照片和点云已锁存。取件码: 999888777")
return sensor_pb2.CaptureResponse(success=True, capture_timestamp_us=999888777)
def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
# 三大假硬件全挂载到同一个网络端口！
chassis_grpc.add_AgvCalibChassisServiceServicer_to_server(FakeChassis(), server)
control_grpc.add_AgvCalibControlServiceServicer_to_server(FakeControl(), server)
sensor_grpc.add_SensorCalibrationServiceServicer_to_server(FakeSensor(), server)
server.add_insecure_port('[::]:50051')
print("🚀 [全能 Windows 假车端] 已启动，正在监听 50051 端口，等待 ROS 2 召唤...")
server.start()
server.wait_for_termination()
if __name__ == '__main__':
serve()
agv_calib_eu/txt.txt
+1
View File
@@ -0,0 +1 @@
cd scripts && python3 -m grpc_tools.protoc -I../proto --python_out=. --grpc_python_out=. ../proto/*.proto