Skip to content

Documentation of experiment server API

Public API

experiment_server.Experiment

Experiment(config_file: str, default_participant_index: int = 1)

Load and manage an experiment configuration file and participant states.

Initialize experiment state and watch the configuration file for changes.

PARAMETER DESCRIPTION

config_file

Path to the TOML configuration file.

TYPE: str

default_participant_index

Default 1-based index used when none is provided.

TYPE: int DEFAULT: 1

global_state instance-attribute

global_state: Dict[int, ParticipantState] = {}

default_participant_index instance-attribute

default_participant_index = default_participant_index

watchdog instance-attribute

watchdog = FileModifiedWatcher(config_file, _config_file_modified_callback)

config_file instance-attribute

config_file = config_file

get_next_participant

get_next_participant() -> int

Allocate and return the next participant index (max existing + 1).

add_participant_index

add_participant_index(participant_index) -> bool

Add a participant by index.

Returns True if added, False if the index already exists.

get_participant_state

get_participant_state(participant_index) -> ParticipantState

Return the ParticipantState for the given index.

If participant_index is None, uses the default. Raises ExperimentServerExcetion if not present.

get_state

get_state(participant_index: int | None = None) -> bool

Return whether the participant is active.

move_to_next

move_to_next(participant_index: int | None = None) -> str

Advance the participant to the next block and return the new block_name.

get_config

get_config(participant_index: int | None = None) -> Union[Dict[str, Any], None]

Return the current block's config for the participant, or None if experiment not started or finished.

reset_participant

reset_participant(participant_index: int | None = None) -> bool

Reload the participant's configuration from file and replace their stored config.

get_blocks_count

get_blocks_count(participant_index: int | None = None) -> int

Return the number of blocks for the participant.

get_all_configs

get_all_configs(participant_index: int | None = None) -> List[dict]

Return the list of all block 'config' dicts for the participant in order.

move_to_block

move_to_block(block_id: int, participant_index: int | None = None) -> str

Move the participant pointer to a specific block index and return its block_name.

block_id must be an integer.

move_all_to_block

move_all_to_block(block_id: int) -> str

Move every participant's pointer to block_id.

Returns the block_name of the first participant after the move.

experiment_server.Client

Client(server_host: str = '127.0.0.1', server_port: Union[str, int] = '5000')

HTTP client for the Experiment Server API.

Usage: - Instantiate with optional server_host and server_port. - Methods operate on a per-participant basis when applicable (provide participant_index or omit to use the server's default participant).

Return value for API methods

  • All public API methods return a tuple (success: bool, data: dict|str). On success data is the parsed JSON response (or "" if empty). On failure success is False and data contains an error message.

Main methods

  • move_to_next(participant_index=None)
  • get_config(participant_index=None)
  • server_is_active(participant_index=None)
  • get_blocks_count(participant_index=None)
  • get_all_configs(participant_index=None)
  • move_to_block(block_id, participant_index=None)
  • new_participant()
  • add_participant(participant_index)
  • shutdown()
PARAMETER DESCRIPTION

server_host

Hostname or IP of the server (default "127.0.0.1").

TYPE: str DEFAULT: '127.0.0.1'

server_port

Port of the server (default "5000").

TYPE: str | int DEFAULT: '5000'

move_to_next

move_to_next(participant_index: int | None = None) -> Tuple[bool, dict]

Moves the pointer to the current block to the next block for participant_index. if participant_index is None, seld.default_participant_index is used.

get_config

get_config(participant_index: int | None = None) -> Tuple[bool, dict]

Return the config of the current block for participant_index. if participant_index is None, seld.default_participant_index is used. If the experiment has not started (move_to_next has not been called atleast once), this will return None.

server_is_active

server_is_active(participant_index: int | None = None) -> Tuple[bool, dict]

Returns the status for participant-id, if participant-id is not provided, will return the status of the default participant. Will be false if the participant was just initialized or the participant has gone through all blocks. To initialize the participant's status (or move to a given block), use the move-to-next or move-to-block

get_blocks_count

get_blocks_count(participant_index: int | None = None) -> Tuple[bool, dict]

Return the number of blocks in the configuration loaded. For a given config, the blocks-count will be the same for all participants.

get_all_configs

get_all_configs(participant_index: int | None = None) -> Tuple[bool, dict]

Returns all the configs as a list for the participant_index, if participant_index is not provided, returns the configs for the default participant. This is akin having all the results from calling config for each block in one list.

move_to_block

move_to_block(block_id: int, participant_index: int | None = None) -> Tuple[bool, dict]

Move participant-id to the block number indicated by block_id, if participant_index is not provided, move the default participant to the block number indicated by block_id. If the participant was not initialized (active is false), will make be marked as active (active will be set to true). Will fail if the block_id is below 0 or above the length of the config.

new_participant

new_participant() -> Tuple[bool, dict]

Adds a new participant and returns the new participant_index. The new participant_index will be the largest current participant_index +1.

add_participant

add_participant(participant_index: int) -> Tuple[bool, dict]

Add a new participant with participant_index. If there is already a participant with the participant_index, this will fail.

shutdown

shutdown() -> Tuple[bool, dict]

Shuts down the server.

experiment_server.server_process

server_process(config_file, default_participant_index=None, host='127.0.0.1', port='5000')

Returns a Process object which can be used to launch experiment_server. For example:

p = server_process(config_file=config_file)
p.start()

Full API

experiment_server._api

ParticipantState

ParticipantState(config, participant_index, active)

Track a single participant's progress through an experiment configuration.

PARAMETER DESCRIPTION
config

Ordered list of block dictionaries for this participant.

TYPE: List[Dict[str, Any]]

participant_index

1-based participant index.

TYPE: int

active

True if participant is currently between START and END (inclusive of blocks).

TYPE: bool

participant_index instance-attribute
participant_index = participant_index
config instance-attribute
config = config
active instance-attribute
active = active
block_id property writable
block_id: int

Current zero-based index of the active block, -1 if before START, len(config) if after END.

block property
block: Dict[str, Any] | None

Return the current block dict or None if before START or after END.

block_name property
block_name: str

Human-readable block position: 'START', 'END' or the current block name.

move_to_next_block
move_to_next_block() -> str

Advance to the next block and return its block_name.

status_string
status_string() -> str

Single-line status summary suitable for logs or simple UIs.

_generate_config_json

_generate_config_json(config_file: Union[str, Path], participant_indices: Iterable[int], out_dir: Union[str, Path, None] = None) -> None

Emit the resolved JSON configuration for the given participant indices.

If out_dir is provided, writes one file per participant named "-participant_.json". If out_dir is None, writes each participant's JSON to stdout (one line per participant).

PARAMETER DESCRIPTION
config_file

Path to the TOML configuration file.

TYPE: Union[str, Path]

participant_indices

Iterable of 1-based participant indices to generate.

TYPE: Iterable[int]

out_dir

Optional directory to write files into. Created if missing.

TYPE: Union[str, Path, None] DEFAULT: None

experiment_server._process_config

ChoicesFunction

ChoicesFunction(args, params)

Wrapper for random.choices function call. args will be passed to random.choices. If params has unique whose value is True, will ensure no duplicate values seen in any of the choices call.

args instance-attribute
args = args
unique instance-attribute
unique = False
params instance-attribute
params = params
previous_choices instance-attribute
previous_choices = []
__call__
__call__(args, params) -> Any

process_config_file

process_config_file(f: Union[str, Path], participant_index: int, supress_message: bool = False) -> List[Dict[str, Any]]

Load and resolve an experiment configuration file for a single participant.

This function reads a configuration file (currently only TOML files are supported), applies configuration-level variables and ordering rules, resolves inheritance (extends), evaluates configured function calls, and returns the list of block configurations for the specified participant index in the order they will be executed.

PARAMETER DESCRIPTION
f

Path or string filename pointing to the configuration file. Files must have a ".toml" suffix; other formats will raise an ExperimentServerExcetion.

TYPE: Union[str, Path]

participant_index

1-based index of the participant for which to construct the configuration. Must be greater than 0; otherwise an ExperimentServerConfigurationExcetion is raised. This index is used where deterministic rotations (e.g. latin-square or per-participant assignment) are requested and is added to any configured random seed to provide reproducible per-participant randomness.

TYPE: int

supress_message

If False (default), a JSON summary of the resolved configuration for the participant is logged. If True, the info log is suppressed.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
List[Dict[str, Any]]

List[Dict[str, Any]]: A list of resolved block configuration dictionaries for the participant in execution order. Each block's "config" dict will include the keys "participant_index", "name", and "block_id" (the zero-based index within the returned list).

RAISES DESCRIPTION
ExperimentServerConfigurationExcetion

If the participant_index is invalid, required variables are missing, inheritance references are invalid, or other configuration validation errors occur.

ExperimentServerExcetion

If the file type is unsupported (non-.toml).

TomlDecodeError or IOError

If the TOML file cannot be read or parsed.

Notes

The function delegates group and per-participant ordering logic to the participant ordering utilities; randomization behavior depends on the provided configuration's "random_seed" combined with participant_index and on the global random state. The function performs variable replacement, resolves "extends" inheritance (merging dictionaries), and evaluates configured function-calls such as choices(...) before returning the final block list.

See also experiment_server._participant_ordering.construct_participant_condition.

resolve_extends

resolve_extends(configs)

Resolve 'extends' inheritance for a list of block configuration dicts.

PARAMETER DESCRIPTION
configs

List of dicts each containing a unique "name". Dicts may include an "extends" key referencing another config's name.

TYPE: list

RETURNS DESCRIPTION
list

The provided list with inheritance applied so each config includes merged

values from its ancestor(s). The input list is mutated in-place.

RAISES DESCRIPTION
ExperimentServerConfigurationExcetion

If an "extends" reference names a non-existent config.

_replace_variables

_replace_variables(config: Union[Dict[str, Any], List[Any]], variabels: Dict[str, Any]) -> Union[Dict[str, Any], List[Any]]

Recursively substitute variable placeholders in a config.

Any string beginning with '$' is replaced by the value from variabels using the name after the '$'. Nested dicts and lists are traversed and preserved in structure.

PARAMETER DESCRIPTION
config

A dict or list tree containing values to resolve.

TYPE: Union[Dict[str, Any], List[Any]]

variabels

Mapping of variable names (without '$') to replacement values.

TYPE: Dict[str, Any]

RETURNS DESCRIPTION
Union[Dict[str, Any], List[Any]]

A new dict or list with all '$'-prefixed variables substituted.

RAISES DESCRIPTION
ExperimentServerConfigurationExcetion

if a referenced variable is not found.

resolve_function_calls

resolve_function_calls(configs: list) -> list

Check all function calls and replace the values with the result of the function calls.

verify_config

verify_config(f: Union[str, Path], test_func: Callable[[List[Dict[str, Any]]], Tuple[bool, str]] = None) -> bool

Verify an experiment TOML config by constructing participant orders for participants 1–5.

For each participant index 1..5 this loads the resolved configuration via process_config_file. If provided, test_func is called with the resolved blocks and must return (True, reason) on success; otherwise an assertion or exception is raised. On success a summary table is logged and the function returns True.

PARAMETER DESCRIPTION
f

Path or filename of the TOML configuration file.

TYPE: Union[str, Path]

test_func

Optional callable receiving the resolved blocks and returning a (bool, str) tuple indicating success and an optional reason.

TYPE: Callable[[List[Dict[str, Any]]], Tuple[bool, str]] DEFAULT: None

RETURNS DESCRIPTION
bool

True if verification completed successfully for all checked participants.

experiment_server._participant_ordering

construct_participant_condition

construct_participant_condition(config: List[Dict], participant_index: int, order: Union[dict, list], init_block_names: Union[dict, list], final_block_names: Union[dict, list], within_groups_strategy: Union[str, None] = None, groups_strategy: Union[str, None] = None, init_blocks_strategy: Union[str, None] = None, final_blocks_strategy: Union[str, None] = None) -> List

Construct the per-participant ordered list of block configurations based on a global experiment configuration, ordering specification, and a participant index.

The function maps block names defined in config to a concrete ordering for a single participant, applying ordering strategies at three levels: - groups (ordering of groups of blocks) - within_groups (ordering of elements inside each group) - init_blocks and final_blocks (ordering of initial and final blocks appended before/after main groups)

Behavior details and constraints
  • Names in order, init_block_names, and final_block_names are validated to be strings and must match unique names in config. Duplicate names in config cause an error.
  • When groups_strategy or within_groups_strategy is "randomize", Python's random.shuffle is used (non-deterministic unless the caller seeds the RNG).
  • When "latin_square" is requested for groups or within-group ordering, a balanced Latin square generator is used and participant_index selects the row; latin-square requires equal-sized values where appropriate (e.g., all groups must have the same size when using within-group latin-square).
  • When dictionary mappings are provided for order/init/final blocks, keys are expected to be 1-based consecutive integer indices (or strings coercible to those integers). For dicts, the corresponding strategy argument must be "as_is".
Notes

The function does not modify the input config objects beyond coercing the "name" field to str. Deterministic rotation behavior for latin-square and dict-based selection is 1-based and uses (participant_index - 1) modulo the appropriate period.

PARAMETER DESCRIPTION
config

A list of block configuration dictionaries. Each dict must contain a

TYPE: List[Dict]

participant_index

1-based index of the participant used to select deterministic

TYPE: int

order

Specification of the main experimental order. Two forms are supported:

  • list of groups: e.g. [["A","B"], ["C","D","E"]] where each inner list is a group of block names. If a flat list of strings is provided (["A","B","C"]) it will be treated as a single group.
  • dict keyed by participant index (string or int keys) mapping to a group-list for that participant. When a dict is provided, the groups_strategy must be "as_is" (per-participant assignment by key).

All block names in the resolved order must be strings and must exist in config.

TYPE: Union[dict, list]

init_block_names

Names of blocks to prepend for the participant. Accepts a list of block names (strings) or a dictionary keyed by participant indices (1-based) to select a list for the participant. When a dict is supplied, the strategy for init blocks must be "as_is" (per-participant selection).

TYPE: Union[dict, list]

final_block_names

Names of blocks to append for the participant. Accepts same forms and constraints as init_block_names.

TYPE: Union[dict, list]

within_groups_strategy

Strategy applied to the ordering of elements inside each group. Allowed values:

  • "as_is" - leave element order as provided
  • "randomize" - shuffle elements inside each group (non-deterministic)
  • "latin_square" - apply a balanced Latin square to permute positions across participants

If None, defaults to "as_is". If order was provided as a list-of-lists and groups_strategy was set to "randomize" or "latin_square" and not explicitly set for within_groups_strategy, the groups_strategy may be reused for within-groups behavior in some calling patterns.

TYPE: Union[str, None] DEFAULT: None

groups_strategy

Strategy applied to the sequence of groups. Allowed values:

  • "as_is"
  • "randomize"
  • "latin_square"

If None, defaults to "as_is". When order is a dict keyed by participant index, groups_strategy must be "as_is" because the dict already selects per-participant grouping.

TYPE: Union[str, None] DEFAULT: None

init_blocks_strategy

Strategy for ordering initial blocks. Allowed values:

  • "as_is"
  • "randomize"

If None, defaults to "as_is". When init_block_names is a dict keyed by participant index, the strategy must be "as_is" (per-participant selection).

TYPE: Union[str, None] DEFAULT: None

final_blocks_strategy

Strategy for ordering final blocks. Same semantics and allowed values as init_blocks_strategy.

TYPE: Union[str, None] DEFAULT: None

RETURNS DESCRIPTION
List

A list of block configuration dictionaries (the original dicts from config) in the final order constructed for the participant: [init_blocks..., main_blocks..., final_blocks...]. The main_blocks portion is produced by flattening the possibly nested group structure after applying the requested group- and within-group strategies.

TYPE: List

RAISES DESCRIPTION
ExperimentServerConfigurationExcetion

on invalid configuration, such as:

  • duplicate block names in config
  • non-string block names in orders
  • unsupported strategy names
  • inconsistent group sizes for latin-square within-group ordering
  • improper dict key sets or types when dict-based per-participant selection is used

experiment_server.utils

FileModifiedWatcher

FileModifiedWatcher(config_file: Union[Path, str], callback: Callable)

Bases: PatternMatchingEventHandler

on_modified
on_modified(event)
end_watch
end_watch()

balanced_latin_square

balanced_latin_square(number_of_conditions)

merge_dicts

merge_dicts(dict_a, dict_b)

Merge the values from dict_a and dict_b recursively. If there is a conflict, the value in dict_a will be selected.