ROS service callback contracts

For each registered service server below, the declaration identifies the type and name expression. Request/response wire fields come from the corresponding installed .srv definition; this index adds callback behavior observed statically. It does not turn unresolved names into deployed endpoints.

Use ros2 service list -t and ros2 interface show PACKAGE/srv/TYPE before a direct call. Read-only inspection is different from invoking a service: callbacks can move hardware, synthesize speech, change state or call external providers.

performances: _join_name(self.namespace, 'reload_properties')

Type: Trigger. Callback: self.reload_properties_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:126.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:217.

Return expressions:

_response(response, Trigger.Response, success=True)

Direct calls to follow for delegated behavior: _response, self.load_properties.

performances: _join_name(self.namespace, 'set_properties')

Type: perf_srv.SetProperties. Callback: self.set_properties_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:127.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:235.

Request attributes read: id, properties.

Return expressions:

_response(response, perf_srv.SetProperties.Response, success=True)

Direct calls to follow for delegated behavior: _response, json.loads, self.ros.set_variable.

performances: _join_name(self.namespace, 'load')

Type: perf_srv.Load. Callback: self.load_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:128.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:239.

Request attributes read: id.

Return expressions:

_response(response, perf_srv.Load.Response, success=True, performance=json.dumps(self.load(request.id)))

Direct calls to follow for delegated behavior: _response, json.dumps, self.load.

performances: _join_name(self.namespace, 'load_performance')

Type: perf_srv.LoadPerformance. Callback: self.load_performance_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:129.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:247.

Request attributes read: performance.

Return expressions:

_response(response, perf_srv.LoadPerformance.Response, success=True)

Direct calls to follow for delegated behavior: _response, json.loads, self.load_performance.

performances: _join_name(self.namespace, 'unload')

Type: Trigger. Callback: self.unload_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:130.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:221.

Return expressions:

_response(response, Trigger.Response, success=True)

Direct calls to follow for delegated behavior: _response, self.unload.

performances: _join_name(self.namespace, 'run')

Type: perf_srv.Run. Callback: self.run_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:131.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:419.

Return expressions:

_response(response, perf_srv.Run.Response, success=self.run(getattr(request, 'start_time', 0.0)))

Direct calls to follow for delegated behavior: _response, getattr, self.run.

performances: _join_name(self.namespace, 'run_by_name')

Type: perf_srv.RunByName. Callback: self.run_by_name_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:132.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:251.

Request attributes read: id.

Return expressions:

_response(response, perf_srv.RunByName.Response, success=self.run(0.0))
_response(response, perf_srv.RunByName.Response, success=False)

Direct calls to follow for delegated behavior: _response, self.load, self.run, self.stop.

performances: _join_name(self.namespace, 'run_full_performance')

Type: perf_srv.RunByName. Callback: self.run_full_performance_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:133.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:265.

Request attributes read: id.

Return expressions:

_response(response, perf_srv.RunByName.Response, success=self.run_full_performance(request.id, unload_finished=True))

Direct calls to follow for delegated behavior: _response, self.run_full_performance.

performances: _join_name(self.namespace, 'resume')

Type: perf_srv.Resume. Callback: self.resume_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:134.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:459.

Return expressions:

_response(response, perf_srv.Resume.Response, success=success, time=run_time)

Direct calls to follow for delegated behavior: _response, self.get_run_time, self.resume.

performances: _join_name(self.namespace, 'pause')

Type: perf_srv.Pause. Callback: self.pause_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:135.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:511.

Return expressions:

_response(response, perf_srv.Pause.Response, success=False, time=0.0)
_response(response, perf_srv.Pause.Response, success=True, time=self.get_run_time())

Direct calls to follow for delegated behavior: _response, self.get_run_time, self.pause.

performances: _join_name(self.namespace, 'stop')

Type: perf_srv.Stop. Callback: self.stop_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:136.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:508.

Return expressions:

_response(response, perf_srv.Stop.Response, success=True, time=self.stop())

Direct calls to follow for delegated behavior: _response, self.stop.

performances: _join_name(self.namespace, 'current')

Type: perf_srv.Current. Callback: self.current_callback. Registration: head-ros-nodes/ros/performances/scripts/runner.py:137.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:538.

Return expressions:

_response(response, perf_srv.Current.Response, performance=json.dumps(self.running_performance), current_time=current_time, running=running, paused=self.paused, active_timeline_index=active_timeline_index, active_timeline_id=active_timeline_id)

Direct calls to follow for delegated behavior: _response, json.dumps, self.get_active_timeline, self.get_run_time.

performances: _join_name(self.namespace, 'is_performance_loaded')

Type: Trigger. Callback: self.check_is_performance_loaded. Registration: head-ros-nodes/ros/performances/scripts/runner.py:138.

Callback source: head-ros-nodes/ros/performances/scripts/runner.py:554.

Return expressions:

_response(response, Trigger.Response, success=bool(success))

Direct calls to follow for delegated behavior: _response, bool, len.

asr: '~/set_context'

Type: StringTrigger. Callback: self.ros_service_set_context. Registration: head-ros-nodes/ros/asr/scripts/recognizer_node.py:171.

Callback source: head-ros-nodes/ros/asr/scripts/recognizer_node.py:616.

Request attributes read: data.

Response assignments (conditional branches may set different values):

response.success = True
response.message = 'ok'

Return expressions:

response

Direct calls to follow for delegated behavior: callable, getattr, p.strip, req.data.split, self.phrase_engine.add_speech_phrase, self.phrase_engine.clear_speech_phrases, self.phrase_engine.get_phrases, update_phrases.

ros_tts: speech_name('length')

Type: TTSLength. Callback: self.tts_length. Registration: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:202.

Callback source: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:403.

Request attributes read: lang, txt.

Return expressions:

self._service_response(response, TTSLength.Response(length=1), length=1)
self._service_response(response, TTSLength.Response(length=duration), length=duration)

Direct calls to follow for delegated behavior: TTSLength.Response, self._get_tts_params, self._service_response, self.client.tts, self.word_process, tts_response.get_duration.

ros_tts: speech_name('data')

Type: TTSData. Callback: self.tts_data. Registration: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:203.

Callback source: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:419.

Request attributes read: lang, txt.

Return expressions:

self._service_response(response, TTSData.Response(data=''), data='')
self._service_response(response, TTSData.Response(data=tts_data), data=tts_data)

Direct calls to follow for delegated behavior: TTSData.Response, self._get_tts_params, self._service_response, self.client.tts, self.format_tts_data, self.word_process.

ros_tts: speech_name('interrupt')

Type: StringTrigger. Callback: self.interrupt. Registration: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:204.

Callback source: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:436.

Interrupts the TTS, and returns the remaining text it left out

Request attributes read: data.

Response assignments (conditional branches may set different values):

response.success = False
response.message = 'Illegal interruption type: %r' % req.data
response.message = self.executor.remaining_tts
response.success = True
response.message = 'TTS is not running'

Return expressions:

response

Direct calls to follow for delegated behavior: StringTrigger.Response, self.executor.interrupt.set, self.executor.tts_done.wait, self.executor.tts_started.is_set.

ros_tts: speech_name('wait_for_tts')

Type: Trigger. Callback: self.wait_for_tts. Registration: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:207.

Callback source: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:454.

Wait for all the tts done

Response assignments (conditional branches may set different values):

response.success = True
response.success = False
response.message = str(ex)

Return expressions:

response

Direct calls to follow for delegated behavior: Trigger.Response, self.tts_queue.join, str.

ros_tts: speech_name('request_queue_size')

Type: GetInteger. Callback: self.request_queue_size. Registration: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:208.

Callback source: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:466.

Response assignments (conditional branches may set different values):

response.value = self.tts_queue.unfinished_tasks

Return expressions:

response

Direct calls to follow for delegated behavior: GetInteger.Response.

ros_tts: speech_name('clear_tts_queue')

Type: Trigger. Callback: self.clear_tts_queue. Registration: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:209.

Callback source: head-ros-nodes/ros/ros_tts/scripts/tts_talker.py:471.

Response assignments (conditional branches may set different values):

response.success = True
response.success = False
response.message = 'TTS has already started'
response.message = f'Cleared {job_count} TTS requests'

Return expressions:

response

Direct calls to follow for delegated behavior: Trigger.Response, self.executor.tts_started.is_set, self.executor.update_status, self.tts_cleared.clear, self.tts_cleared.set, self.tts_queue.empty, self.tts_queue.get, self.tts_queue.task_done, time.sleep.

ros_chatbot: f'{ELEVENLABS_AGENT_NAMESPACE}/send_user_message'

Type: StringTrigger. Callback: self._send_user_message_service. Registration: head-ros-nodes/ros/ros_chatbot/scripts/11_labs_agent.py:787.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/11_labs_agent.py:993.

Return expressions:

_string_trigger_response(False, 'Missing text')
_string_trigger_response(True, 'restarted')
_string_trigger_response(False, 'ElevenLabs session is not running')
_string_trigger_response(False, 'ElevenLabs SDK does not support send_user_message')
_string_trigger_response(True, 'sent')
_string_trigger_response(True, 'ElevenLabs agent is disabled')
_string_trigger_response(False, str(exc))

Direct calls to follow for delegated behavior: _string_trigger_response, getattr, self._is_reset_command, self._publish_agent_event, self._remember_pending_text_transcript, self._touch_activity, self.start_session, self.stop_session, send_user_message, str, str(getattr(request, 'data', '') or '').strip.

ros_chatbot: f'{ELEVENLABS_AGENT_NAMESPACE}/contextual_update'

Type: StringTrigger. Callback: self._contextual_update_service. Registration: head-ros-nodes/ros/ros_chatbot/scripts/11_labs_agent.py:792.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/11_labs_agent.py:1020.

Return expressions:

_string_trigger_response(False, 'Missing contextual update')
_string_trigger_response(True, 'sent')
_string_trigger_response(False, str(exc))

Direct calls to follow for delegated behavior: _string_trigger_response, getattr, self._send_contextual_update, str, str(getattr(request, 'data', '') or '').strip.

ros_chatbot: f'{ELEVENLABS_AGENT_NAMESPACE}/restart'

Type: StringTrigger. Callback: self._restart_service. Registration: head-ros-nodes/ros/ros_chatbot/scripts/11_labs_agent.py:797.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/11_labs_agent.py:1030.

Return expressions:

_string_trigger_response(True, 'restarted')
_string_trigger_response(True, 'ElevenLabs agent is disabled')

Direct calls to follow for delegated behavior: _string_trigger_response, getattr, self.start_session, self.stop_session, str, str(getattr(request, 'data', '') or 'manual').strip.

ros_chatbot: 'available_agents'

Type: StringArray. Callback: self.available_agents. Registration: head-ros-nodes/ros/ros_chatbot/scripts/chat.py:204.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/chat.py:309.

Response assignments (conditional branches may set different values):

response.data = self.publish_available_agents_param()

Return expressions:

response

Direct calls to follow for delegated behavior: self.publish_available_agents_param.

ros_chatbot: '/hr/interaction/prompts/response_prompt'

Type: Trigger. Callback: self.trigger_service_callback. Registration: head-ros-nodes/ros/ros_chatbot/scripts/priming.py:33.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/priming.py:175.

Response assignments (conditional branches may set different values):

response.success = True
response.message = self.format_response_prompt()

Return expressions:

response

Direct calls to follow for delegated behavior: self.format_response_prompt.

ros_chatbot: 'available_scenes'

Type: StringArray. Callback: self.available_scenes. Registration: head-ros-nodes/ros/ros_chatbot/scripts/scene_manager.py:75.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/scene_manager.py:107.

Response assignments (conditional branches may set different values):

response.data = [scene.name for scene in self.scenes]

Return expressions:

response

ros_chatbot: 'get_intent'

Type: GetIntent. Callback: self.get_intent. Registration: head-ros-nodes/ros/ros_chatbot/scripts/services.py:53.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/services.py:75.

Request attributes read: lang, text.

Response assignments (conditional branches may set different values):

response.intent = result['intent']['name']
response.confidence = result['intent']['confidence']

Return expressions:

response

Direct calls to follow for delegated behavior: self._intent_detector.detect_intent.

ros_chatbot: 'describe_view'

Type: StringTrigger. Callback: self.describe_view. Registration: head-ros-nodes/ros/ros_chatbot/scripts/services.py:54.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/services.py:85.

Request attributes read: data.

Response assignments (conditional branches may set different values):

response.success = False
response.message = 'OpenAI client is not configured'
response.success = True
response.message = json.dumps({'content': api_response.choices[0].message.content, 'image_file_name': image_file_name, 'time': {'value': datetime_str, 'format': '%Y%m%d%H%M%S'}, 'utc_time': current_time.isoformat()})
response.message = str(exc)
response.message = 'No image feed'

Return expressions:

response
self.client.chat.completions.create(model=self.model, messages=[{'role': 'user', 'content': [{'type': 'text', 'text': request.data}, {'type': 'image_url', 'image_url': {'url': f'data:image/jpeg;base64,{base64_image}'}}]}], max_tokens=800)

Direct calls to follow for delegated behavior: base64.b64encode, base64.b64encode(image_bytes).decode, current_time.isoformat, datetime.now, datetime.strftime, getattr, image_file.write, json.dumps, open, os.makedirs, os.path.join, preferred_base_url, run_with_provider_proxy_fallback, self._build_openai_client, self.client.chat.completions.create, setattr, str, uuid.uuid4.

ros_chatbot: f'{base_node}/chat'

Type: AgentChat. Callback: self.handle_agent_chat. Registration: head-ros-nodes/ros/ros_chatbot/scripts/spark_client.py:104.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/spark_client.py:193.

Request attributes read: lang, request_id, session, text.

Response assignments (conditional branches may set different values):

response.answer = ''
response.state = 0
response.score = 0.0
response.state = 1
response.score = 0.95

Return expressions:

response

Direct calls to follow for delegated behavior: self.bridge.build_user_message, self.bridge.pending_requests.pop, self.send_to_spark, self.ws_connected.is_set.

ros_chatbot: f'{base_node}/feedback'

Type: AgentFeedback. Callback: self.handle_agent_feedback. Registration: head-ros-nodes/ros/ros_chatbot/scripts/spark_client.py:105.

Callback source: head-ros-nodes/ros/ros_chatbot/scripts/spark_client.py:222.

Request attributes read: chosen, hybrid, request_id.

Response assignments (conditional branches may set different values):

response.success = True

Return expressions:

response

Direct calls to follow for delegated behavior: self.bridge.build_chosen_feedback, self.recent_request_texts.get, self.send_feedback.

hr_actuators: '/hr/actuators/set_control'

Type: SetActuatorsControl. Callback: self.set_control. Registration: head-ros-nodes/ros/hr_actuators/scripts/actuators_node.py:147.

Callback source: head-ros-nodes/ros/hr_actuators/scripts/actuators_node.py:157.

Request attributes read: actuators, control.

Response assignments (conditional branches may set different values):

response.success = error == ''
response.error = error

Return expressions:

response

Raised errors:

RuntimeError('actuators are not initialized')

Direct calls to follow for delegated behavior: RuntimeError, SetActuatorsControl.Response, int, list, self.actuators.set_control, self.expand_names.

ros_misc: DETECT_SERVICE

Type: StringTrigger. Callback: runtime.detect_service. Registration: head-ros-nodes/ros/ros_misc/nodes/lang_detector.py:25.

Callback source: head-ros-nodes/ros/ros_misc/src/ros_misc/lang_detector.py:722.

Request attributes read: data.

Response assignments (conditional branches may set different values):

response.success = result.success
response.message = result.language or ''

Return expressions:

response

Direct calls to follow for delegated behavior: self._update_recent_language, self.detect_text.

ros_feetech: '/hr/actuators/set_torque_enable'

Type: MotorValuesBool. Callback: self.torque_callback. Registration: head-ros-nodes/ros/ros_feetech/scripts/feetech_node.py:103.

Callback source: head-ros-nodes/ros/ros_feetech/scripts/feetech_node.py:217.

Request attributes read: names, values.

Return expressions:

response

Direct calls to follow for delegated behavior: bool, self._motors.get, self.set_torque_enable, zip.

ros_feetech: '/hr/actuators/set_torque_enable'

Type: MotorValuesBool. Callback: self.torque_callback. Registration: head-ros-nodes/ros/ros_feetech/scripts/feetech_node.py:109.

Callback source: head-ros-nodes/ros/ros_feetech/scripts/feetech_node.py:217.

Request attributes read: names, values.

Return expressions:

response

Direct calls to follow for delegated behavior: bool, self._motors.get, self.set_torque_enable, zip.

r2_perception: '/hr/perception/extract_face_embedding'

Type: ExtractFaceEmbedding. Callback: self._extract_service_callback. Registration: head-perception/ros2_ws/src/r2_perception/src/r2_perception/face_embedding_node.py:107.

Callback source: head-perception/ros2_ws/src/r2_perception/src/r2_perception/face_embedding_node.py:485.

Request attributes read: image_base64, options_json, request_id.

Response assignments (conditional branches may set different values):

response.ok = bool(result.get('accepted'))
response.error = '' if response.ok else str(result.get('reason', 'embedding_rejected'))
response.result_json = json.dumps(result, separators=(',', ':'))
response.ok = False
response.error = result['reason']
response.error = str(exc)
response.result_json = '{}'

Return expressions:

response

Direct calls to follow for delegated behavior: base64.b64decode, bool, cv2.imdecode, json.dumps, json.loads, np.frombuffer, options.get, result.get, result.update, self._embedding_from_pose, self._embedding_from_still_image, str, time.time.