import aeon.io.reader as _reader
from aeon.schema.streams import Stream, StreamGroup
[docs]
class Photodiode(Stream):
def __init__(self, path):
super().__init__(_reader.Harp(f"{path}_44_*", columns=["adc", "encoder"]))
[docs]
class OSC(StreamGroup):
def __init__(self, path):
super().__init__(path)
[docs]
class BackgroundColor(Stream):
def __init__(self, pattern):
super().__init__(
_reader.Csv(f"{pattern}_backgroundcolor_*", columns=["typetag", "r", "g", "b", "a"])
)
[docs]
class ChangeSubjectState(Stream):
def __init__(self, pattern):
super().__init__(
_reader.Csv(f"{pattern}_changesubjectstate_*", columns=["typetag", "id", "weight", "event"])
)
[docs]
class EndTrial(Stream):
def __init__(self, pattern):
super().__init__(_reader.Csv(f"{pattern}_endtrial_*", columns=["typetag", "value"]))
[docs]
class Slice(Stream):
def __init__(self, pattern):
super().__init__(
_reader.Csv(
f"{pattern}_octagonslice_*", columns=["typetag", "wall_id", "r", "g", "b", "a", "delay"]
)
)
[docs]
class GratingsSlice(Stream):
def __init__(self, pattern):
super().__init__(
_reader.Csv(
f"{pattern}_octagongratingsslice_*",
columns=[
"typetag",
"wall_id",
"contrast",
"opacity",
"spatial_frequency",
"temporal_frequency",
"angle",
"delay",
],
)
)
[docs]
class Poke(Stream):
def __init__(self, pattern):
super().__init__(
_reader.Csv(
f"{pattern}_poke_*",
columns=[
"typetag",
"wall_id",
"poke_id",
"reward",
"reward_interval",
"delay",
"led_delay",
],
)
)
[docs]
class Response(Stream):
def __init__(self, pattern):
super().__init__(
_reader.Csv(
f"{pattern}_response_*", columns=["typetag", "wall_id", "poke_id", "response_time"]
)
)
[docs]
class RunPreTrialNoPoke(Stream):
def __init__(self, pattern):
super().__init__(
_reader.Csv(
f"{pattern}_run_pre_no_poke_*",
columns=[
"typetag",
"wait_for_poke",
"reward_iti",
"timeout_iti",
"pre_trial_duration",
"activity_reset_flag",
],
)
)
[docs]
class StartNewSession(Stream):
def __init__(self, pattern):
super().__init__(_reader.Csv(f"{pattern}_startnewsession_*", columns=["typetag", "path"]))
[docs]
class TaskLogic(StreamGroup):
def __init__(self, path):
super().__init__(path)
[docs]
class TrialInitiation(Stream):
def __init__(self, pattern):
super().__init__(_reader.Harp(f"{pattern}_1_*", columns=["trial_type"]))
[docs]
class Response(Stream):
def __init__(self, pattern):
super().__init__(_reader.Harp(f"{pattern}_2_*", columns=["wall_id", "poke_id"]))
[docs]
class PreTrialState(Stream):
def __init__(self, pattern):
super().__init__(_reader.Harp(f"{pattern}_3_*", columns=["state"]))
[docs]
class InterTrialInterval(Stream):
def __init__(self, pattern):
super().__init__(_reader.Harp(f"{pattern}_4_*", columns=["state"]))
[docs]
class SliceOnset(Stream):
def __init__(self, pattern):
super().__init__(_reader.Harp(f"{pattern}_10_*", columns=["wall_id"]))
[docs]
class DrawBackground(Stream):
def __init__(self, pattern):
super().__init__(_reader.Harp(f"{pattern}_11_*", columns=["state"]))
[docs]
class GratingsSliceOnset(Stream):
def __init__(self, pattern):
super().__init__(_reader.Harp(f"{pattern}_12_*", columns=["wall_id"]))
[docs]
class Wall(StreamGroup):
def __init__(self, path):
super().__init__(path)
[docs]
class BeamBreak0(Stream):
def __init__(self, pattern):
super().__init__(_reader.DigitalBitmask(f"{pattern}_32_*", 0x1, columns=["state"]))
[docs]
class BeamBreak1(Stream):
def __init__(self, pattern):
super().__init__(_reader.DigitalBitmask(f"{pattern}_32_*", 0x2, columns=["state"]))
[docs]
class BeamBreak2(Stream):
def __init__(self, pattern):
super().__init__(_reader.DigitalBitmask(f"{pattern}_32_*", 0x4, columns=["state"]))
[docs]
class SetLed0(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_34_*", 0x1, "Set"))
[docs]
class SetLed1(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_34_*", 0x2, "Set"))
[docs]
class SetLed2(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_34_*", 0x4, "Set"))
[docs]
class SetValve0(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_34_*", 0x8, "Set"))
[docs]
class SetValve1(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_34_*", 0x10, "Set"))
[docs]
class SetValve2(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_34_*", 0x20, "Set"))
[docs]
class ClearLed0(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x1, "Clear"))
[docs]
class ClearLed1(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x2, "Clear"))
[docs]
class ClearLed2(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x4, "Clear"))
[docs]
class ClearValve0(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x8, "Clear"))
[docs]
class ClearValve1(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x10, "Clear"))
[docs]
class ClearValve2(Stream):
def __init__(self, pattern):
super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x20, "Clear"))