Data Composition

Data composition classes.

class ComposedReward(*rewarders: GlobalReward)[source]

Bases: GlobalReward

Rewarder for composed data types.

This type can be automatically constructed by passing a tuple of rewarders to the environment constructor’s reward argument.

Parameters:: rewarders (GlobalReward) – Global rewarders to compose.

data_store_type: alias of ComposedDataStore

reset_pre_sim_init() → None[source]

Handle resetting for all rewarders.

Return type:: None

reset_post_sim_init() → None[source]

Handle resetting for all rewarders.

Return type:: None

reset_overwrite_previous() → None[source]

Handle resetting for all rewarders.

Return type:: None

link_scenario(scenario: Scenario) → None[source]

Link every component to the shared scenario.

Parameters:: scenario (Scenario)
Return type:: None

initial_data(satellite: Satellite) → ComposedData[source]

Furnish every component’s initial data in a fixed order.

Parameters:: satellite (Satellite)
Return type:: ComposedData

create_data_store(satellite: Satellite) → None[source]

Create a CompositeDataStore for a satellite.

Parameters:: satellite (Satellite)
Return type:: None

calculate_reward(new_data_dict: dict[str, ComposedData]) → dict[str, float][source]

Calculate reward for each data type and combine them.

Parameters:: new_data_dict (dict[str, ComposedData])
Return type:: dict[str, float]

reward(new_data_dict: dict[str, ComposedData]) → dict[str, float][source]

Return combined reward calculation and update data.

Parameters:: new_data_dict (dict[str, ComposedData])
Return type:: dict[str, float]

is_truncated(satellite: Satellite) → bool[source]

Check if the episode is truncated by any rewarder.

Parameters:: satellite (Satellite)
Return type:: bool

is_terminated(satellite) → bool[source]

Check if the episode is terminated by any rewarder.

Return type:: bool

class ComposedDataStore(satellite: Satellite, *data_store_types: type[DataStore], initial_data: ComposedData | None = None, data_store_kwargs: list | None = None)[source]

Bases: DataStore

DataStore for composed data types.

Parameters:

satellite (Satellite) – Satellite which data is being stored for.
data_store_types (type[DataStore]) – DataStore types to compose.
initial_data (ComposedData | None) – Initial data to start the store with. Usually comes from initial_data.
data_store_kwargs (list | None) – List of data_store kwargs matching data_store_types.

data_type: alias of ComposedData

get_log_state() → list[source]

Pull information used in determining current data contribution.

Return type:: list

compare_log_states(prev_state: list, new_state: list) → Data[source]

Generate a unit of composed data based on previous step and current step logs.

Parameters:

prev_state (list)
new_state (list)

Return type:

Data

update_from_logs() → Data[source]

Update the data store based on collected information.

Return type:: Data

update_with_communicated_data() → None[source]

Update the data store based on collected information from other satellites.

Return type:: None

class ComposedData(*data: Data)[source]

Bases: Data

Data for composed data types.

Parameters:: data (Data) – Data types to compose.

class MixedReward(rewarders: Mapping[str, GlobalReward], satellite_mapping: Mapping[str, Sequence[str]], scenario_mapping: Mapping[str, str] | None = None)[source]

Bases: GlobalReward

Initialize explicit channel or scenario bindings.

Parameters:

rewarders (Mapping[str, GlobalReward])
satellite_mapping (Mapping[str, Sequence[str]])
scenario_mapping (Mapping[str, str] | None)

data_store_type: alias of MixedDataStore

link_scenario(scenario: Scenario) → None[source]

Bind reward channels to the selected scenario instances.

Parameters:: scenario (Scenario)
Return type:: None

pass_data() → None[source]

Synchronize component data with the parent channel mapping.

Return type:: None

reset_overwrite_previous() → None[source]

Clear episode state and reset each component once.

Return type:: None

reset_pre_sim_init() → None[source]

Prepare each component before simulator construction.

Return type:: None

reset_during_sim_init() → None[source]

Prepare each component during simulator construction.

Return type:: None

reset_post_sim_init() → None[source]

Finalize component setup after simulator initialization.

Return type:: None

initial_data(satellite: Satellite) → MixedData[source]

Provide initial knowledge for the satellite’s assigned channels.

Parameters:: satellite (Satellite)
Return type:: MixedData

create_data_store(satellite: Satellite) → None[source]

Create assigned stores using each rewarder’s setup hooks.

Parameters:: satellite (Satellite)
Return type:: None

calculate_reward(new_data_dict: dict[str, MixedData]) → dict[str, float][source]

Route local deltas and sum rewards against previous global data.

Parameters:: new_data_dict (dict[str, MixedData])
Return type:: dict[str, float]

reward(new_data_dict: dict[str, MixedData]) → dict[str, float][source]

Calculate rewards, merge global deltas, and synchronize child data.

Parameters:: new_data_dict (dict[str, MixedData])
Return type:: dict[str, float]

is_truncated(satellite: Satellite) → bool[source]

Check only rewarders assigned to this satellite.

Parameters:: satellite (Satellite)
Return type:: bool

is_terminated(satellite: Satellite) → bool[source]

Check only rewarders assigned to this satellite.

Parameters:: satellite (Satellite)
Return type:: bool

class MixedDataStore(satellite: Satellite, data_stores: Mapping[str, DataStore], initial_data: MixedData | None = None)[source]

Bases: DataStore

Initialize explicit channel or scenario bindings.

Parameters:

satellite (Satellite)
data_stores (Mapping[str, DataStore])
initial_data (MixedData | None)

data_type: alias of MixedData

pass_data() → None[source]

Synchronize component data with the parent channel mapping.

Return type:: None

get_log_state() → dict[source]

Collect log states from locally active channels.

Return type:: dict

compare_log_states(prev_state: dict, new_state: dict) → MixedData[source]

Generate deltas only for locally active channels.

Parameters:

prev_state (dict)
new_state (dict)

Return type:

MixedData

update_from_logs() → MixedData[source]

Update local knowledge and synchronize component stores.

Return type:: MixedData

update_with_communicated_data() → None[source]

Merge received knowledge and synchronize local stores.

Return type:: None

class MixedData(data: Mapping[str, Data] | None = None)[source]

Bases: Data

Copy the channel mapping; values follow the Data copying contract.

Parameters:: data (Mapping[str, Data] | None)