TaskClient

Emit experiment structure and stimulus events from your task code so they land in the recording, time-aligned with the neural data.

Setup

pip install kernel_sdk

The defaults match where kortex listens for task events, so on a normal setup there is nothing to pass:

from kernel.sdk.task import TaskClient

tc = TaskClient()

Events are sent over TCP. Point ip at another machine if your task runs somewhere other than the acquisition box.

How events work

Everything the client emits is a name/value pair with a timestamp and a monotonically increasing id. The structural methods below are conveniences that emit matched start_* and end_* pairs around a block of your code.

Counters never reset. The third trial of the second block is trial 3 overall, not trial 1 of block 2 — use the counter accessors if you need per-grouping numbers.

Experiment, task, block, trial

Each behaves two ways depending on num:

• num == 1 (default) — a context manager. Emits start_* on entry and end_* on exit.
• num > 1 — an iterator for for ... in .... Emits a start_*/end_* pair per iteration and tracks the counter.

Nesting is enforced: experiment() must be called first and at the top level; a task must be inside an experiment; a block inside a task or experiment; a trial inside a block, task, or experiment.

import random

with tc.experiment() as e:                       # start_experiment: 1
    with tc.block(type="practice") as b:         # start_block: 1
                                                 # block_type: "practice"
        for t in tc.trial(
            num=5,
            type=lambda n: random.choice(["left", "right"])
        ):                                       # start_trial: 1 .. 5
                                                 # trial_type: "left" | "right"
            print(e.current_count())             # 1
            print(b.current_count())             # 1
            print(t.current_count())             # 1 .. 5
            print(b.current_type())              # "practice"
            print(t.current_type())              # "left" | "right"
                                                 # end_trial: 1 .. 5
                                                 # end_block: 1

    with tc.block(type="actual") as b:           # start_block: 2
                                                 # block_type: "actual"
        with tc.trial() as t:                    # start_trial: 6
            print(t.current_count())             # 6
                                                 # end_trial: 6

        trial_list = ["type1", "type2", "type3"]
        for t in tc.trial(num=6, type=trial_list):
            print(t.current_count(), t.current_type())
            # 8 type1 / 9 type2 / 10 type3 / 11 type1 / 12 type2 / 13 type3
                                                 # end_trial: 8 .. 13
                                                 # end_block: 2
                                                 # end_experiment: 1

Without a context manager

The returned object also supports explicit start() and stop():

e = tc.experiment().start()      # start_experiment: 2
print(e.current_count())         # 2
e.stop()                         # end_experiment: 2

Counters & types

The type argument accepts three forms, and emits a *_type event each time it changes:

Task registry

Passing a registered type to experiment() also emits an experiment_meta event carrying the canonical task URN.

from kernel.sdk.task_registry import TaskRegistry

with tc.experiment(type=TaskRegistry.NBACK02.name) as e:
    ...

An unrecognised type logs a warning and continues, unless the client was constructed with enforce_registry_entry=True, in which case it raises InvalidTaskType.

iti

Emits start_inter_trial_interval with an incrementing counter, then inter_trial_interval with the duration, then sleeps, then emits end_inter_trial_interval with the matching counter.

with tc.experiment() as e:
    for t in tc.trial(num=10):
        ...                  # trial code
        tc.iti(5.3)          # start_inter_trial_interval: 1 .. 10
                             # inter_trial_interval: 5.3
                             # sleeps 5.3 seconds
                             # end_inter_trial_interval: 1 .. 10

The sleep argument is called with time_in_seconds, so you can substitute a higher-precision sleep — for example one that spins on your display's vsync.

rest

Same shape as iti, emitting start_rest, rest_interval, and end_rest.

with tc.experiment() as e:
    for t in tc.trial(num=10):
        ...
        tc.rest(5.3)         # start_rest: 1 .. 10
                             # rest_interval: 5.3
                             # end_rest: 1 .. 10

cue

Same shape again, emitting start_cue, cue_duration, and end_cue. Use it for a fixation cross or instruction cue preceding a stimulus.

with tc.trial() as t:
    tc.cue(1.0)              # start_cue: 1 / cue_duration: 1.0 / end_cue: 1
    tc.stim("target.png")

stim

Emits a stimulus event. Every keyword argument becomes its own event, which is how you attach stimulus properties.

with tc.experiment() as e:
    for t in tc.trial(num=9):
        key_to_press = get_random_key()
        tc.stim(key_to_press)               # stimulus: "k"

with tc.experiment() as e:
    for t in tc.trial(num=10):
        photo_img, photo_gender, photo_age = random_photo()
        tc.stim(
            photo_img,
            photo_gender=photo_gender,
            photo_age=photo_age)            # stimulus: "img_2345.png"
                                            # photo_gender: "male"
                                            # photo_age: 42
        ...                                 # show photo_img

rt

Called on the object returned by stim(). Emits a reaction_time event measured from that stimulus and returns the interval in seconds. Because each stimulus carries its own start time, you can measure against several stimuli independently.

with tc.experiment() as e:
    for t in tc.trial(num=9):
        key_to_press = get_random_key()
        s1 = tc.stim(key_to_press)          # stimulus: "k"
        ...                                 # show key to participant
        s2 = tc.stim("photo323.png")        # stimulus: "photo323.png"
        ...                                 # show distractor, wait for keypress
        rt1 = s1.rt()                       # reaction_time: 0.5
        print(rt1)                          # 0.5 — measured from s1, not s2
        tc.iti(5.0)

sync

A context manager emitting start_sync and end_sync with an incrementing counter. Wrap whatever produces your photodiode pulse so the optical sync can be aligned with the data stream.

with tc.experiment() as e:
    for t in tc.trial(num=10):
        with tc.sync():      # start_sync: 1 .. 10
            ...              # show white square, sleep, hide it
                             # end_sync: 1 .. 10

event

A context manager emitting start_{name} and end_{name} with an incrementing counter. Use it for any interval the built-in groupings do not cover.

with tc.event("instructions"):   # start_instructions: 1
    ...                          # show instructions, wait for key
                                 # end_instructions: 1

with tc.event("instructions"):   # start_instructions: 2
    ...
                                 # end_instructions: 2

send_event

The primitive underneath everything else: emits one raw event per keyword argument.

tc.send_event(pressed="k")                  # pressed: "k"
tc.send_event(pressed="k", correct=True)    # pressed: "k"
                                            # correct: True

A complete task: finger tapping

Everything above, assembled into a real experiment. This is the structure Kernel's finger-tapping task uses: a baseline rest, then ten left/right block pairs of thirteen trials each, cueing one finger per trial, with an inter-trial interval between trials and a jittered rest between blocks.

The cues here just print to the console so the script runs as-is; replace cue() with whatever drives your display, audio, or Unity front end.

import json
import random
import time

from kernel.sdk.task import TaskClient
from kernel.sdk.task_registry import TaskRegistry

BLOCK_PAIRS = 10          # each pair is one left block and one right block
TRIALS_PER_BLOCK = 13
TRIAL_DURATION = 0.75     # seconds the finger cue stays up
ITI_MEAN, ITI_SD = 0.5, 0.0
BASELINE_REST = 20.0
BLOCK_REST, REST_JITTER = 20.0, 5.0
FINGERS = ["index", "middle", "ring", "pinky"]

PARAMS = {
    "trials_per_block": TRIALS_PER_BLOCK,
    "blocks_per_side": BLOCK_PAIRS,
    "trial_duration": TRIAL_DURATION,
    "ITI_mean": ITI_MEAN,
    "ITI_std": ITI_SD,
    "baseline_rest": BASELINE_REST,
    "rest_dur": BLOCK_REST,
    "rest_jitter_max": REST_JITTER,
}


def cue(text):
    """Show the participant what to do. Swap this for your own display."""
    print(text, flush=True)


tc = TaskClient(experiment_name="finger_tapping")


def rest(seconds, label="REST"):
    cue(f"{label}  ({seconds:.1f}s)")
    tc.rest(seconds)


with tc.experiment(type=TaskRegistry.FT02) as experiment:
    tc.send_event(experiment_params=json.dumps(PARAMS))
    rest(BASELINE_REST)

    for _pair in range(BLOCK_PAIRS):
        sides = ["left", "right"]
        random.shuffle(sides)          # counterbalance which hand goes first

        for side in sides:
            with tc.block(type=side) as block:
                cue(f"{side.upper()} HAND")
                for i in range(TRIALS_PER_BLOCK):
                    finger = FINGERS[i % len(FINGERS)]
                    with tc.trial(type=finger) as trial:
                        cue(f"  tap {finger}  (trial {trial.current_count()})")
                        time.sleep(TRIAL_DURATION)
                    # the trial has ended; the ITI sits between trials
                    tc.iti(max(0.0, random.gauss(ITI_MEAN, ITI_SD)))

            rest(BLOCK_REST + random.uniform(-REST_JITTER, REST_JITTER))

cue("DONE")

What the operator sees

REST  (20.0s)
RIGHT HAND
  tap index  (trial 1)
  tap middle  (trial 2)
  tap ring  (trial 3)
  tap pinky  (trial 4)
  tap index  (trial 5)
  ...
REST  (17.3s)
LEFT HAND
  tap index  (trial 14)
  ...

What lands in the recording

experiment_meta             {"task_urn": "urn:kernel.com/task/flow_neuro_FT/02", ...}
start_experiment            1
experiment_type             urn:kernel.com/task/flow_neuro_FT/02
experiment_params           {"trials_per_block": 13, ...}
start_rest                  1
rest_interval               20.0
end_rest                    1
start_block                 1
block_type                  right
start_trial                 1
trial_type                  index
end_trial                   1
start_inter_trial_interval  1
inter_trial_interval        0.5
end_inter_trial_interval    1
start_trial                 2
trial_type                  middle
...

Counters run across the whole session, so trial 14 is the first trial of the second block, not trial 1 again. Use block.current_count() and trial.current_type() when you need position within a grouping.

Emitted event names

All of these arrive on the acquisition side through SdkClient.on_task().

Debugging

Construct with debug=True and read back everything the client would have sent:

tc = TaskClient(debug=True)

with tc.experiment() as e:
    tc.send_event(pressed="k")

for event in tc.debug_events:
    print(event)
    # {"id": 1, "timestamp": ..., "event": "start_experiment", "value": "1"}
    # {"id": 2, "timestamp": ..., "event": "pressed", "value": "k"}