From efaa29eddc12188a0590c9f0eb04916175b7a7ff Mon Sep 17 00:00:00 2001 From: chrizog Date: Sat, 1 Jun 2024 12:04:47 +0200 Subject: [PATCH] Enable calling methods via UDP --- example_apps/addition_method_parameters.py | 6 +- example_apps/call_method_udp.py | 84 +++++++++ src/someipy/__init__.py | 4 +- src/someipy/_internal/someip_message.py | 3 + src/someipy/client_service_instance.py | 83 ++++++++- test_apps/CMakeLists.txt | 2 +- test_apps/call_method_udp/call_method_udp.cpp | 174 ++++++++++++++++++ test_apps/call_method_udp/start_app.sh | 6 + test_apps/call_method_udp/vsomeip-client.json | 45 +++++ 9 files changed, 398 insertions(+), 9 deletions(-) create mode 100644 example_apps/call_method_udp.py create mode 100644 test_apps/call_method_udp/call_method_udp.cpp create mode 100755 test_apps/call_method_udp/start_app.sh create mode 100644 test_apps/call_method_udp/vsomeip-client.json diff --git a/example_apps/addition_method_parameters.py b/example_apps/addition_method_parameters.py index bd924b4..a373a77 100644 --- a/example_apps/addition_method_parameters.py +++ b/example_apps/addition_method_parameters.py @@ -15,9 +15,9 @@ class Addends(SomeIpPayload): addend1: Sint16 addend2: Sint16 - def __init__(self): - self.addend1 = Sint16() - self.addend2 = Sint16() + def __init__(self, addend1: int = 0, addend2: int = 0): + self.addend1 = Sint16(addend1) + self.addend2 = Sint16(addend2) @dataclass class Sum(SomeIpPayload): diff --git a/example_apps/call_method_udp.py b/example_apps/call_method_udp.py new file mode 100644 index 0000000..a2a75bd --- /dev/null +++ b/example_apps/call_method_udp.py @@ -0,0 +1,84 @@ +import asyncio +import ipaddress +import logging + +from someipy import TransportLayerProtocol +from someipy.client_service_instance import MethodResult, construct_client_service_instance +from someipy.service import ServiceBuilder +from someipy.service_discovery import construct_service_discovery +from someipy.logging import set_someipy_log_level +from addition_method_parameters import Addends, Sum + +SD_MULTICAST_GROUP = "224.224.224.245" +SD_PORT = 30490 +INTERFACE_IP = "127.0.0.1" + +SAMPLE_SERVICE_ID = 0x1234 +SAMPLE_INSTANCE_ID = 0x5678 +SAMPLE_METHOD_ID = 0x0123 + +async def main(): + + # It's possible to configure the logging level of the someipy library, e.g. logging.INFO, logging.DEBUG, logging.WARN, .. + set_someipy_log_level(logging.DEBUG) + + # Since the construction of the class ServiceDiscoveryProtocol is not trivial and would require an async __init__ function + # use the construct_service_discovery function + # The local interface IP address needs to be passed so that the src-address of all SD UDP packets is correctly set + service_discovery = await construct_service_discovery(SD_MULTICAST_GROUP, SD_PORT, INTERFACE_IP) + + addition_service = ( + ServiceBuilder() + .with_service_id(SAMPLE_SERVICE_ID) + .with_major_version(1) + .build() + ) + + # For calling methods construct a ClientServiceInstance + client_instance_addition = await construct_client_service_instance( + service=addition_service, + instance_id=SAMPLE_INSTANCE_ID, + endpoint=(ipaddress.IPv4Address(INTERFACE_IP), 3002), + ttl=5, + sd_sender=service_discovery, + protocol=TransportLayerProtocol.UDP + ) + + # The service instance has to be attached always to the ServiceDiscoveryProtocol object, so that the service instance + # is notified by the ServiceDiscoveryProtocol about e.g. subscriptions or offers from other ECUs + service_discovery.attach(client_instance_addition) + + try: + while True: + + method_parameter = Addends(addend1=1, addend2=2) + method_success, method_result = await client_instance_addition.call_method(SAMPLE_METHOD_ID, method_parameter.serialize()) + if method_success == MethodResult.SUCCESS: + print(f"Received result for method: {' '.join(f'0x{b:02x}' for b in method_result)}") + try: + sum = Sum().deserialize(method_result) + print(f"Sum: {sum.value.value}") + except e: + print(f"Error during deserialization: {e}") + elif method_success == MethodResult.ERROR: + print("Method call failed..") + elif method_success == MethodResult.TIMEOUT: + print("Method call timed out..") + elif method_success == MethodResult.SERVICE_NOT_FOUND: + print("Service not yet available..") + + await asyncio.sleep(2) + except asyncio.CancelledError: + print("Shutdown..") + finally: + print("Service Discovery close..") + service_discovery.close() + + print("End main task..") + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + pass diff --git a/src/someipy/__init__.py b/src/someipy/__init__.py index e1fe4b0..85ad072 100644 --- a/src/someipy/__init__.py +++ b/src/someipy/__init__.py @@ -1,4 +1,6 @@ from ._internal.someip_sd_header import TransportLayerProtocol # noqa: F401 -from .server_service_instance import ServerServiceInstance, construct_server_service_instance # noqa: F401 from .service import Service, ServiceBuilder, EventGroup # noqa: F401 +from .server_service_instance import ServerServiceInstance, construct_server_service_instance # noqa: F401 +from .client_service_instance import ClientServiceInstance, construct_client_service_instance, MethodResult # noqa: F401 + from ._internal.someip_message import SomeIpMessage # noqa: F401 diff --git a/src/someipy/_internal/someip_message.py b/src/someipy/_internal/someip_message.py index 7cfd443..5cb6d52 100644 --- a/src/someipy/_internal/someip_message.py +++ b/src/someipy/_internal/someip_message.py @@ -5,3 +5,6 @@ class SomeIpMessage: header: SomeIpHeader payload: bytes + + def serialize(self) -> bytes: + return self.header.to_buffer() + self.payload diff --git a/src/someipy/client_service_instance.py b/src/someipy/client_service_instance.py index 46554f8..cf2ba67 100644 --- a/src/someipy/client_service_instance.py +++ b/src/someipy/client_service_instance.py @@ -1,7 +1,7 @@ import asyncio from enum import Enum import struct -from typing import Tuple, Callable, Set, List +from typing import Iterable, Tuple, Callable, Set, List from someipy import Service from someipy._internal.someip_sd_header import ( @@ -15,7 +15,7 @@ ServiceDiscoveryObserver, ServiceDiscoverySender, ) -from someipy._internal.utils import create_udp_socket, EndpointType +from someipy._internal.utils import create_udp_socket, EndpointType, endpoint_to_str_int_tuple from someipy._internal.logging import get_logger from someipy._internal.message_types import MessageType from someipy._internal.someip_endpoint import ( @@ -27,10 +27,25 @@ _logger_name = "client_service_instance" +class MethodResult(Enum): + SUCCESS = 0 + TIMEOUT = 1 + SERVICE_NOT_FOUND = 2 + ERROR = 3 + class ExpectedAck: def __init__(self, eventgroup_id: int) -> None: self.eventgroup_id = eventgroup_id +class FoundService: + service: SdService + + def __init__(self, service: SdService) -> None: + self.service = service + + def __eq__(self, __value: object) -> bool: + if isinstance(__value, FoundService): + return self.service == __value.service class ClientServiceInstance(ServiceDiscoveryObserver): _service: Service @@ -44,6 +59,8 @@ class ClientServiceInstance(ServiceDiscoveryObserver): _eventgroups_to_subscribe: Set[int] _expected_acks: List[ExpectedAck] _callback: Callable[[bytes], None] + _found_services: Iterable[FoundService] + _method_call_future: asyncio.Future def __init__( self, @@ -72,6 +89,10 @@ def __init__( self._tcp_connect_lock = asyncio.Lock() self._tcp_task = None + self._found_services = [] + self._method_call_future = None + + def register_callback(self, callback: Callable[[SomeIpMessage], None]) -> None: """ Register a callback function to be called when a SOME/IP event is received. @@ -85,6 +106,43 @@ def register_callback(self, callback: Callable[[SomeIpMessage], None]) -> None: """ self._callback = callback + async def call_method(self, method_id: int, payload: bytes) -> Tuple[MethodResult, bytes]: + + has_service = False + for s in self._found_services: + if s.service.service_id == self._service.id and s.service.instance_id == self._instance_id: + has_service = True + break + + if not has_service: + get_logger(_logger_name).warn(f"Do not execute method call. Service 0x{self._service.id:04X} with instance 0x{self._instance_id:04X} not found.") + return MethodResult.SERVICE_NOT_FOUND, b"" + + header = SomeIpHeader( + service_id=self._service.id, + method_id=method_id, + client_id=0x00, + session_id=0x00, + protocol_version=0x01, + interface_version=0x00, + message_type=MessageType.REQUEST.value, + return_code=0x00, + length=len(payload) + 8, + ) + someip_message = SomeIpMessage(header, payload) + + self._method_call_future = asyncio.get_running_loop().create_future() + self._someip_endpoint.sendto(someip_message.serialize(), endpoint_to_str_int_tuple(self._found_services[0].service.endpoint)) + + try: + await asyncio.wait_for(self._method_call_future, 1.0) + except asyncio.TimeoutError: + get_logger(_logger_name).error(f"Waiting on response for method call 0x{method_id:04X} timed out.") + return MethodResult.TIMEOUT, b"" + + return MethodResult.SUCCESS, self._method_call_future.result() + + def someip_message_received( self, someip_message: SomeIpMessage, addr: Tuple[str, int] ) -> None: @@ -96,6 +154,18 @@ def someip_message_received( ): if self._callback is not None: self._callback(someip_message) + + if (someip_message.header.message_type == MessageType.RESPONSE.value + and someip_message.header.return_code == 0x00): # E_OK + if self._method_call_future is not None: + self._method_call_future.set_result(someip_message.payload) + return + + if (someip_message.header.message_type == MessageType.ERROR.value + and someip_message.header.return_code == 0x01): # E_NOT_OK + if self._method_call_future is not None: + self._method_call_future.set_result(b"") + return def subscribe_eventgroup(self, eventgroup_id: int): """ @@ -140,8 +210,8 @@ def find_service_update(self): pass def offer_service_update(self, offered_service: SdService): - if len(self._eventgroups_to_subscribe) == 0: - return + #if len(self._eventgroups_to_subscribe) == 0: + # return if self._service.id != offered_service.service_id: return @@ -152,6 +222,11 @@ def offer_service_update(self, offered_service: SdService): offered_service.service_id == self._service.id and offered_service.instance_id == self._instance_id ): + # TODO: Check does not work, fix + if FoundService(offered_service) not in self._found_services: + self._found_services.append(FoundService(offered_service)) + + # Try to subscribe to requested event groups for eventgroup_to_subscribe in self._eventgroups_to_subscribe: ( session_id, diff --git a/test_apps/CMakeLists.txt b/test_apps/CMakeLists.txt index 0c19e81..e340d9a 100644 --- a/test_apps/CMakeLists.txt +++ b/test_apps/CMakeLists.txt @@ -28,6 +28,6 @@ create_test_target("send_events") create_test_target("receive_events_udp") create_test_target("receive_events_tcp") - create_test_target("offer_method_udp") create_test_target("offer_method_tcp") +create_test_target("call_method_udp") diff --git a/test_apps/call_method_udp/call_method_udp.cpp b/test_apps/call_method_udp/call_method_udp.cpp new file mode 100644 index 0000000..6e704af --- /dev/null +++ b/test_apps/call_method_udp/call_method_udp.cpp @@ -0,0 +1,174 @@ +#include +#include +#include +#include +#include +#include +#include + +#include + +#define SAMPLE_SERVICE_ID 0x1234 +#define SAMPLE_INSTANCE_ID 0x5678 +#define SAMPLE_METHOD_ID 0x0123 + + +class service_sample { +public: + service_sample(bool _use_static_routing) : + app_(vsomeip::runtime::get()->create_application("Hello")), + is_registered_(false), + use_static_routing_(_use_static_routing), + blocked_(false), + running_(true), + offer_thread_(std::bind(&service_sample::run, this)) { + } + + bool init() { + std::lock_guard its_lock(mutex_); + + if (!app_->init()) { + std::cerr << "Couldn't initialize application" << std::endl; + return false; + } + app_->register_state_handler( + std::bind(&service_sample::on_state, this, + std::placeholders::_1)); + app_->register_message_handler( + SAMPLE_SERVICE_ID, SAMPLE_INSTANCE_ID, SAMPLE_METHOD_ID, + std::bind(&service_sample::on_message, this, + std::placeholders::_1)); + + std::cout << "Static routing " << (use_static_routing_ ? "ON" : "OFF") + << std::endl; + return true; + } + + void start() { + app_->start(); + } + + void stop() { + running_ = false; + blocked_ = true; + app_->clear_all_handler(); + stop_offer(); + condition_.notify_one(); + if (std::this_thread::get_id() != offer_thread_.get_id()) { + if (offer_thread_.joinable()) { + offer_thread_.join(); + } + } else { + offer_thread_.detach(); + } + app_->stop(); + } + + void offer() { + app_->offer_service(SAMPLE_SERVICE_ID, SAMPLE_INSTANCE_ID); + } + + void stop_offer() { + app_->stop_offer_service(SAMPLE_SERVICE_ID, SAMPLE_INSTANCE_ID); + } + + void on_state(vsomeip::state_type_e _state) { + std::cout << "Application " << app_->get_name() << " is " + << (_state == vsomeip::state_type_e::ST_REGISTERED ? + "registered." : "deregistered.") + << std::endl; + + if (_state == vsomeip::state_type_e::ST_REGISTERED) { + if (!is_registered_) { + is_registered_ = true; + blocked_ = true; + condition_.notify_one(); + } + } else { + is_registered_ = false; + } + } + + void on_message(const std::shared_ptr &_request) { + std::cout << "Received a message with Client/Session [" + << std::setfill('0') << std::hex + << std::setw(4) << _request->get_client() << "/" + << std::setw(4) << _request->get_session() << "]" + << std::endl; + + std::shared_ptr its_response + = vsomeip::runtime::get()->create_response(_request); + + std::shared_ptr its_payload + = vsomeip::runtime::get()->create_payload(); + std::vector its_payload_data; + for (std::size_t i = 0; i < 4; ++i) + its_payload_data.push_back(vsomeip::byte_t(i % 256)); + its_payload->set_data(its_payload_data); + its_response->set_payload(its_payload); + + app_->send(its_response); + } + + void run() { + std::unique_lock its_lock(mutex_); + while (!blocked_) + condition_.wait(its_lock); + + bool is_offer(true); + + if (use_static_routing_) { + offer(); + while (running_); + } else { + while (running_) { + if (is_offer) + offer(); + else + stop_offer(); + + for (int i = 0; i < 10 && running_; i++) + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + is_offer = !is_offer; + } + } + } + +private: + std::shared_ptr app_; + bool is_registered_; + bool use_static_routing_; + + std::mutex mutex_; + std::condition_variable condition_; + bool blocked_; + bool running_; + + // blocked_ must be initialized before the thread is started. + std::thread offer_thread_; +}; + + + service_sample *its_sample_ptr(nullptr); + void handle_signal(int _signal) { + if (its_sample_ptr != nullptr && + (_signal == SIGINT || _signal == SIGTERM)) + its_sample_ptr->stop(); + } + + +int main(int argc, char **argv) { + bool use_static_routing(false); + service_sample its_sample(use_static_routing); + + its_sample_ptr = &its_sample; + signal(SIGINT, handle_signal); + signal(SIGTERM, handle_signal); + + if (its_sample.init()) { + its_sample.start(); + return 0; + } else { + return 1; + } +} diff --git a/test_apps/call_method_udp/start_app.sh b/test_apps/call_method_udp/start_app.sh new file mode 100755 index 0000000..16171a0 --- /dev/null +++ b/test_apps/call_method_udp/start_app.sh @@ -0,0 +1,6 @@ +#!/bin/bash + +script_dir="$(dirname "$0")" + +export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:~/projects/someip/vsomeip_install/lib/ +VSOMEIP_CONFIGURATION="${script_dir}/vsomeip-client.json" "${script_dir}/call_method_udp" \ No newline at end of file diff --git a/test_apps/call_method_udp/vsomeip-client.json b/test_apps/call_method_udp/vsomeip-client.json new file mode 100644 index 0000000..bc4117b --- /dev/null +++ b/test_apps/call_method_udp/vsomeip-client.json @@ -0,0 +1,45 @@ +{ + "unicast" : "127.0.0.2", + "logging" : + { + "level" : "verbose", + "console" : "true", + "file" : { "enable" : "false", "path" : "./vsomeip.log" }, + "dlt" : "false" + }, + "applications" : + [ + { + "name" : "Hello", + "id" : "0x1313" + } + ], + "services" : + [ + { + "service" : "0x1234", + "instance" : "0x5678", + "unreliable" : "30509", + "multicast" : + { + "address" : "224.225.226.234", + "port" : "32344" + } + } + ], + "routing" : "Hello", + "service-discovery" : + { + "enable" : "true", + "multicast" : "224.224.224.245", + "port" : "30490", + "protocol" : "udp", + "initial_delay_min" : "10", + "initial_delay_max" : "100", + "repetitions_base_delay" : "200", + "repetitions_max" : "3", + "ttl" : "3", + "cyclic_offer_delay" : "2000", + "request_response_delay" : "1500" + } +} \ No newline at end of file