coopihc.interactiontask.PipeTaskWrapper.PipeTaskWrapper

class PipeTaskWrapper(task, pipe)[source]

Bases: coopihc.interactiontask.InteractionTask.InteractionTask, abc.ABC

A Wrapper for tasks so that messages are passed through a pipe. Subclass this task to use tasks defined externally (e.g. that pass messages via websockets to a server which forwards the message to a task via a pipe)

Note

Need to explain interface here

Parameters

task (InteractionTask<coopihc.interactiontask.InteractionTask.InteractionTask) – task to wrap
pipe (subprocess.Pipe) – pipe

Methods

`base_on_assistant_action`	base assistant step
`base_on_user_action`	base user step
`finit`
`on_assistant_action`	Same as on_user_action
`on_user_action`	Transform user action into dictionnary with appropriate interface
`render`	Render the task on the main plot.
`reset`	Send reset dic over pipe
`update_state`	Remove the 'type' entry from the state dictionnary
`update_task_state`	Redefine this.
`update_user_state`	See update_task_state

Attributes

`assistant_action`	assistant action
`parameters`
`round_number`
`state`	The current state of the task.
`turn_number`	Turn number.
`user_action`	user action

property assistant_action

assistant action

The last action input by the assistant.

Returns: assistant action
Return type: State

base_on_assistant_action(*args, **kwargs)

base assistant step

Wraps the assistant defined on_assistant_action() method. For now does little but provide default values, may be useful later.

Returns: (task state, task reward, is_done flag, metadata):
Return type: tuple(State, float, boolean, dictionnary)

base_on_user_action(*args, **kwargs)

base user step

Wraps the user defined on_user_action() method. For now does little but provide default values, may be useful later.

Returns: (task state, task reward, is_done flag, metadata):
Return type: tuple(State, float, boolean, dictionnary)

on_assistant_action(*args, **kwargs)[source]

Same as on_user_action

Returns: (task state, task reward, is_done flag, {})
Return type: tuple(State, float, boolean, dictionnary)

on_user_action(*args, **kwargs)[source]

Transform user action into dictionnary with appropriate interface
Send message over pipe
Wait for pipe message
Update state and return

Returns: (task state, task reward, is_done flag, {})
Return type: tuple(State, float, boolean, dictionnary)

render(mode='text', ax_user=None, ax_assistant=None, ax_task=None)

Render the task on the main plot.

Parameters

mode – (str) text or plot
args – (list) list of axis in order axtask, axuser, axassistant

reset(dic=None)[source]

Send reset dic over pipe
Wait for pipe message
Update state and return

Note

verify the dic=None signature

Parameters: dic (dictionnary, optional) – reset dic, defaults to None
Returns: Task state
Return type: State

property state

The current state of the task.

Returns: task state
Return type: State

property turn_number

Turn number.

The turn number of the game

Returns: turn number
Return type: numpy.ndarray

update_state(state)[source]

Remove the ‘type’ entry from the state dictionnary

Parameters: state (dictionnary) – state received via pipe

abstract update_task_state(state)[source]

Redefine this. Example here

Parameters: state (dictionnary) – state received via pipe

abstract update_user_state(state)[source]

See update_task_state

Parameters: state (dictionnary) – state received via pipe

property user_action

user action

The last action input by the user.

Returns: user action
Return type: State