pyoperant.utils module

class pyoperant.utils.AuditoryStimulus(*args, **kwargs)

Bases: pyoperant.utils.Stimulus

docstring for AuditoryStimulus

class pyoperant.utils.Command(command)

Bases: object

Enables to run subprocess commands in a different thread with TIMEOUT option.

via https://gist.github.com/kirpit/1306188

Based on jcollado’s solution: http://stackoverflow.com/questions/1191374/subprocess-with-timeout/4825933#4825933

command = None

error = ''

output = ''

process = None

run(timeout=None, **kwargs)

Run a command then return: (status, output, error).

status = None

class pyoperant.utils.Event(time=None, duration=None, label='', name=None, description=None, file_origin=None, *args, **kwargs)

Bases: object

docstring for Event

annotate(**kwargs)

class pyoperant.utils.NumpyAwareJSONEncoder(skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, encoding='utf-8', default=None)

Bases: json.encoder.JSONEncoder

this json encoder converts numpy arrays to lists so that json can write them.

example usage:

>>> import numpy as np
>>> dict_to_save = {'array': np.zeros((5,))}
>>> json.dumps(dict_to_save,
               cls=NumpyAwareJSONEncoder
               )
'{"array": [0.0, 0.0, 0.0, 0.0, 0.0]}'

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

class pyoperant.utils.Stimulus(*args, **kwargs)

Bases: pyoperant.utils.Event

docstring for Stimulus

class pyoperant.utils.Trial(index=None, type_='normal', class_=None, *args, **kwargs)

Bases: pyoperant.utils.Event

docstring for Trial

pyoperant.utils.auditory_stim_from_wav(wav)

pyoperant.utils.check_cmdline_params(parameters, cmd_line)

pyoperant.utils.check_time(schedule, fmt='%H:%M')

determine whether trials should be done given the current time and the light schedule

returns Boolean if current time meets schedule

schedule=’sun’ will change lights according to local sunrise and sunset

schedule=[(‘07:00’,‘17:00’)] will have lights on between 7am and 5pm schedule=[(‘06:00’,‘12:00’),(‘18:00’,‘24:00’)] will have lights on between

pyoperant.utils.concat_wav(input_file_list, output_filename='concat.wav')

concat a set of wav files into a single wav file and return the output filename

takes in a tuple list of files and duration of pause after the file

input_file_list = [

(‘a.wav’, 0.1), (‘b.wav’, 0.09), (‘c.wav’, 0.0), ]

returns a list of AuditoryStimulus objects

TODO: add checks for sampling rate, number of channels, etc.

pyoperant.utils.get_num_open_fds()

return the number of open file descriptors for current process

pyoperant.utils.is_day(latitude='32.82', longitude='-117.14')

Is it daytime?

(lat,long) – latitude and longitude of location to check (default is San Diego) Returns True if it is daytime

pyoperant.utils.parse_commandline(arg_str=['-T', '-b', 'readthedocs', '-d', '_build/doctrees-readthedocs', '-D', 'language=en', '.', '_build/html'])

parse command line arguments note: optparse is depreciated w/ v2.7 in favor of argparse

pyoperant.utils.rand_from_log_shape_dist(alpha=10)

randomly samples from a distribution between 0 and 1 with pdf shaped like the log function low probability of getting close to zero, increasing probability going towards 1 alpha determines how sharp the curve is, higher alpha, sharper curve.

pyoperant.utils.run_state_machine(start_in='pre', error_state=None, error_callback=None, **state_functions)

runs a state machine defined by the keyword arguments

>>> def run_start():
>>>    print "in 'run_start'"
>>>    return 'next'
>>> def run_next():
>>>    print "in 'run_next'"
>>>    return None
>>> run_state_machine(start_in='start',
>>>                   start=run_start,
>>>                   next=run_next)
in 'run_start'
in 'run_next'
None

pyoperant.utils.time_in_range(start, end, x)

Return true if x is in the range [start, end]

pyoperant.utils.wait(secs=1.0, final_countdown=0.0, waitfunc=None)

Smartly wait for a given time period.

secs – total time to wait in seconds final_countdown – time at end of secs to wait and constantly poll the clock waitfunc – optional function to run in a loop during hogCPUperiod

If secs=1.0 and final_countdown=0.2 then for 0.8s python’s time.sleep function will be used, which is not especially precise, but allows the cpu to perform housekeeping. In the final hogCPUsecs the more precise method of constantly polling the clock is used for greater precision.