|
|
|
|
|
|
assert la[i] == lb[i] |
|
|
|
|
|
|
|
|
|
|
|
def construct_fake_buffer(): |
|
|
|
def construct_fake_processing_buffer(): |
|
|
|
b["vector_observation"].append( |
|
|
|
b[fake_agent_id]["vector_observation"].append( |
|
|
|
[ |
|
|
|
100 * fake_agent_id + 10 * step + 1, |
|
|
|
100 * fake_agent_id + 10 * step + 2, |
|
|
|
|
|
|
b["action"].append( |
|
|
|
b[fake_agent_id]["action"].append( |
|
|
|
[ |
|
|
|
100 * fake_agent_id + 10 * step + 4, |
|
|
|
100 * fake_agent_id + 10 * step + 5, |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_buffer(): |
|
|
|
b = construct_fake_buffer() |
|
|
|
b = construct_fake_processing_buffer() |
|
|
|
a = b[1]["vector_observation"].get_batch( |
|
|
|
batch_size=2, training_length=1, sequential=True |
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_buffer_sample(): |
|
|
|
b = construct_fake_buffer() |
|
|
|
b = construct_fake_processing_buffer() |
|
|
|
update_buffer = AgentBuffer() |
|
|
|
b.append_update_buffer(update_buffer, 3, batch_size=None, training_length=2) |
|
|
|
b.append_update_buffer(update_buffer, 2, batch_size=None, training_length=2) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_buffer_truncate(): |
|
|
|
b = construct_fake_buffer() |
|
|
|
b = construct_fake_processing_buffer() |
|
|
|
assert len(b.update_buffer["action"]) == 2 |
|
|
|
assert len(update_buffer["action"]) == 2 |
|
|
|
|
|
|
|
b.append_update_buffer(update_buffer, 3, batch_size=None, training_length=2) |
|
|
|
b.append_update_buffer(update_buffer, 2, batch_size=None, training_length=2) |
|
|
|