diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index 52011d3..abe7dfa 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -21,7 +21,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', '3.12'] + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12', '3.13', '3.14'] name-prefix: ['Linux (Python '] env: @@ -59,7 +59,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ['3.10'] + python-version: ['3.11'] name-prefix: ['macOS (Python '] env: @@ -95,13 +95,9 @@ jobs: strategy: fail-fast: false matrix: - python-version: ['3.10'] + python-version: ['3.11'] architecture: ['x64'] name-prefix: ['Windows (Python '] - include: - - python-version: '3.10' - architecture: 'x86' - name-prefix: 'Windows 32-bit (Python ' env: SDL_VIDEODRIVER: dummy diff --git a/docs/CHANGELOG.rst b/docs/CHANGELOG.rst index e1c66be..9bbf4ee 100644 --- a/docs/CHANGELOG.rst +++ b/docs/CHANGELOG.rst @@ -11,10 +11,36 @@ This is a log of the latest changes and improvements to KLibs. Runtime Changes: * KLibs now requires Python 3.7 or newer to run, dropping support for 2.7. +* Trial recycling behaviour has been changed, such that recycled trials are + now re-inserted at a random position in the list of remaining trials, avoiding + re-insertion as the next trial (to avoid an immediate repeat) unless it is the + only trial remaining. Previously recycling a trial would shuffle the order of + *all* remaining trials, which could unexpectedly affect the even distribution + of trial factors within blocks. +* Practice blocks are no longer added when `P.run_practice_blocks` is False. + +API Changes: + +* The :class:`~klibs.KLTrialFactory.TrialFactory` class has been removed from + the public API and should no longer be used directly by new projects. +* Added a new class :class:`~klibs.KLTrialFactory.TrialSet` to allow for + defining custom block types and block structures as well as setting labels + for blocks (accessible during blocks through the `self.block_label` attribute + in the experiment runtime). +* Removed the :class:`~klibs.KLTrialFactory.BlockIterator` class. + The `self.blocks` attribute of the Experiment class is now a list of + :class:`~klibs.KLTrialFactory.TrialSet` objects. +* Experiment factor names and levels are now accessible directly through the + :attr:`~klibs.KLExperiment.exp_factors` attribute during the Experiment + runtime (e.g. `self.exp_factors`). +* Added a new method :method:`~klibs.KLExperiment.write_trials_txt` for + exporting the full sequence of generated trials and blocks to a human + readable text file. Fixed Bugs: * KLibs no longer crashes on launch with Python 3.12. +* KLibs no longer briefly shows a blank screen when a trial is recycled. 0.7.7b1 diff --git a/docs/source/api/trial_factory.rst b/docs/source/api/trial_factory.rst index 36a3a80..fe4a0a1 100755 --- a/docs/source/api/trial_factory.rst +++ b/docs/source/api/trial_factory.rst @@ -1,6 +1,6 @@ KLTrialFactory ============== -.. automodule:: klibs.KLTrialFactory +.. automodule:: klibs.KLTrialSet :undoc-members: :members: diff --git a/klibs/KLExperiment.py b/klibs/KLExperiment.py index a14d0af..6b95039 100755 --- a/klibs/KLExperiment.py +++ b/klibs/KLExperiment.py @@ -2,13 +2,14 @@ __author__ = 'Jonathan Mulle & Austin Hurst' import os +import random from abc import abstractmethod from traceback import print_tb, print_stack from klibs import P from klibs.KLEnvironment import EnvAgent from klibs.KLExceptions import TrialException -from klibs.KLInternal import full_trace +from klibs.KLInternal import full_trace, iterable from klibs.KLInternal import colored_stdout as cso @@ -28,44 +29,60 @@ def __init__(self): self.incomplete = True # flag for keeping track of session completeness self.blocks = None # blocks of trials for the experiment self.tracker_dot = None # overlay of eye tracker gaze location in devmode + self.block_label = None # runtime attribute containing label of current block self.audio = AudioManager() # initialize audio management for the experiment self.rc = ResponseCollector() # add default response collector self.database = self.db # use database from env self._evm = EventManager() - self.trial_factory = TrialFactory() + self._exp_factors = self._get_exp_factors() + self.trial_factory = TrialFactory(self._exp_factors) if P.manual_trial_generation is False: self.trial_factory.generate() - self.event_code_generator = None + + + def _get_exp_factors(self): + # Reads in the trial factors for the study, including any local overrides + from klibs.KLTrialFactory import _load_factors + + # Load experiment factors from the project's _independent_variables.py file(s) + factors = _load_factors(P.ind_vars_file_path) + if os.path.exists(P.ind_vars_file_local_path): + if not P.dm_ignore_local_overrides: + local_factors = _load_factors(P.ind_vars_file_local_path) + factors.update(local_factors) + + return factors def __execute_experiment__(self, *args, **kwargs): """For internal use, actually runs the blocks/trials of the experiment in sequence. """ - from klibs.KLGraphics import clear - if self.blocks == None: self.blocks = self.trial_factory.export_trials() + P.blocks_per_experiment = len(self.blocks) P.block_number = 0 P.trial_id = 0 for block in self.blocks: P.recycle_count = 0 P.block_number += 1 P.practicing = block.practice + self.block_label = block.label self.block() P.trial_number = 1 - for trial in block: # ie. list of trials + remaining = list(block.trials) + while len(remaining): + trial = remaining.pop(0) try: P.trial_id += 1 # Increments regardless of recycling - self.__trial__(trial, block.practice) + self.__trial__(trial) P.trial_number += 1 except TrialException: - block.recycle() + remaining = self._recycle_trial(remaining, trial) P.recycle_count += 1 - clear() # NOTE: is this actually wanted? self.rc.reset() self.clean_up() @@ -75,7 +92,7 @@ def __execute_experiment__(self, *args, **kwargs): self.database.update('session_info', {'complete': True}, where) - def __trial__(self, trial, practice): + def __trial__(self, trial): """ Private method; manages a trial. """ @@ -132,6 +149,34 @@ def __log_trial__(self, trial_data): return self.database.insert(trial_template) + def _recycle_trial(self, remaining, trial): + """Internal method for recycling a trial within the current block. + + This method re-inserts a trial into the set of remaining trials at a + random position, avoiding an immediate repeat of the trial unless it is + the only trial remaining in the block. + + Recycling behaviour can be customized by overriding this method. + + Args: + remaining (list): The remaining trials for the current block. + trial (dict): The trial factors to recycle into the block. + + Returns: + list: The new set of remaining trials. + + """ + # NOTE: Should this be part of public API or stay unofficial/internal? + if len(remaining): + # Re-insert the trial in a random position after the first element + tmp = remaining.copy() + new_idx = random.randrange(1, len(tmp)) if len(tmp) > 1 else 1 + tmp.insert(new_idx, trial) + return tmp + else: + return [trial] + + ## Define abstract methods to be overridden in experiment.py ## @abstractmethod @@ -197,52 +242,94 @@ def clean_up(self): def insert_practice_block(self, block_nums, trial_counts=None, factor_mask=None): - """ - Adds one or more practice blocks to the experiment. This function must be called during setup(), - otherwise the trials will have already been exported and this function will no longer have - any effect. If you want to add a block to the experiment after setup() for whatever reason, - you can manually generate one using trial_factory.generate() and then insert it using - self.blocks.insert(). - - If multiple block indexes are given but only a single integer is given for trial counts, - then all practice blocks inserted will be trial_counts trials long. If not trial_counts - value is provided, the number of trials per practice block defaults to the global - experiment trials_per_block parameter. + """Adds a practice block to the experiment. - If multiple block indexes are given but only a single factor mask is provided, the same - factor mask will be applied to all appended practice blocks. If no factor mask is provided, - the function will generate a full set of trials based on all possible combination of factors, - and will randomly select trial_counts trials from it for each practice block. + This method adds an extra block of trials at a given position in the block + sequence, optionally with a different trial count and/or different factor + levels than the rest of the task. For example, to add a practice block with + 20 trials at the start of the task, you would add the following somewhere in + the `setup()` block of your `experiment.py` file:: - Args: - block_nums (:obj:`list` of int): Index numbers at which to insert the blocks. - trial_counts (:obj:`list` of int, optional): The numbers of trials to insert for each - of the inserted blocks. - factor_mask (:obj:`dict` of :obj:`list`, optional): Override values for the variables - specified in independent_variables.py. + self.insert_practice_block(1, 20) - Raises: - TrialException: If called after the experiment's :meth:`setup` method has run. + During practice blocks the klibs parameter `P.practicing` will be set to True, + allowing easy conditional changes during practice blocks (e.g. showing + additional feedback if practicing). + + You can also provide a factor mask to override one or more factor levels for + the practice block. For example, if the task has a factor 'difficulty' with + the levels 'easy' and 'hard' and you want to add separate practice blocks for + each trial type, you can specify overrides for the factor levels like so:: + + self.insert_practice_block(1, 32, factor_mask={'difficulty': ['easy']}) + self.insert_practice_block(2, 32, factor_mask={'difficulty': ['hard']}) + + This function must be called during setup(), otherwise the block structure of + the study will already be set and can no longer be changed. + + Args: + block_nums (int): Position at which to insert the block. + trial_counts (int, optional): The trial count for the practice block. + Defaults to `P.trials_per_block`. + factor_mask (:obj:`dict` of :obj:`list`, optional): Overrides for one or + more factors in the task's `independent_variables.py` file. """ + # [Compat]: Messy API to allow multiple insertions at once, fix when possible. + # Only TOJ_Motion uses multiple insertions. Multiple projects use 'trial_counts' + # keyword, however. + if self.blocks: # If setup has passed and trial execution has started, blocks have already been exported # from trial_factory so this function will no longer work. If it is called after it is no # longer useful, we throw a TrialException - raise TrialException("Practice blocks cannot be inserted after setup() is complete.") - try: - iter(block_nums) - except TypeError: - block_nums = [block_nums] - try: - iter(trial_counts) - except TypeError: - trial_counts = ([P.trials_per_block] if trial_counts is None else [trial_counts]) * len(block_nums) - while len(trial_counts) < len(block_nums): - trial_counts.append(P.trials_per_block) - for i in range(0, len(block_nums)): - self.trial_factory.insert_block(block_nums[i], True, trial_counts[i], factor_mask) - P.blocks_per_experiment += 1 + raise RuntimeError("Cannot insert practice blocks after setup() is complete.") + + if not trial_counts: + trial_counts = P.trials_per_block + + if iterable(block_nums): + # [Compat]: Only TOJ_Motion uses this and it's a bad idea, remove when fixed. + for b in block_nums: + self.insert_practice_block(b, trial_counts, factor_mask) + else: + self.trial_factory.insert_block(block_nums, trial_counts, True, factor_mask) + + + def write_trials_txt(self, outpath=None): + """Writes the current block/trial structure to a text file. + + This method is intended for verifying your blocks and trials are being + generated and sequenced as expected during development. The factors for + each trial within each block are written in a human-readable format:: + + ========================= + == Block 1 (3 trials) == + ========================= + + trial cue_validity soa target_loc + ----- ------------ --- ---------- + 1 valid 200 left + 2 invalid 800 right + 3 neutral 800 left + + A summary of the block structure and a list of the experiment factors + and their base levels is also included at the top of the file. + + Args: + outpath (str, optional): The path at which to save the text file. + Defaults to `ExpAssets/Local/[project_name]_trials.txt`. + + """ + from klibs.KLTrialFactory import _structure_to_str + + if not outpath: + fname = "{0}_trials.txt".format(P.project_name) + outpath = os.path.join(P.local_dir, fname) + + with open(outpath, "w") as out: + blocks = self.blocks if self.blocks else self.trial_factory.blocks + out.write(_structure_to_str(blocks, self.exp_factors)) def before_flip(self): @@ -340,6 +427,16 @@ def show_logo(self): any_key() + @property + def exp_factors(self): + """dict: The names and levels of all categorical factors in the study. + + This attribute is read-only, meaning that any changes to this attribute's + keys or values will have no effect on the experiment runtime. + + """ + return self._exp_factors.copy() + @property def evm(self): """:obj:`~klibs.KLEventInterface.EventManager`: The trial event sequencer for diff --git a/klibs/KLParams.py b/klibs/KLParams.py index a2cf23d..4eb05eb 100755 --- a/klibs/KLParams.py +++ b/klibs/KLParams.py @@ -54,7 +54,7 @@ conditions = [] default_condition = None table_defaults = {} # default column values for db tables when using EntryTemplate -run_practice_blocks = True # (not implemented in klibs itself) +run_practice_blocks = True color_output = False # whether cso() outputs colorized text or not # Eye Tracking Settings diff --git a/klibs/KLTrialFactory.py b/klibs/KLTrialFactory.py index 96bc372..e035a1f 100755 --- a/klibs/KLTrialFactory.py +++ b/klibs/KLTrialFactory.py @@ -60,145 +60,195 @@ def _generate_blocks(factors, block_count, trial_count): return blocks -class BlockIterator(object): +def _block_to_str(block, num): + # Generates a string describing the structure and factor levels for each + # trial in a given block + + # Generate a block header + info = "{0} trials".format(len(block)) + info += ", '{0}'".format(block.label) if block.label else "" + info += ", Practice" if block.practice else "" + block_info = "== Block {0} ({1}) ==".format(num, info) + out = ["=" * len(block_info), block_info, "=" * len(block_info)] + + # Get max character length for each factor level for sake of alignment + col_pad = {'trial': max(len(str(len(block))), len('trial'))} + factors = list(block.trials[0].keys()) + for f in factors: + if not f in col_pad.keys(): + col_pad[f] = len(f) + for row in block.trials: + if len(str(row[f])) > col_pad[f]: + col_pad[f] = len(str(row[f])) + + if len(factors): + cols = ['trial'] + factors + + # Generate a header for the different factors + out.append("") + out.append(" ".join([col.ljust(col_pad[col]) for col in cols])) + out.append(" ".join(["-" * col_pad[col] for col in cols])) + + # Write the factor levels for each trial in the block + t = 1 + for trial in block.trials: + row = str(t).ljust(col_pad['trial']) + " " + row += " ".join([str(trial[f]).ljust(col_pad[f]) for f in factors]) + out.append(row) + t += 1 + + out.append("") + return "\n".join(out) + + +def _structure_to_str(blocks, factors): + # Converts the block structure, experiment factors, and trial sequence for + # each block into a human-readable string + + # Write out the block structure + out = [] + out.append("") + out.append("Blocks:") + for i in range(len(blocks)): + b = str(blocks[i]).replace("TrialSet", "") + out.append(" - Block {0}: ".format(i+1) + b) + out.append("") + + # Write out the factor list + if len(factors.items()): + out.append("Factors:") + for name, values in factors.items(): + out.append(" - {0}: {1}".format(name, values)) + else: + out.append("Factors: None") - def __init__(self, blocks): - self.blocks = blocks - self.practice_blocks = [] - self.length = len(blocks) - self.i = 0 + # Write out the trials (and factors) for each block + out.append("\n") + block_num = 1 + for b in blocks: + out.append("") + out.append(_block_to_str(b, block_num)) + block_num += 1 - def __iter__(self): - return self + return "\n".join(out) - def __len__(self): - return self.length - def __getitem__(self, i): - return self.blocks[i] - def __setitem__(self, i, x): - self.blocks[i] = x - - def insert(self, index, block, practice): - if self.i <= index: - if practice: - self.practice_blocks.append(index) - self.blocks.insert(index, block) - self.length = len(self.blocks) - else: - insert_err = "Can't insert block at index {0}; it has already passed." - raise ValueError(insert_err.format(index)) +class TrialSet(object): + """Class for representing blocks of trials. - def next(self): # alias for python2 - return self.__next__() + TrialSet objects are how klibs represents blocks of trials internally, and + can be used to manually generate custom sequences of blocks/trials during + the `self.setup()` phase of the Experiment runtime. - def __next__(self): - if self.i >= self.length: - self.i = 0 # reset index so we can iterate over it again - raise StopIteration - else: - self.i += 1 - trials = TrialIterator(self.blocks[self.i - 1]) - trials.practice = self.i - 1 in self.practice_blocks - return trials + The full set of blocks of trials for an experiment is stored as a list of + TrialSets in the experiment attribute `self.blocks`. By default these blocks + and trials are generated for you using the defined factors and specified + block/trial counts in the project's configuration files, but you can use + your own set of custom blocks by replacing the block list with your own:: + # Define a sequence of 3 trials + trials = [{'image': 'a'}, {'image': 'b'}, {'image': 'c'}] -class TrialIterator(BlockIterator): + # Set block sequence for task as 2 identical blocks of trials + self.blocks = [ + TrialSet(trials, practice=True), # Flag block 1 as practice + TrialSet(trials), + ] - def __init__(self, block_of_trials): - self.trials = block_of_trials - self.length = len(block_of_trials) - self.i = 0 - self.__practice = False + If a block is provided with a label, the value of the label can be accessed + during the block through the Experiment attribute `self.block_label`. This + can be used to change things like stimuli or instructions conditionally + in your code based on the block label (e.g. different cues for 'endo' and + 'exo' blocks). - def __next__(self): - if self.i >= self.length: - self.i = 0 - raise StopIteration - else: - self.i += 1 - return self.trials[self.i - 1] + Note that custom block sequences can only be set during the `self.setup()` + phase of the task. + + Args: + trials (List): A list of dicts containing trial factors, with each dict + representing a trial in the block. + practice (bool, optional): Whether the block is a practice block. + Defaults to False. + label (str, optional): A label optionally specifying the block type for + experiments with multiple types of block. Defaults to None. - def recycle(self): - self.trials.append(self.trials[self.i - 1]) - temp = self.trials[self.i:] - random.shuffle(temp) - self.trials[self.i:] = temp - self.length += 1 + """ + def __init__(self, trials, practice=False, label=None): + self._trials = trials.copy() + self.practice = practice + self.label = label + + def __len__(self): + return len(self._trials) + + def __str__(self): + # Custom print method for better readability + s = "{0} trials".format(len(self._trials)) + s += ", '{0}'".format(self.label) if self.label else "" + s += ", Practice" if self.practice else "" + return "TrialSet(" + s + ")" @property - def practice(self): - return self.__practice + def trials(self): + """List: The list of trials contained within the block.""" + return self._trials.copy() - @practice.setter - def practice(self, practicing): - self.__practice = practicing == True +# Alias for backwards compatibility +TrialIterator = TrialSet class TrialFactory(object): + """Generates blocks of trials using a given set of categorical factors. - def __init__(self): + For internal use. - self.blocks = None - self.trial_generator = self.__generate_trials - - # Load experiment factors from the project's _independent_variables.py file(s) - factors = _load_factors(P.ind_vars_file_path) - if os.path.exists(P.ind_vars_file_local_path): - if not P.dm_ignore_local_overrides: - local_factors = _load_factors(P.ind_vars_file_local_path) - factors.update(local_factors) - + Args: + factors (dict): A dict containing the factor names and factor levels + to use for generating trials. + + """ + def __init__(self, factors): # Create alphabetically-sorted ordered dict from factors self.exp_factors = OrderedDict(sorted(factors.items(), key=lambda t: t[0])) + self.blocks = None - def __load_ind_vars(self, path): - - set_name = "{0}_ind_vars".format(P.project_name) - try: - ind_vars = load_source(path) - factors = ind_vars[set_name].to_dict() - except KeyError: - err = 'Unable to find IndependentVariableSet in independent_vars.py.' - raise RuntimeError(err) - - return factors - + def trial_generator(self, factors, block_count, trial_count): + """Method that actually generates blocks of trials. - def __generate_trials(self, factors, block_count, trial_count): + """ # NOTE: Factored into a separate function for easier unit testing return _generate_blocks(factors, block_count, trial_count) - def generate(self, exp_factors=None, block_count=None, trial_count=None): + def generate(self, num_blocks=None, trials_per_block=None): + """Generates an initial set of blocks. + """ # If block/trials-per-block counts aren't specified, use values from params.py - if block_count is None: - block_count = 1 if not P.blocks_per_experiment > 0 else P.blocks_per_experiment - if trial_count is None: - trial_count = P.trials_per_block + if num_blocks is None: + num_blocks = 1 if P.blocks_per_experiment <= 0 else P.blocks_per_experiment + if trials_per_block is None: + trials_per_block = P.trials_per_block - exp_factors = self.exp_factors if exp_factors == None else exp_factors - blocks = self.trial_generator(exp_factors, block_count, trial_count) - self.blocks = BlockIterator(blocks) + blocks = self.trial_generator(self.exp_factors, num_blocks, trials_per_block) + self.blocks = [TrialSet(b) for b in blocks] - def export_trials(self): - if not self.blocks: - raise RuntimeError("Trials must be generated before they can be exported.") - return self.blocks + def insert_block(self, block_num, trials=0, practice=False, factor_mask=None): + """Inserts a new block of trials into the block sequence. + Args: + block_num (int): The block number for the inserted block. + trials (int, optional): The trial count for the block. If not specified, a + block containing a full set of factor combinations will be inserted. + practice (bool, optional): Whether to flag the block as a practice block. + Defaults to False. + factor_mask (dict, optional): A dict containing overrides for the levels of + one or more of the factors. - def insert_block(self, block_num, practice=False, trial_count=0, factor_mask=None): - """ - - :param block_num: - :param practice: - :param trial_count: - :param factor_mask: """ if factor_mask: if not isinstance(factor_mask, dict): @@ -212,51 +262,33 @@ def insert_block(self, block_num, practice=False, trial_count=0, factor_mask=Non new_values = [new_values] # if not iterable, put in list factors[name] = new_values else: - e = "'{0}' is not the name of an active independent variable".format(name) + e = "'{0}' is not the name of an active factor".format(name) raise ValueError(e) else: # If no factor mask, generate trials randomly based on self.exp_factors factors = self.exp_factors - block = self.trial_generator(factors, 1, trial_count) + # Don't insert practice blocks if practice blocks disabled + if P.run_practice_blocks == False: + return + + block = self.trial_generator(factors, 1, trials)[0] # there is no "zero" block from the UI/UX perspective, so adjust insertion accordingly - self.blocks.insert(block_num - 1, block[0], practice) + self.blocks.insert(block_num - 1, TrialSet(block, practice=practice)) - def num_values(self, factor): - """ + def export_trials(self): + """Exports the current block sequence. - :param factor: - :return: :raise ValueError: """ - try: - n = len(self.exp_factors[factor]) - return n - except KeyError: - e_msg = "Factor '{0}' not found.".format(factor) - raise ValueError(e_msg) + if not self.blocks: + self.generate() + + return self.blocks def dump(self): - # TODO: Needs a rewrite - with open(os.path.join(P.local_dir, "TrialFactory_dump.txt"), "w") as log_f: - log_f.write("Blocks: {0}, ".format(P.blocks_per_experiment)) - log_f.write("Trials: {0}\n\n".format(P.trials_per_block)) - log_f.write("*****************************************\n") - log_f.write("* Factors *\n") - log_f.write("*****************************************\n\n") - for name, values in self.exp_factors.items(): - log_f.write("{0}: {1}\n".format(name, values)) - log_f.write("\n\n\n") - log_f.write("*****************************************\n") - log_f.write("* Trials *\n") - log_f.write("*****************************************\n\n") - block_num = 1 - for b in self.blocks: - log_f.write("Block {0}\n".format(block_num)) - trial_num = 1 - for t in b: - log_f.write("\tTrial {0}: {1} \n".format(trial_num, t)) - trial_num += 1 - block_num += 1 - log_f.write("\n") + # Compat: Can remove once taken out of TraceLab + outpath = os.path.join(P.local_dir, "TrialFactory_dump.txt") + with open(outpath, "w") as log_f: + log_f.write(_structure_to_str(self.blocks, self.exp_factors)) diff --git a/klibs/tests/conftest.py b/klibs/tests/conftest.py index 0900b79..715f0b9 100644 --- a/klibs/tests/conftest.py +++ b/klibs/tests/conftest.py @@ -31,8 +31,7 @@ def get_resource_path(resource): def with_sdl(): sdl2.SDL_ClearError() ret = sdl2.SDL_Init(sdl2.SDL_INIT_VIDEO | sdl2.SDL_INIT_TIMER) - assert sdl2.SDL_GetError() == b"" - assert ret == 0 + assert ret == 0, sdl2.SDL_GetError().decode('utf-8', 'replace') yield sdl2.SDL_Quit() diff --git a/klibs/tests/test_KLExperiment.py b/klibs/tests/test_KLExperiment.py index f493de5..af884a1 100755 --- a/klibs/tests/test_KLExperiment.py +++ b/klibs/tests/test_KLExperiment.py @@ -1,27 +1,182 @@ import os import mock import pytest +import tempfile +from collections import OrderedDict import klibs from klibs.KLJSON_Object import AttributeDict +from klibs.KLTrialFactory import TrialIterator, TrialSet +from klibs.KLExceptions import TrialException +from klibs.KLExperiment import Experiment from conftest import get_resource_path +class MockExperiment(Experiment): + def __trial__(self, trial): + pass + @pytest.fixture -def experiment(): - from klibs.KLExperiment import Experiment +def run_environment(): from klibs import P template_path = get_resource_path('template') P.ind_vars_file_path = os.path.join(template_path, "independent_variables.py") P.ind_vars_file_local_path = os.path.join(template_path, "doesnt_exist.py") P.manual_trial_generation = True P.project_name = "PROJECT_NAME" - return Experiment() + +@pytest.fixture +def experiment(run_environment): + exp = MockExperiment() + exp.database = AttributeDict({'tables': []}) + return exp def test_Experiment(experiment): with mock.patch.object(experiment, 'quit', return_value=None): experiment.blocks = [] - experiment.database = AttributeDict({'tables': []}) experiment.run() + + +def test_execute(run_environment): + from klibs import P + + class TestExperiment(Experiment): + + def setup(self): + self.last_block = 0 + self.last_trial = 0 + self.total_trials = 0 + self.was_recycled = False + self.database = AttributeDict({'tables': []}) + self.blocks = [] + + def block(self): + # Check block number incrementing as expected + assert P.block_number == (self.last_block + 1) + self.last_block = P.block_number + self.last_trial = 0 + # Check block label getting set as expected + expected = 'test' if P.block_number == 1 else None + assert self.block_label == expected + + def __trial__(self, trial): + # Check trial id increments correctly + self.total_trials += 1 + assert P.trial_id == self.total_trials + # Check trial numbers increment correctly + trial_num = self.last_trial + 1 + if P.recycle_count == 0: + assert P.trial_number == trial_num + else: + assert trial_num == (P.trial_number + P.recycle_count) + self.last_trial += 1 + # Test that trial factors getting passed properly + assert isinstance(trial, dict) + assert 'fac1' in list(trial.keys()) + assert 'fac2' in list(trial.keys()) + # Test that practice getting set correctly + assert P.practicing == (P.block_number == 1) + # Test trial recycling + if self.was_recycled: + assert P.recycle_count == 1 + self.was_recycled = False + if trial_num == 3: + self.was_recycled = True + raise TrialException("recycling") + + # Initialize test blocks/trials + trials = [ + {'fac1': True, 'fac2': 200}, + {'fac1': False, 'fac2': 200}, + {'fac1': True, 'fac2': 400}, + {'fac1': False, 'fac2': 400}, + ] + + # Test with blocks as list + tst = TestExperiment() + tst.setup() + tst.blocks = [ + TrialSet(trials, label='test', practice=True), + TrialIterator(trials), # alias for backwards compat + ] + tst.__execute_experiment__() + assert tst.last_block == 2 + assert tst.last_trial == 5 # 4 + 1 recycled + assert tst.total_trials == 10 + + +def test_insert_practice_block(experiment): + from klibs import P + + # Set dummy trial factors and generate trials + P.trials_per_block = 30 + P.blocks_per_experiment = 1 + experiment.trial_factory.exp_factors = OrderedDict([ + ('fac1', [True, False]), + ('fac2', [200, 400, 800]), + ]) + experiment.trial_factory.generate() + blocks_init = experiment.trial_factory.export_trials() + assert len(blocks_init) == 1 + assert len(blocks_init[0]) == 30 + + # Try adding a single practice block + experiment.insert_practice_block(1, 12) + blocks_a = experiment.trial_factory.export_trials() + assert len(blocks_a) == 2 + assert len(blocks_a[0]) == 12 + assert len(blocks_a[1]) == 30 + + # Try adding two more practice blocks with a factor mask + mask = {'fac2': [800]} + experiment.insert_practice_block([1, 3], 6, factor_mask=mask) + blocks_b = experiment.trial_factory.export_trials() + assert len(blocks_b) == 4 + assert len(blocks_b[0]) == 6 + assert len(blocks_b[1]) == 12 + assert len(blocks_b[2]) == 6 + for trial in blocks_b[0].trials: + assert trial['fac2'] == 800 + + # Test to make sure method does nothing if practice blocks disabled + P.run_practice_blocks = False + experiment.insert_practice_block(1) + blocks_c = experiment.trial_factory.export_trials() + assert len(blocks_c) == 4 + + # Test that block count parameter updated when experiment run + experiment.database = AttributeDict({'tables': []}) + experiment.__execute_experiment__() + assert P.blocks_per_experiment == 4 + + +def test_trials_txt(experiment): + from klibs import P + + # Set dummy trial factors and generate trials + P.trials_per_block = 12 + P.blocks_per_experiment = 2 + P.run_practice_blocks = True + experiment.trial_factory.exp_factors = OrderedDict([ + ('fac1', [True, False]), + ('fac2', [200, 400, 800]), + ]) + experiment.trial_factory.generate() + experiment.insert_practice_block(1, 6) + + # Try exporting to a temporary file + tmpdir = tempfile.mkdtemp() + tmpfile = os.path.join(tmpdir, "trials.txt") + assert not os.path.exists(tmpfile) + experiment.write_trials_txt(tmpfile) + assert os.path.exists(tmpfile) + + # Check file to make sure it's working + with open(tmpfile, "r") as f: + output = f.read() + assert "Blocks:" in output + assert "Factors:" in output + assert "Block 1 (6 trials" in output + assert "Block 2 (12 trials" in output