remotivelabs.topology.namespaces.some_ip

class SomeIPNamespace(remotivelabs.topology.namespaces.namespace.Namespace[typing.Union[remotivelabs.topology.namespaces.filters.SomeIPRequestFilter, remotivelabs.topology.namespaces.filters.SomeIPEventFilter], typing.Union[typing.Callable[[remotivelabs.topology.namespaces.some_ip.request.SomeIPRequest], typing.Awaitable[typing.Union[remotivelabs.topology.namespaces.some_ip.response.SomeIPResponse, remotivelabs.topology.namespaces.some_ip.response.SomeIPError, NoneType]]], typing.Callable[[remotivelabs.topology.namespaces.some_ip.event.SomeIPEvent], typing.Awaitable[NoneType]]]]):

SomeIP provides a client-side interface for sending SOME/IP requests and handling responses within a specific namespace using a broker client.

The client maintains a session ID counter and ensures that the namespace is of type 'someip' before issuing requests. To validate and activate the namespace, this class must be used as an async context manager.

SomeIPNamespace( name: str, broker_client: remotivelabs.broker.BrokerClient, client_id: int, decode_named_values: bool = False)

Initialize the SomeIP namespace client

Arguments:
  • name: The namespace name to operate in.
  • broker_client: The client used to communicate with the broker.
  • client_id: The SOME/IP client ID used for requests.
  • decode_named_values: True will decode named values to str.
Note:

Use together with a BehavioralModel or start the instance using a context manager:

async with SomeIPNamespace(...) as namespace:
    ...
async def open(self) -> typing_extensions.Self:

Opens the SOME/IP namespace and validates that the namespace is of the correct type. This is an idempotent operation - calling it multiple times has no additional effect.

Returns:

The namespace

Raises:
  • ValueError: If the namespace is not of type 'someip'.
async def request( self, req: SomeIPRequest) -> _asyncio.Task[SomeIPResponse | SomeIPError | None]:

Send a SOME/IP request and return an asyncio Task that resolves to the response or error.

Arguments:
  • req: A SomeIPRequest instance specifying service, method, type, and parameters.
Returns:

asyncio.Task that resolves to SomeIPResponse, SomeIPError, or None if no return is expected.

Raises:
  • ValueError: If a response is expected but the response frame is missing.
async def notify( self, event: SomeIPEvent) -> None:

Emit a SOME/IP event

Arguments:
  • event: The SOME/IP event
async def subscribe( self, *events: tuple[str, str], on_change: bool = False) -> AsyncIterator[SomeIPEvent]:

Subscribes to a stream of SOME/IP events.

Arguments:
  • *events: One or more tuples, each containing event name and service name of events to subscribe to.
  • on_change: If True, only yield updates when signal values change.
Returns:

An asynchronous iterator with SomeIPEvent.

def create_input_handler( self, filters: Sequence[remotivelabs.topology.namespaces.filters.SomeIPRequestFilter | remotivelabs.topology.namespaces.filters.SomeIPEventFilter], callback: Union[Callable[[SomeIPRequest], Awaitable[Union[SomeIPResponse, SomeIPError, NoneType]]], Callable[[SomeIPEvent], Awaitable[NoneType]]]) -> tuple[str, remotivelabs.topology.namespaces.input_handlers.InputHandler]:

Creates a list of input handlers for the given namespace to be used with a BehavioralModel.

Each handler defines a filter and a corresponding async callback for processing matching requests/events.

Arguments:
  • - filters: A sequence of SomeIPRequestFilter or SomeIPEventFilter objects to select relevant requests or events
  • - callback: An async callback function that receives and processes a request or event.
Inherited Members
remotivelabs.topology.namespaces.namespace.Namespace
name
close
@dataclass(frozen=True)
class SomeIPRequest:

Represents a SOME/IP request

Attributes:
  • name: The name of the request.
  • service_instance_name: The name of the service associated with the request.
  • raw: The raw data to be sent with the request. If non-empty, it takes priority over parameters.
  • parameters: A dictionary of key-value pairs representing decoded request data. Note: str is only supported for named values (e.g., Enums).
Note:

When sending a request, if raw is non-empty, it overrides the contents of parameters.

SomeIPRequest( name: str, service_instance_name: str, message_type: RequestType, raw: bytes = b'', parameters: dict[str, typing.Union[int, float, bytes, str]] = <factory>)
name: str
service_instance_name: str
message_type: RequestType
raw: bytes = b''
parameters: dict[str, typing.Union[int, float, bytes, str]]
@dataclass(frozen=True)
class SomeIPRequestReturn(remotivelabs.topology.namespaces.some_ip.SomeIPRequest):
SomeIPRequestReturn( name: str, service_instance_name: str, raw: bytes = b'', parameters: dict[str, typing.Union[int, float, bytes, str]] = <factory>)
message_type: RequestType = <RequestType.REQUEST: 0>
@dataclass(frozen=True)
class SomeIPRequestNoReturn(remotivelabs.topology.namespaces.some_ip.SomeIPRequest):
SomeIPRequestNoReturn( name: str, service_instance_name: str, raw: bytes = b'', parameters: dict[str, typing.Union[int, float, bytes, str]] = <factory>)
@dataclass
class SomeIPResponse:

Represents a SOME/IP response

Attributes:
  • raw: The raw data received in the response. If non-empty, it takes priority over parameters.
  • parameters: A dictionary of key-value pairs representing decoded response data. Note: str is only supported for named values (e.g., Enums).
Note:

When processing a response, if raw is non-empty, it overrides the contents of parameters.

SomeIPResponse( raw: bytes = b'', parameters: dict[str, typing.Union[int, float, bytes, str]] = <factory>)
raw: bytes = b''
parameters: dict[str, typing.Union[int, float, bytes, str]]
@dataclass
class SomeIPError:

Represents a SOME/IP error response

Attributes:
  • return_code: The return code of the response.
SomeIPError(return_code: int | str)
return_code: int | str
@dataclass
class SomeIPEvent:

Represents a SOME/IP event.

Attributes:
  • name: The name of the event.
  • service_instance_name: The name of the service associated with the event.
  • raw: Raw bytes of the event payload. If non-empty, it takes precedence over parameters when emitting an event.
  • parameters: A dictionary of key-value pairs representing decoded event data. Note: str is only supported for named values (e.g., Enums).
Note:

When handling the event, if raw is non-empty, it overrides the contents of parameters.

SomeIPEvent( name: str, service_instance_name: str, raw: bytes = b'', parameters: dict[str, typing.Union[int, float, bytes, str]] = <factory>)
name: str
service_instance_name: str
raw: bytes = b''
parameters: dict[str, typing.Union[int, float, bytes, str]]
class RequestType(enum.IntEnum):
REQUEST = <RequestType.REQUEST: 0>
REQUEST_NO_RETURN = <RequestType.REQUEST_NO_RETURN: 1>
class ReturnCode(enum.IntEnum):
E_OK = <ReturnCode.E_OK: 0>
E_NOT_OK = <ReturnCode.E_NOT_OK: 1>
E_UNKNOWN_SERVICE = <ReturnCode.E_UNKNOWN_SERVICE: 2>
E_UNKNOWN_METHOD = <ReturnCode.E_UNKNOWN_METHOD: 3>
E_NOT_READY = <ReturnCode.E_NOT_READY: 4>
E_NOT_REACHABLE = <ReturnCode.E_NOT_REACHABLE: 5>
E_TIMEOUT = <ReturnCode.E_TIMEOUT: 6>
E_WRONG_PROTOCOL_VERSION = <ReturnCode.E_WRONG_PROTOCOL_VERSION: 7>
E_WRONG_INTERFACE_VERSION = <ReturnCode.E_WRONG_INTERFACE_VERSION: 8>
E_MALFORMED_MESSAGE = <ReturnCode.E_MALFORMED_MESSAGE: 9>
E_WRONG_MESSAGE_TYPE = <ReturnCode.E_WRONG_MESSAGE_TYPE: 10>
class ErrorReturnCode(enum.IntEnum):
E_NOT_OK = <ErrorReturnCode.E_NOT_OK: 1>
E_UNKNOWN_SERVICE = <ErrorReturnCode.E_UNKNOWN_SERVICE: 2>
E_UNKNOWN_METHOD = <ErrorReturnCode.E_UNKNOWN_METHOD: 3>
E_NOT_READY = <ErrorReturnCode.E_NOT_READY: 4>
E_NOT_REACHABLE = <ErrorReturnCode.E_NOT_REACHABLE: 5>
E_TIMEOUT = <ErrorReturnCode.E_TIMEOUT: 6>
E_WRONG_PROTOCOL_VERSION = <ErrorReturnCode.E_WRONG_PROTOCOL_VERSION: 7>
E_WRONG_INTERFACE_VERSION = <ErrorReturnCode.E_WRONG_INTERFACE_VERSION: 8>
E_MALFORMED_MESSAGE = <ErrorReturnCode.E_MALFORMED_MESSAGE: 9>
E_WRONG_MESSAGE_TYPE = <ErrorReturnCode.E_WRONG_MESSAGE_TYPE: 10>
ServiceName = <class 'str'>