gRPCProtocol

Full path: schola.core.protocols.protobuf.gRPC.gRPCProtocol

schola.core.protocols.protobuf.gRPC.gRPCProtocol

gRPCProtocol

gRPCProtocol(url, port=None, environment_start_timeout=45)

Bases: BaseRLProtocol, SocketProtocolMixin

Methods

Item	Description
init	—
close	Close the Unreal Connection.
connect_stubs	Connects the gRPC stubs to the Unreal Engine channel
get_definition	Get the environment definition from Unreal Engine.
on_close()	Close the Unreal Connection.
on_start()	Bind the tcp_socket
send_action_msg	Send actions to the environment and receive the next state.
send_reset_msg	Send a reset message to restart the environment.
send_startup_msg	Send the startup message with auto-reset configuration.
start	Open the Connection to Unreal Engine.

Attributes

Item	Description
address	Returns the address of the connection
channel_connected	Returns whether the connection is active or not
has_socket	Returns whether the connection is active or not
mixin_properties	Get mixin-specific properties.
properties	Get protocol-specific properties.

Parameters

url (str)

port (int)

environment_start_timeout (int)

init

__init__(url, port=None, environment_start_timeout=45)

Parameters

url (str)

port (int)

environment_start_timeout (int)

---

channel_connected

channel_connected: bool

Returns whether the connection is active or not

---

close

close()

Close the Unreal Connection. Method must be safe to call multiple times.

Returns

None

---

connect_stubs

connect_stubs(*stubs)

Connects the gRPC stubs to the Unreal Engine channel

Parameters

*stubs (List["grpc.Stub"]) : The gRPC stubs to connect to the Unreal Engine channel

Returns

List[Stub]

---

get_definition

get_definition()

Get the environment definition from Unreal Engine.

Returns

A tuple containing:

• List of agent IDs per environment
• List of agent groups per environment (used for grouping agents)
• Observation spaces for each environment and agent
• Action spaces for each environment and agent

Return type: Tuple[List[List[str]], List[Dict[int, str]], Dict[int, Dict[str, gym.Space]], Dict[int, Dict[str, gym.Space]]]

---

properties

properties: Dict[str, Any]

Get protocol-specific properties.

---

send_action_msg

send_action_msg(actions, action_space)

Send actions to the environment and receive the next state.

Parameters

actions (Dict[int, Dict[str, Any]]) : Actions to take, indexed by environment ID and agent ID.

action_space (Dict[str, gym.Space]) : The action spaces used to serialize the actions.

Returns

A tuple containing:

• Observations for each environment
• Rewards for each environment
• Termination flags for each environment
• Truncation flags for each environment
• Info dicts for each environment
• Initial observations if auto-reset occurred
• Initial info dicts if auto-reset occurred

Return type: Tuple[List[Dict[str,Any]], List[float], List[Dict[str,bool]], List[Dict[str,bool]], List[Dict[str,str]], Dict[int,Dict[str, Any]], Dict[int,Dict[str, str]]]

---

send_reset_msg

send_reset_msg(seeds=None, options=None)

Send a reset message to restart the environment.

Parameters

seeds (List, optional) : List of random seeds for each environment.

options (List, optional) : List of reset options for each environment.

Returns

A tuple containing:

• List of initial observations for each environment
• List of initial info dicts for each environment

Return type: Tuple[List[Dict[str, Any]], List[Dict[str, Dict[str,str]]]]

---

send_startup_msg

send_startup_msg(auto_reset_type=<AutoresetMode SAME_STEP>)

Send the startup message with auto-reset configuration.

Parameters

auto_reset_type (AutoResetType, default=AutoResetType.SAME_STEP) : The type of auto-reset behavior to use when episodes end.

---

start

start()

Open the Connection to Unreal Engine.

Returns

None

---