API reference

Shared constants for the currentsensation package.

class currentsensation.constants.CaptureConfig(width: int = 640, height: int = 480, fps: int = 25, device: str = '/dev/video0')

Default camera and recording parameters.

device: str = '/dev/video0'

fps: int = 25

height: int = 480

width: int = 640

class currentsensation.constants.ExperimentTiming(shake_dur: float = 30.0, peri_stim_dur: float = 30.0, stim_dur: float = 1800.0, delay_offset: float = 5.0, reps: int = 3)

Default time intervals for an experimental session.

All durations are in seconds.

delay_offset: float = 5.0

peri_stim_dur: float = 30.0

reps: int = 3

shake_dur: float = 30.0

stim_dur: float = 1800.0

class currentsensation.constants.SessionPaths(root: Path)

Container for output paths derived from a session save directory.

classmethod from_root(root: Path | str) → SessionPaths

Build a SessionPaths from a root directory (created if missing).

root: Path

schedule_file: Path

GPIO control with pluggable backends.

The GPIOController exposes domain-level operations (set_line_high, all_off …) and delegates the actual pin toggling to a backend. Two backends ship with the package:

• RPiBackend — real hardware via RPi.GPIO (only importable on a Raspberry Pi).

• MockBackend — in-memory state for tests and dry-runs.

Picking the right backend is the responsibility of make_default_controller(), which tries the real one and falls back to the mock with a warning.

class currentsensation.hardware.GPIOBackend(*args, **kwargs)

Minimal interface a GPIO backend must implement.

cleanup() → None

output_high(pin: int) → None

output_low(pin: int) → None

setup(pins: list[int]) → None

class currentsensation.hardware.GPIOController(backend: GPIOBackend, pin_map: dict[str, int] | None = None)

Domain-level GPIO operations for the current-sensing rig.

property active_line: str

all_off() → None

Drive every configured pin low.

change_current_line(line: str) → None

Switch to line: turn off all current lines first, then energise.

cleanup() → None

Release backend resources (idempotent).

property pin_map: dict[str, int]

set_line_high(line: str) → None

Energise a named line, or turn everything off if line == 'off'.

set_line_low(line: str) → None

De-energise a named line and mark the rig as off.

class currentsensation.hardware.MockBackend(pin_state: dict[int, bool] = <factory>, transitions: list[~currentsensation.hardware.PinTransition] = <factory>)

In-memory GPIO substitute for tests and dry-runs.

cleanup() → None

output_high(pin: int) → None

output_low(pin: int) → None

pin_state: dict[int, bool]

setup(pins: list[int]) → None

transitions: list[PinTransition]

class currentsensation.hardware.PinTransition(timestamp: float, pin: int, state: bool)

A single recorded pin state change.

pin: int

state: bool

timestamp: float

class currentsensation.hardware.RPiBackend

Backend that drives real Broadcom GPIO pins via RPi.GPIO.

cleanup() → None

output_high(pin: int) → None

output_low(pin: int) → None

setup(pins: list[int]) → None

exception currentsensation.hardware.UnknownLineError

Raised when a line tag does not exist in the pin map.

currentsensation.hardware.make_default_controller(pin_map: dict[str, int] | None = None, *, force_mock: bool = False) → GPIOController

Return a controller with the best available backend.

Parameters:

• pin_map – Optional override for the line→pin assignment.

• force_mock – If True, never attempt the real backend.

Returns:

A GPIOController ready to drive lines.

Video and image capture wrappers.

VideoCapture spawns an ffmpeg process for the requested duration; ImageCapture grabs single frames via pygame.camera. Both expose a uniform record() method that writes a single timestamped file into the session directory.

The capture backends are interchangeable from the experiment orchestrator’s perspective — picking video vs images is a CLI decision.

class currentsensation.capture.CaptureBackend(*args, **kwargs)

Any capture backend the orchestrator can drive.

close() → None

record(save_dir: Path, line: str, duration_s: float) → Path

class currentsensation.capture.ImageCapture(config: CaptureConfig = CaptureConfig(width=640, height=480, fps=25, device='/dev/video0'))

pygame.camera-based single-frame grabber.

close() → None

record(save_dir: Path, line: str, duration_s: float = 0.0) → Path

Grab one frame and write it as JPEG.

class currentsensation.capture.NullCapture(extension: str = '.avi')

No-op capture backend that only logs what would have been recorded.

close() → None

No-op.

extension: str = '.avi'

record(save_dir: Path, line: str, duration_s: float) → Path

class currentsensation.capture.VideoCapture(config: CaptureConfig = CaptureConfig(width=640, height=480, fps=25, device='/dev/video0'))

ffmpeg-based video recorder.

close() → None

No-op; ffmpeg processes terminate themselves at -t.

config: CaptureConfig = CaptureConfig(width=640, height=480, fps=25, device='/dev/video0')

record(save_dir: Path, line: str, duration_s: float) → Path

Spawn ffmpeg to record duration_s seconds into a new file.

currentsensation.capture.build_filename(save_dir: Path, line: str, extension: str) → Path

Compose <save_dir>/<timestamp>-<line><ext> with current time.

Build a deterministic, replayable experiment timeline.

The schedule is the single source of truth for what happens when. The orchestrator merely walks it; the hardware modules merely react. Splitting this out means you can dry-run an experiment, inspect the timeline, and unit-test it without ever touching a GPIO library.

The session layout follows the 2017 original:

` [ off-trial ]            ← baseline [ rep 1: line A, B, C ]  ← order randomised if requested [ rep 2: line A, B, C ] ... [ off-trial ]            ← recovery baseline [ finish ] `

Each trial consists of a shake stimulus immediately followed by a current line being switched on for stim_dur seconds, with a video (or stream of images) recorded across the shake-and-stim interval.

class currentsensation.scheduling.ScheduleConfig(timing: ~currentsensation.constants.ExperimentTiming = <factory>, lines: tuple[str, ...] = ('red', 'blue', 'yellow'), randomise_lines: bool = True, mode: ~typing.Literal['video', 'image'] = 'video', image_fps: int = 2)

Inputs to build_schedule().

image_fps: int = 2

lines: tuple[str, ...] = ('red', 'blue', 'yellow')

mode: Literal['video', 'image'] = 'video'

randomise_lines: bool = True

timing: ExperimentTiming

class currentsensation.scheduling.TimedEvent(offset_s: float, kind: ~typing.Literal['pin_high', 'pin_low', 'change_line', 'capture_video', 'capture_image', 'finish'], payload: dict = <factory>, priority: int = 2)

A single thing-to-do at a given offset from session start.

offset_s

Seconds from the start of the session.

Type:

float

kind

What kind of event this is (drives orchestrator dispatch).

Type:

Literal[‘pin_high’, ‘pin_low’, ‘change_line’, ‘capture_video’, ‘capture_image’, ‘finish’]

payload

Event-specific arguments.

Type:

dict

priority

Tiebreaker for events at the same offset (lower fires first).

Type:

int

kind: Literal['pin_high', 'pin_low', 'change_line', 'capture_video', 'capture_image', 'finish']

offset_s: float

payload: dict

priority: int = 2

currentsensation.scheduling.build_schedule(config: ScheduleConfig, rng: Random | None = None) → list[TimedEvent]

Generate the full event timeline for one experimental session.

Parameters:

• config – Timing, line set, capture mode.

• rng – Random source for line-order shuffling. Defaults to a fresh random.Random so tests can pass a seeded one.

Returns:

List of TimedEvent sorted by (offset_s, priority).

currentsensation.scheduling.default_schedule_path(save_dir: Path) → Path

Where the schedule text file lives within a session directory.

currentsensation.scheduling.serialise_schedule(events: list[TimedEvent], out_path: Path, session_start: datetime | None = None) → Path

Write a human-readable schedule file mirroring the 2017 format.

Run an experimental session given a schedule and the hardware to drive.

class currentsensation.experiment.Experiment(controller: GPIOController, capture: CaptureBackend, paths: SessionPaths)

Drive a single experimental session.

capture: CaptureBackend

controller: GPIOController

paths: SessionPaths

run(config: ScheduleConfig, *, dry_run: bool = False) → list[TimedEvent]

Build a schedule, write it to disk, and (optionally) execute it.

Parameters:

• config – What to schedule.

• dry_run – If True, do not enter the scheduler loop; just write the schedule file and return the event list.

Returns:

The sorted event list that was scheduled.

exception currentsensation.experiment.ScheduleFinished

Raised inside a scheduled callback to terminate the session cleanly.

Command-line entry point for currentsensation run.

currentsensation.cli.main(argv: list[str] | None = None) → int

Parse arguments, build the experiment, dispatch.