Moving Stimulus Responses

Rendering

flyvision.datasets.moving_bar.RenderedOffsets

Bases: Directory

Rendered offsets for the moving bar stimulus.

This class precomputes the offsets for moving bar (edge) stimuli and stores them in a directory. At runtime, the offsets are resampled to efficiently generate stimuli with different durations and temporal resolutions.

Parameters:

Name	Type	Description	Default
offsets	list[int]	List of offset values.	list(range(-10, 11))
angles	list[int]	List of angle values in degrees.	[0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330]
widths	list[int]	List of width values.	[1, 2, 4]
intensities	list[int]	List of intensity values.	[0, 1]
led_width	float	Width of LED in radians.	radians(2.25)
height	float	Height of the bar in radians.	radians(2.25) * 9
n_bars	int	Number of bars.	1
bg_intensity	float	Background intensity.	0.5
bar_loc_horizontal	float	Horizontal location of the bar in radians.	radians(90)

Attributes:

Name	Type	Description
offsets	ArrayFile	Rendered offsets for different stimulus parameters.

Source code in flyvision/datasets/moving_bar.py

@root(renderings_dir)
class RenderedOffsets(Directory):
    """Rendered offsets for the moving bar stimulus.

    This class precomputes the offsets for moving bar (edge) stimuli and stores them
    in a directory. At runtime, the offsets are resampled to efficiently generate
    stimuli with different durations and temporal resolutions.

    Args:
        offsets: List of offset values.
        angles: List of angle values in degrees.
        widths: List of width values.
        intensities: List of intensity values.
        led_width: Width of LED in radians.
        height: Height of the bar in radians.
        n_bars: Number of bars.
        bg_intensity: Background intensity.
        bar_loc_horizontal: Horizontal location of the bar in radians.

    Attributes:
        offsets (ArrayFile): Rendered offsets for different stimulus parameters.
    """

    def __init__(
        self,
        offsets: list[int] = list(range(-10, 11)),
        angles: list[int] = [0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330],
        widths: list[int] = [1, 2, 4],
        intensities: list[int] = [0, 1],
        led_width: float = np.radians(2.25),
        height: float = np.radians(2.25) * 9,
        n_bars: int = 1,
        bg_intensity: float = 0.5,
        bar_loc_horizontal: float = np.radians(90),
    ):
        eye = HexEye(721, 25)

        params = list(product(angles, widths, intensities))

        sequences = {}
        _tqdm = tqdm(total=len(params), desc="building stimuli")
        for angle in angles:
            for width in widths:
                for intensity in intensities:
                    offset_bars = eye.render_offset_bars(
                        bar_width_rad=width * led_width,
                        bar_height_rad=height,
                        n_bars=n_bars,
                        offsets=np.array(offsets) * led_width,
                        bar_intensity=intensity,
                        bg_intensity=bg_intensity,
                        moving_angle=angle,
                        bar_loc_horizontal=bar_loc_horizontal,
                    )
                    sequences[(angle, width, intensity)] = offset_bars
                    _tqdm.update()

        _tqdm.close()
        self.offsets = torch.stack([sequences[p] for p in params]).cpu().numpy()

Datasets

flyvision.datasets.moving_bar.MovingBar

Bases: StimulusDataset

Moving bar stimulus.

Parameters:

Name	Type	Description	Default
widths	list[int]	Width of the bar in half ommatidia.	[1, 2, 4]
offsets	tuple[int, int]	First and last offset to the central column in half ommatidia.	(-10, 11)
intensities	list[float]	Intensity of the bar.	[0, 1]
speeds	list[float]	Speed of the bar in half ommatidia per second.	[2.4, 4.8, 9.7, 13, 19, 25]
height	int	Height of the bar in half ommatidia.	9
dt	float	Time step in seconds.	1 / 200
device	str	Device to store the stimulus.	device
bar_loc_horizontal	float	Horizontal location of the bar in radians from left to right of image plane. np.radians(90) is the center.	radians(90)
post_pad_mode	Literal['continue', 'value', 'reflect']	Padding mode after the stimulus. One of ‘continue’, ‘value’, ‘reflect’. If ‘value’ the padding is filled with bg_intensity.	'value'
t_pre	float	Time before the stimulus in seconds.	1.0
t_post	float	Time after the stimulus in seconds.	1.0
build_stim_on_init	bool	Build the stimulus on initialization.	True
shuffle_offsets	bool	Shuffle the offsets to remove spatio-temporal correlation.	False
seed	int	Seed for the random state.	0
angles	list[int]	List of angles in degrees.	[0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330]

Attributes:

Name	Type	Description
config	Namespace	Configuration parameters.
omm_width	float	Width of ommatidium in radians.
led_width	float	Width of LED in radians.
angles	ndarray	Array of angles in degrees.
widths	ndarray	Array of widths in half ommatidia.
offsets	ndarray	Array of offsets in half ommatidia.
intensities	ndarray	Array of intensities.
speeds	ndarray	Array of speeds in half ommatidia per second.
bg_intensity	float	Background intensity.
n_bars	int	Number of bars.
bar_loc_horizontal	float	Horizontal location of bar in radians.
t_stim	ndarray	Stimulation times for each speed.
t_stim_max	float	Maximum stimulation time.
height	float	Height of bar in radians.
post_pad_mode	str	Padding mode after the stimulus.
arg_df	DataFrame	DataFrame of stimulus parameters.
arg_group_df	DataFrame	Grouped DataFrame of stimulus parameters.
device	str	Device for storing stimuli.
shuffle_offsets	bool	Whether to shuffle offsets.
randomstate	RandomState	Random state for shuffling.

Source code in flyvision/datasets/moving_bar.py

class MovingBar(StimulusDataset):
    """Moving bar stimulus.

    Args:
        widths: Width of the bar in half ommatidia.
        offsets: First and last offset to the central column in half ommatidia.
        intensities: Intensity of the bar.
        speeds: Speed of the bar in half ommatidia per second.
        height: Height of the bar in half ommatidia.
        dt: Time step in seconds.
        device: Device to store the stimulus.
        bar_loc_horizontal: Horizontal location of the bar in radians from left to
            right of image plane. np.radians(90) is the center.
        post_pad_mode: Padding mode after the stimulus. One of 'continue', 'value',
            'reflect'. If 'value' the padding is filled with `bg_intensity`.
        t_pre: Time before the stimulus in seconds.
        t_post: Time after the stimulus in seconds.
        build_stim_on_init: Build the stimulus on initialization.
        shuffle_offsets: Shuffle the offsets to remove spatio-temporal correlation.
        seed: Seed for the random state.
        angles: List of angles in degrees.

    Attributes:
        config (Namespace): Configuration parameters.
        omm_width (float): Width of ommatidium in radians.
        led_width (float): Width of LED in radians.
        angles (np.ndarray): Array of angles in degrees.
        widths (np.ndarray): Array of widths in half ommatidia.
        offsets (np.ndarray): Array of offsets in half ommatidia.
        intensities (np.ndarray): Array of intensities.
        speeds (np.ndarray): Array of speeds in half ommatidia per second.
        bg_intensity (float): Background intensity.
        n_bars (int): Number of bars.
        bar_loc_horizontal (float): Horizontal location of bar in radians.
        t_stim (np.ndarray): Stimulation times for each speed.
        t_stim_max (float): Maximum stimulation time.
        height (float): Height of bar in radians.
        post_pad_mode (str): Padding mode after the stimulus.
        arg_df (pd.DataFrame): DataFrame of stimulus parameters.
        arg_group_df (pd.DataFrame): Grouped DataFrame of stimulus parameters.
        device (str): Device for storing stimuli.
        shuffle_offsets (bool): Whether to shuffle offsets.
        randomstate (np.random.RandomState): Random state for shuffling.
    """

    arg_df: pd.DataFrame = None

    def __init__(
        self,
        widths: list[int] = [1, 2, 4],
        offsets: tuple[int, int] = (-10, 11),
        intensities: list[float] = [0, 1],
        speeds: list[float] = [2.4, 4.8, 9.7, 13, 19, 25],
        height: int = 9,
        dt: float = 1 / 200,
        device: str = flyvision.device,
        bar_loc_horizontal: float = np.radians(90),
        post_pad_mode: Literal["continue", "value", "reflect"] = "value",
        t_pre: float = 1.0,
        t_post: float = 1.0,
        build_stim_on_init: bool = True,
        shuffle_offsets: bool = False,
        seed: int = 0,
        angles: list[int] = [0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330],
    ) -> None:
        super().__init__()
        # HexEye parameter
        self.omm_width = np.radians(5.8)  # radians(5.8 degree)

        # Monitor parameter
        self.led_width = np.radians(2.25)  # Gruntman et al. 2018

        _locals = locals()
        self.config = Namespace({
            arg: _locals[arg]
            for arg in [
                "widths",
                "offsets",
                "intensities",
                "speeds",
                "height",
                "bar_loc_horizontal",
                "shuffle_offsets",
                "post_pad_mode",
                "t_pre",
                "t_post",
                "dt",
                "angles",
            ]
        })

        # Stim parameter
        self.angles = np.array(angles)
        self.widths = np.array(widths)  # half ommatidia
        if len(offsets) == 2:
            self.offsets = np.arange(*offsets)  # half ommatidia
        else:
            assert (
                np.mean(offsets[1:] - offsets[:-1]) == 1
            )  # t_stim assumes spacing of 1 corresponding to 2.25 deg
            self.offsets = offsets
        self.intensities = np.array(intensities)
        self.speeds = np.array(speeds)
        self.bg_intensity = 0.5
        self.n_bars = 1
        self.bar_loc_horizontal = bar_loc_horizontal

        self.t_stim = (len(self.offsets) * self.led_width) / (
            self.speeds * self.omm_width
        )
        self.t_stim_max = np.max(self.t_stim)

        self._speed_to_t_stim = dict(zip(self.speeds, self.t_stim))

        self.height = self.led_width * height

        self.post_pad_mode = post_pad_mode
        self._t_pre = t_pre
        self._t_post = t_post

        params = [
            (*p[:-1], *p[-1])
            for p in list(
                product(
                    self.angles,
                    self.widths,
                    self.intensities,
                    zip(self.t_stim, self.speeds),
                )
            )
        ]
        self.arg_df = pd.DataFrame(
            params, columns=["angle", "width", "intensity", "t_stim", "speed"]
        )

        self.arg_group_df = self.arg_df.groupby(
            ["angle", "width", "intensity"], sort=False, as_index=False
        ).all()

        self.device = device
        self.shuffle_offsets = shuffle_offsets
        self.randomstate = None
        if self.shuffle_offsets:
            self.randomstate = np.random.RandomState(seed=seed)

        self._dt = dt

        self._built = False
        if build_stim_on_init:
            self._build()
            self._resample()
            self._built = True

    @property
    def dt(self) -> float:
        """Time step in seconds."""
        return getattr(self, "_dt", None)

    @dt.setter
    def dt(self, value: float) -> None:
        if self._dt == value:
            self._dt = value
            if self._built:
                self._resample()
            return
        logging.warning(
            "Cannot override dt=%s because responses with dt=%s are initialized. "
            "Keeping dt=%s.",
            value,
            self._dt,
            self._dt,
        )

    def __repr__(self) -> str:
        return (
            f"{self.__class__.__name__}\n"
            + "Config:\n"
            + repr(self.config)
            + "Stimulus parameter:\n"
            + repr(self.arg_df)
        )

    @property
    def t_pre(self) -> float:
        """Time before stimulus onset in seconds."""
        return self._t_pre

    @property
    def t_post(self) -> float:
        """Time after stimulus offset in seconds."""
        return self._t_post

    def _build(self) -> None:
        """Build the stimulus."""
        self.wrap = RenderedOffsets(
            dict(
                angles=self.angles,
                widths=self.widths,
                intensities=self.intensities,
                offsets=self.offsets,
                led_width=self.led_width,
                height=self.height,
                n_bars=self.n_bars,
                bg_intensity=self.bg_intensity,
                bar_loc_horizontal=self.bar_loc_horizontal,
            )
        )

        self._offsets = torch.tensor(self.wrap.offsets[:], device=self.device)
        self._built = True

    def _resample(self) -> None:
        """Resample the stimulus at runtime."""
        # resampling at runtime to allow for changing dt and to save GB of
        # storage.
        self.sequences = {}
        self.indices = {}
        for t, speed in zip(self.t_stim, self.speeds):
            sequence, indices = resample(
                self._offsets,
                t,
                self.dt,
                dim=1,
                device=self.device,
                return_indices=True,
            )
            if self.shuffle_offsets:
                # breakpoint()
                sequence = shuffle(sequence, self.randomstate)
            sequence = pad(
                sequence,
                t + self.t_pre,
                self.dt,
                mode="start",
                fill=self.bg_intensity,
            )
            sequence = pad(
                sequence,
                t + self.t_pre + self.t_post,
                self.dt,
                mode="end",
                pad_mode=self.post_pad_mode,
                fill=self.bg_intensity,
            )
            # Because we fix the distance that the bar moves but vary speeds we
            # have different stimulation times. To make all sequences equal
            # length for storing them in a single tensor, we pad them with nans
            # based on the maximal stimulation time (slowest speed). The nans
            # can later be removed before processing the traces.
            sequence = pad(
                sequence,
                self.t_stim_max + self.t_pre + self.t_post,
                self.dt,
                mode="end",
                fill=np.nan,
            )
            self.sequences[speed] = sequence
            self.indices[speed] = indices

    def _key(self, angle: float, width: float, intensity: float, speed: float) -> int:
        """Get the key for a specific stimulus configuration."""
        try:
            return self.arg_df.query(
                f"angle=={angle}"
                f" & width=={width}"
                f" & intensity == {intensity}"
                f" & speed == {speed}"
            ).index.values.item()
        except ValueError:
            raise ValueError(
                f"angle: {angle}, width: {width}, intensity: {intensity}, "
                f"speed: {speed} invalid."
            ) from None

    def get_sequence_id_from_arguments(
        self, angle: float, width: float, intensity: float, speed: float
    ) -> int:
        """Get sequence ID from stimulus arguments."""
        return self.get_stimulus_index(locals())

    def _params(self, key: int) -> np.ndarray:
        """Get parameters for a given key."""
        return self.arg_df.iloc[key].values

    def _group_key(self, angle: float, width: float, intensity: float) -> int:
        """Get group key for a specific stimulus configuration."""
        return self.arg_group_df.query(
            f"angle=={angle}" f" & width=={width}" f" & intensity == {intensity}"
        ).index.values.item()

    def _group_params(self, key: int) -> np.ndarray:
        """Get group parameters for a given key."""
        return self.arg_group_df.iloc[key].values

    def get(
        self, angle: float, width: float, intensity: float, speed: float
    ) -> torch.Tensor:
        """Get stimulus for specific parameters."""
        key = self._key(angle, width, intensity, speed)
        return self[key]

    def get_item(self, key: int) -> torch.Tensor:
        """Get stimulus for a specific key."""
        angle, width, intensity, _, speed = self._params(key)
        return self.sequences[speed][self._group_key(angle, width, intensity)]

    def mask(
        self,
        angle: Optional[float] = None,
        width: Optional[float] = None,
        intensity: Optional[float] = None,
        speed: Optional[float] = None,
        t_stim: Optional[float] = None,
    ) -> np.ndarray:
        """Create a mask for specific stimulus parameters."""
        # 22x faster than df.query
        values = self.arg_df.values

        def iterparam(param, name, axis, and_condition):
            condition = np.zeros(len(values)).astype(bool)
            if isinstance(param, Iterable):
                for p in param:
                    _new = values.take(axis, axis=1) == p
                    assert any(_new), f"{name} {p} not in dataset."
                    condition = np.logical_or(condition, _new)
            else:
                _new = values.take(axis, axis=1) == param
                assert any(_new), f"{name} {param} not in dataset."
                condition = np.logical_or(condition, _new)
            return condition & and_condition

        condition = np.ones(len(values)).astype(bool)
        if angle is not None:
            condition = iterparam(angle, "angle", 0, condition)
        if width is not None:
            condition = iterparam(width, "width", 1, condition)
        if intensity is not None:
            condition = iterparam(intensity, "intensity", 2, condition)
        if t_stim is not None:
            condition = iterparam(t_stim, "t_stim", 3, condition)
        if speed is not None:
            condition = iterparam(speed, "speed", 4, condition)
        return condition

    @property
    def time(self) -> np.ndarray:
        """Time array for the stimulus."""
        return np.arange(-self.t_pre, self.t_stim_max + self.t_post - self.dt, self.dt)

    def stimulus(
        self,
        angle: Optional[float] = None,
        width: Optional[float] = None,
        intensity: Optional[float] = None,
        speed: Optional[float] = None,
        pre_stim: bool = True,
        post_stim: bool = True,
    ) -> np.ndarray:
        """Get stimulus for specific parameters.

        Args:
            angle: Angle of the bar.
            width: Width of the bar.
            intensity: Intensity of the bar.
            speed: Speed of the bar.
            pre_stim: Include pre-stimulus period.
            post_stim: Include post-stimulus period.

        Returns:
            Stimulus array.
        """
        key = self._key(angle, width, intensity, speed)
        stim = self[key][:, 360].cpu().numpy()
        if not post_stim:
            stim = filter_post([stim], self.t_post, self.dt).squeeze()
        if not pre_stim:
            stim = filter_pre(stim[None], self.t_pre, self.dt).squeeze()
        return stim

    def stimulus_parameters(
        self,
        angle: Optional[float] = None,
        width: Optional[float] = None,
        intensity: Optional[float] = None,
        speed: Optional[float] = None,
    ) -> tuple[list, ...]:
        """Get stimulus parameters."""

        def _number_to_list(*args):
            returns = tuple()
            for arg in args:
                if isinstance(arg, Number):
                    returns += ([arg],)
                else:
                    returns += (arg,)
            return returns

        angle, width, speed, intensity = _number_to_list(angle, width, speed, intensity)
        angle = angle or self.angles
        width = width or self.widths
        intensity = intensity or self.intensities
        speed = speed or self.speeds
        return angle, width, intensity, speed

    def sample_shape(
        self,
        angle: Optional[float] = None,
        width: Optional[float] = None,
        intensity: Optional[float] = None,
        speed: Optional[float] = None,
    ) -> tuple[int, ...]:
        """Get shape of stimulus sample for given parameters."""
        if isinstance(angle, Number):
            angle = [angle]
        if isinstance(width, Number):
            width = [width]
        if isinstance(speed, Number):
            speed = [speed]
        if isinstance(intensity, Number):
            intensity = [intensity]
        angle = angle or self.angles
        width = width or self.widths
        intensity = intensity or self.intensities
        speed = speed or self.speeds
        return (
            len(angle),
            len(width),
            len(intensity),
            len(speed),
        )

    def time_to_center(self, speed: float) -> float:
        """Calculate time for bar to reach center at given speed."""
        # Note: time = distance / velocity, i.e.
        #     time = (n_leds * led_width) / (speed * omm_width)
        #     with speed in ommatidia / s.
        return np.abs(self.config.offsets[0]) * self.led_width / (speed * self.omm_width)

    def get_time_with_origin_at_onset(self) -> np.ndarray:
        """Get time array with origin at stimulus onset."""
        return np.linspace(
            -self.t_pre,
            self.t_stim_max - self.t_pre + self.t_post,
            int(self.t_stim_max / self.dt)
            + int(self.t_post / self.dt)
            + int(self.t_pre / self.dt),
        )

    def get_time_with_origin_at_center(self, speed: float) -> np.ndarray:
        """Get time array with origin where bar reaches central column."""
        time_to_center = self.time_to_center(speed)
        n_steps = (
            int(self.t_stim_max / self.dt)
            + int(self.t_post / self.dt)
            + int(self.t_pre / self.dt)
        )
        return np.linspace(
            -(self.t_pre + time_to_center),
            n_steps * self.dt - (self.t_pre + time_to_center),
            n_steps,
        )

    def stimulus_cartoon(
        self,
        angle: float,
        width: float,
        intensity: float,
        speed: float,
        time_after_stimulus_onset: float = 0.5,
        fig: Optional[plt.Figure] = None,
        ax: Optional[plt.Axes] = None,
        facecolor: str = "#000000",
        cmap: Colormap = plt.cm.bone,
        alpha: float = 0.5,
        vmin: float = 0,
        vmax: float = 1,
        edgecolor: str = "none",
        central_hex_color: str = "#2f7cb9",
        **kwargs,
    ) -> tuple[plt.Figure, plt.Axes]:
        """Create a cartoon representation of the stimulus."""
        fig, ax = init_plot(fig=fig, ax=ax)

        time = (
            np.arange(
                0,
                self.t_pre + self.t_stim_max + self.t_post - self.dt,
                self.dt,
            )
            - self.t_pre
        )
        index = np.argmin(np.abs(time - time_after_stimulus_onset))

        fig, ax, _ = quick_hex_scatter(
            self.get(angle=angle, width=width, speed=speed, intensity=intensity)
            .cpu()
            .numpy()[index],
            vmin=vmin,
            vmax=vmax,
            cbar=False,
            figsize=[1, 1],
            max_extent=5,
            fig=fig,
            ax=ax,
            cmap=cmap,
            alpha=alpha,
            edgecolor=edgecolor,
            **kwargs,
        )
        rotation = np.array([
            [
                np.cos(np.radians(angle - 90)),
                -np.sin(np.radians(angle - 90)),
            ],
            [
                np.sin(np.radians(angle - 90)),
                np.cos(np.radians(angle - 90)),
            ],
        ])
        x = rotation @ np.array([0, -5])
        dx = rotation @ np.array([0, 1])
        ax.arrow(
            *x,
            *dx,
            facecolor=facecolor,
            width=0.75,
            head_length=2.5,
            edgecolor="k",
            linewidth=0.25,
        )
        _hex = RegularPolygon(
            (0, 0),
            numVertices=6,
            radius=1,
            linewidth=1,
            orientation=np.radians(30),
            edgecolor=central_hex_color,
            facecolor=central_hex_color,
            alpha=1,
            ls="-",
        )
        ax.add_patch(_hex)

        return fig, ax

dt property writable

dt

Time step in seconds.

t_pre property

t_pre

Time before stimulus onset in seconds.

t_post property

t_post

Time after stimulus offset in seconds.

time property

time

Time array for the stimulus.

get_sequence_id_from_arguments

get_sequence_id_from_arguments(
    angle, width, intensity, speed
)

Get sequence ID from stimulus arguments.

Source code in flyvision/datasets/moving_bar.py

def get_sequence_id_from_arguments(
    self, angle: float, width: float, intensity: float, speed: float
) -> int:
    """Get sequence ID from stimulus arguments."""
    return self.get_stimulus_index(locals())

get

get(angle, width, intensity, speed)

Get stimulus for specific parameters.

Source code in flyvision/datasets/moving_bar.py

def get(
    self, angle: float, width: float, intensity: float, speed: float
) -> torch.Tensor:
    """Get stimulus for specific parameters."""
    key = self._key(angle, width, intensity, speed)
    return self[key]

get_item

get_item(key)

Get stimulus for a specific key.

Source code in flyvision/datasets/moving_bar.py

def get_item(self, key: int) -> torch.Tensor:
    """Get stimulus for a specific key."""
    angle, width, intensity, _, speed = self._params(key)
    return self.sequences[speed][self._group_key(angle, width, intensity)]

mask

mask(
    angle=None,
    width=None,
    intensity=None,
    speed=None,
    t_stim=None,
)

Create a mask for specific stimulus parameters.

Source code in flyvision/datasets/moving_bar.py

def mask(
    self,
    angle: Optional[float] = None,
    width: Optional[float] = None,
    intensity: Optional[float] = None,
    speed: Optional[float] = None,
    t_stim: Optional[float] = None,
) -> np.ndarray:
    """Create a mask for specific stimulus parameters."""
    # 22x faster than df.query
    values = self.arg_df.values

    def iterparam(param, name, axis, and_condition):
        condition = np.zeros(len(values)).astype(bool)
        if isinstance(param, Iterable):
            for p in param:
                _new = values.take(axis, axis=1) == p
                assert any(_new), f"{name} {p} not in dataset."
                condition = np.logical_or(condition, _new)
        else:
            _new = values.take(axis, axis=1) == param
            assert any(_new), f"{name} {param} not in dataset."
            condition = np.logical_or(condition, _new)
        return condition & and_condition

    condition = np.ones(len(values)).astype(bool)
    if angle is not None:
        condition = iterparam(angle, "angle", 0, condition)
    if width is not None:
        condition = iterparam(width, "width", 1, condition)
    if intensity is not None:
        condition = iterparam(intensity, "intensity", 2, condition)
    if t_stim is not None:
        condition = iterparam(t_stim, "t_stim", 3, condition)
    if speed is not None:
        condition = iterparam(speed, "speed", 4, condition)
    return condition

stimulus

stimulus(
    angle=None,
    width=None,
    intensity=None,
    speed=None,
    pre_stim=True,
    post_stim=True,
)

Get stimulus for specific parameters.

Parameters:

Name	Type	Description	Default
angle	Optional[float]	Angle of the bar.	None
width	Optional[float]	Width of the bar.	None
intensity	Optional[float]	Intensity of the bar.	None
speed	Optional[float]	Speed of the bar.	None
pre_stim	bool	Include pre-stimulus period.	True
post_stim	bool	Include post-stimulus period.	True

Returns:

Type	Description
ndarray	Stimulus array.

Source code in flyvision/datasets/moving_bar.py

def stimulus(
    self,
    angle: Optional[float] = None,
    width: Optional[float] = None,
    intensity: Optional[float] = None,
    speed: Optional[float] = None,
    pre_stim: bool = True,
    post_stim: bool = True,
) -> np.ndarray:
    """Get stimulus for specific parameters.

    Args:
        angle: Angle of the bar.
        width: Width of the bar.
        intensity: Intensity of the bar.
        speed: Speed of the bar.
        pre_stim: Include pre-stimulus period.
        post_stim: Include post-stimulus period.

    Returns:
        Stimulus array.
    """
    key = self._key(angle, width, intensity, speed)
    stim = self[key][:, 360].cpu().numpy()
    if not post_stim:
        stim = filter_post([stim], self.t_post, self.dt).squeeze()
    if not pre_stim:
        stim = filter_pre(stim[None], self.t_pre, self.dt).squeeze()
    return stim

stimulus_parameters

stimulus_parameters(
    angle=None, width=None, intensity=None, speed=None
)

Get stimulus parameters.

Source code in flyvision/datasets/moving_bar.py

def stimulus_parameters(
    self,
    angle: Optional[float] = None,
    width: Optional[float] = None,
    intensity: Optional[float] = None,
    speed: Optional[float] = None,
) -> tuple[list, ...]:
    """Get stimulus parameters."""

    def _number_to_list(*args):
        returns = tuple()
        for arg in args:
            if isinstance(arg, Number):
                returns += ([arg],)
            else:
                returns += (arg,)
        return returns

    angle, width, speed, intensity = _number_to_list(angle, width, speed, intensity)
    angle = angle or self.angles
    width = width or self.widths
    intensity = intensity or self.intensities
    speed = speed or self.speeds
    return angle, width, intensity, speed

sample_shape

sample_shape(
    angle=None, width=None, intensity=None, speed=None
)

Get shape of stimulus sample for given parameters.

Source code in flyvision/datasets/moving_bar.py

def sample_shape(
    self,
    angle: Optional[float] = None,
    width: Optional[float] = None,
    intensity: Optional[float] = None,
    speed: Optional[float] = None,
) -> tuple[int, ...]:
    """Get shape of stimulus sample for given parameters."""
    if isinstance(angle, Number):
        angle = [angle]
    if isinstance(width, Number):
        width = [width]
    if isinstance(speed, Number):
        speed = [speed]
    if isinstance(intensity, Number):
        intensity = [intensity]
    angle = angle or self.angles
    width = width or self.widths
    intensity = intensity or self.intensities
    speed = speed or self.speeds
    return (
        len(angle),
        len(width),
        len(intensity),
        len(speed),
    )

time_to_center

time_to_center(speed)

Calculate time for bar to reach center at given speed.

Source code in flyvision/datasets/moving_bar.py

def time_to_center(self, speed: float) -> float:
    """Calculate time for bar to reach center at given speed."""
    # Note: time = distance / velocity, i.e.
    #     time = (n_leds * led_width) / (speed * omm_width)
    #     with speed in ommatidia / s.
    return np.abs(self.config.offsets[0]) * self.led_width / (speed * self.omm_width)

get_time_with_origin_at_onset

get_time_with_origin_at_onset()

Get time array with origin at stimulus onset.

Source code in flyvision/datasets/moving_bar.py

def get_time_with_origin_at_onset(self) -> np.ndarray:
    """Get time array with origin at stimulus onset."""
    return np.linspace(
        -self.t_pre,
        self.t_stim_max - self.t_pre + self.t_post,
        int(self.t_stim_max / self.dt)
        + int(self.t_post / self.dt)
        + int(self.t_pre / self.dt),
    )

get_time_with_origin_at_center

get_time_with_origin_at_center(speed)

Get time array with origin where bar reaches central column.

Source code in flyvision/datasets/moving_bar.py

def get_time_with_origin_at_center(self, speed: float) -> np.ndarray:
    """Get time array with origin where bar reaches central column."""
    time_to_center = self.time_to_center(speed)
    n_steps = (
        int(self.t_stim_max / self.dt)
        + int(self.t_post / self.dt)
        + int(self.t_pre / self.dt)
    )
    return np.linspace(
        -(self.t_pre + time_to_center),
        n_steps * self.dt - (self.t_pre + time_to_center),
        n_steps,
    )

stimulus_cartoon

stimulus_cartoon(
    angle,
    width,
    intensity,
    speed,
    time_after_stimulus_onset=0.5,
    fig=None,
    ax=None,
    facecolor="#000000",
    cmap=plt.cm.bone,
    alpha=0.5,
    vmin=0,
    vmax=1,
    edgecolor="none",
    central_hex_color="#2f7cb9",
    **kwargs
)

Create a cartoon representation of the stimulus.

Source code in flyvision/datasets/moving_bar.py

def stimulus_cartoon(
    self,
    angle: float,
    width: float,
    intensity: float,
    speed: float,
    time_after_stimulus_onset: float = 0.5,
    fig: Optional[plt.Figure] = None,
    ax: Optional[plt.Axes] = None,
    facecolor: str = "#000000",
    cmap: Colormap = plt.cm.bone,
    alpha: float = 0.5,
    vmin: float = 0,
    vmax: float = 1,
    edgecolor: str = "none",
    central_hex_color: str = "#2f7cb9",
    **kwargs,
) -> tuple[plt.Figure, plt.Axes]:
    """Create a cartoon representation of the stimulus."""
    fig, ax = init_plot(fig=fig, ax=ax)

    time = (
        np.arange(
            0,
            self.t_pre + self.t_stim_max + self.t_post - self.dt,
            self.dt,
        )
        - self.t_pre
    )
    index = np.argmin(np.abs(time - time_after_stimulus_onset))

    fig, ax, _ = quick_hex_scatter(
        self.get(angle=angle, width=width, speed=speed, intensity=intensity)
        .cpu()
        .numpy()[index],
        vmin=vmin,
        vmax=vmax,
        cbar=False,
        figsize=[1, 1],
        max_extent=5,
        fig=fig,
        ax=ax,
        cmap=cmap,
        alpha=alpha,
        edgecolor=edgecolor,
        **kwargs,
    )
    rotation = np.array([
        [
            np.cos(np.radians(angle - 90)),
            -np.sin(np.radians(angle - 90)),
        ],
        [
            np.sin(np.radians(angle - 90)),
            np.cos(np.radians(angle - 90)),
        ],
    ])
    x = rotation @ np.array([0, -5])
    dx = rotation @ np.array([0, 1])
    ax.arrow(
        *x,
        *dx,
        facecolor=facecolor,
        width=0.75,
        head_length=2.5,
        edgecolor="k",
        linewidth=0.25,
    )
    _hex = RegularPolygon(
        (0, 0),
        numVertices=6,
        radius=1,
        linewidth=1,
        orientation=np.radians(30),
        edgecolor=central_hex_color,
        facecolor=central_hex_color,
        alpha=1,
        ls="-",
    )
    ax.add_patch(_hex)

    return fig, ax

flyvision.datasets.moving_bar.MovingEdge

Bases: MovingBar

Moving edge stimulus.

This class creates a moving edge stimulus by using a very wide bar.

Parameters:

Name	Type	Description	Default
offsets	tuple[int, int]	First and last offset to the central column in half ommatidia.	(-10, 11)
intensities	list[float]	Intensity of the edge.	[0, 1]
speeds	list[float]	Speed of the edge in half ommatidia per second.	[2.4, 4.8, 9.7, 13, 19, 25]
height	int	Height of the edge in half ommatidia.	9
dt	float	Time step in seconds.	1 / 200
device	str	Device to store the stimulus.	device
post_pad_mode	Literal['continue', 'value', 'reflect']	Padding mode after the stimulus.	'continue'
t_pre	float	Time before the stimulus in seconds.	1.0
t_post	float	Time after the stimulus in seconds.	1.0
build_stim_on_init	bool	Build the stimulus on initialization.	True
shuffle_offsets	bool	Shuffle the offsets to remove spatio-temporal correlation.	False
seed	int	Seed for the random state.	0
angles	list[int]	List of angles in degrees.	[0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330]

Note

This class uses a very wide bar (width=80) under the hood to render an edge stimulus.

Source code in flyvision/datasets/moving_bar.py

class MovingEdge(MovingBar):
    """Moving edge stimulus.

    This class creates a moving edge stimulus by using a very wide bar.

    Args:
        offsets: First and last offset to the central column in half ommatidia.
        intensities: Intensity of the edge.
        speeds: Speed of the edge in half ommatidia per second.
        height: Height of the edge in half ommatidia.
        dt: Time step in seconds.
        device: Device to store the stimulus.
        post_pad_mode: Padding mode after the stimulus.
        t_pre: Time before the stimulus in seconds.
        t_post: Time after the stimulus in seconds.
        build_stim_on_init: Build the stimulus on initialization.
        shuffle_offsets: Shuffle the offsets to remove spatio-temporal correlation.
        seed: Seed for the random state.
        angles: List of angles in degrees.

    Note:
        This class uses a very wide bar (width=80) under the hood to render an
        edge stimulus.
    """

    def __init__(
        self,
        offsets: tuple[int, int] = (-10, 11),
        intensities: list[float] = [0, 1],
        speeds: list[float] = [2.4, 4.8, 9.7, 13, 19, 25],
        height: int = 9,
        dt: float = 1 / 200,
        device: str = flyvision.device,
        post_pad_mode: Literal["continue", "value", "reflect"] = "continue",
        t_pre: float = 1.0,
        t_post: float = 1.0,
        build_stim_on_init: bool = True,
        shuffle_offsets: bool = False,
        seed: int = 0,
        angles: list[int] = [0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330],
        **kwargs,
    ) -> None:
        super().__init__(
            widths=[80],
            offsets=offsets,
            intensities=intensities,
            speeds=speeds,
            height=height,
            dt=dt,
            device=device,
            bar_loc_horizontal=np.radians(0),
            post_pad_mode=post_pad_mode,
            t_pre=t_pre,
            t_post=t_post,
            build_stim_on_init=build_stim_on_init,
            shuffle_offsets=shuffle_offsets,
            seed=seed,
            angles=angles,
        )

Response Analysis

flyvision.analysis.moving_bar_responses

Analysis of responses to moving edges or bars.

Info

Relies on xarray dataset format defined in flyvision.analysis.stimulus_responses.

peak_responses

peak_responses(
    dataset, norm=None, from_degree=None, to_degree=None
)

Compute peak responses from rectified voltages, optionally normalized.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset containing ‘responses’ and necessary coordinates.	required
norm	DataArray	Normalization array.	None
from_degree	float	Starting degree for masking.	None
to_degree	float	Ending degree for masking.	None

Returns:

Type	Description
DataArray	Peak responses with reshaped and transposed dimensions.

Source code in flyvision/analysis/moving_bar_responses.py

def peak_responses(
    dataset: xr.Dataset,
    norm: xr.DataArray = None,
    from_degree: float = None,
    to_degree: float = None,
) -> xr.DataArray:
    """
    Compute peak responses from rectified voltages, optionally normalized.

    Args:
        dataset: Input dataset containing 'responses' and necessary coordinates.
        norm: Normalization array.
        from_degree: Starting degree for masking.
        to_degree: Ending degree for masking.

    Returns:
        Peak responses with reshaped and transposed dimensions.
    """
    config = dataset.attrs['config']
    from_degree = from_degree if from_degree is not None else config['offsets'][0] * 2.25
    to_degree = to_degree if to_degree is not None else (config['offsets'][1] - 1) * 2.25

    # Generate time masks
    masks = get_time_masks(
        dataset, from_column=from_degree / 5.8, to_column=to_degree / 5.8
    )

    # Apply masks to responses and rectify
    responses = dataset['responses']
    masked = responses.where(masks, other=0)
    rectified = masked.clip(min=0)  # Rectify: max(0, response)

    # Normalize if provided
    if norm is not None:
        rectified = rectified / norm

    # Compute peak (maximum over 'frame')
    peak = rectified.max(dim='frame')
    return peak

get_time_masks

get_time_masks(dataset, from_column=-1.5, to_column=1.5)

Generate time masks for each sample based on speed and column range.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset containing ‘speed’ and ‘time’ coordinates.	required
from_column	float	Start of the column range.	-1.5
to_column	float	End of the column range.	1.5

Returns:

Type	Description
DataArray	Boolean mask with dimensions (‘sample’, ‘frame’).

Source code in flyvision/analysis/moving_bar_responses.py

def get_time_masks(
    dataset: xr.Dataset, from_column: float = -1.5, to_column: float = 1.5
) -> xr.DataArray:
    """
    Generate time masks for each sample based on speed and column range.

    Args:
        dataset: Input dataset containing 'speed' and 'time' coordinates.
        from_column: Start of the column range.
        to_column: End of the column range.

    Returns:
        Boolean mask with dimensions ('sample', 'frame').
    """
    speeds = dataset['speed'].values
    unique_speeds = np.unique(speeds)
    config = dataset.attrs['config']
    start, end = config['offsets']
    times = dataset['time'].values

    # Precompute masks for unique speeds
    mask_dict = {}
    for speed in unique_speeds:
        t_start, t_end = time_window(
            speed, from_column=from_column, to_column=to_column, start=start, end=end
        )
        mask_dict[speed] = mask_between_seconds(t_start, t_end, times)

    # Map masks to each sample based on its speed
    masks = np.array([mask_dict[speed] for speed in speeds])

    # Create a DataArray for the masks
    mask_da = xr.DataArray(
        data=masks,
        dims=('sample', 'frame'),
        coords={'sample': dataset['sample'], 'frame': dataset['frame']},
    )

    return mask_da

peak_responses_angular

peak_responses_angular(
    dataset, norm=None, from_degree=None, to_degree=None
)

Compute peak responses and make them complex over angles.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset.	required
norm	DataArray	Normalization array.	None
from_degree	float	Starting degree for masking.	None
to_degree	float	Ending degree for masking.	None

Returns:

Type	Description
DataArray	Complex-valued peak responses.

Source code in flyvision/analysis/moving_bar_responses.py

def peak_responses_angular(
    dataset: xr.Dataset,
    norm: xr.DataArray = None,
    from_degree: float = None,
    to_degree: float = None,
) -> xr.DataArray:
    """
    Compute peak responses and make them complex over angles.

    Args:
        dataset: Input dataset.
        norm: Normalization array.
        from_degree: Starting degree for masking.
        to_degree: Ending degree for masking.

    Returns:
        Complex-valued peak responses.
    """
    peak = peak_responses(
        dataset, norm=norm, from_degree=from_degree, to_degree=to_degree
    )

    # Make complex over angles
    angles = peak['angle'].values
    radians = np.deg2rad(angles)
    # Expand dimensions to match broadcasting
    radians = radians[np.newaxis, :, np.newaxis]
    complex_peak = peak * np.exp(1j * radians)

    return complex_peak

direction_selectivity_index

direction_selectivity_index(
    dataset,
    average=True,
    norm=None,
    from_degree=None,
    to_degree=None,
)

Compute Direction Selectivity Index (DSI).

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset.	required
average	bool	Whether to average over ‘width’ and ‘speed’.	True
norm	DataArray	Normalization array.	None
from_degree	float	Starting degree for masking.	None
to_degree	float	Ending degree for masking.	None

Returns:

Type	Description
DataArray	Direction Selectivity Index.

Source code in flyvision/analysis/moving_bar_responses.py

def direction_selectivity_index(
    dataset: xr.Dataset,
    average: bool = True,
    norm: xr.DataArray = None,
    from_degree: float = None,
    to_degree: float = None,
) -> xr.DataArray:
    """
    Compute Direction Selectivity Index (DSI).

    Args:
        dataset: Input dataset.
        average: Whether to average over 'width' and 'speed'.
        norm: Normalization array.
        from_degree: Starting degree for masking.
        to_degree: Ending degree for masking.

    Returns:
        Direction Selectivity Index.
    """
    view = peak_responses_angular(
        dataset, norm=norm, from_degree=from_degree, to_degree=to_degree
    )
    view = view.set_index(sample=["angle", "width", "intensity", "speed"]).unstack(
        "sample"
    )

    # Compute vector sum over 'angle'
    vector_sum = view.sum(dim='angle')
    vector_length = np.abs(vector_sum)

    # Normalization: sum of absolute responses
    normalization = np.abs(view).sum(dim='angle').max(dim='intensity')
    dsi = vector_length / (normalization + 1e-15)

    if average:
        # Average over 'width' and 'speed'
        dsi = dsi.mean(dim=['width', 'speed'])

    return dsi.squeeze()

prepare_dsi_data

prepare_dsi_data(
    dsis,
    cell_types,
    sorted_type_list,
    known_on_off_first,
    sort_descending,
)

Prepare DSI data for plotting.

Parameters:

Name	Type	Description	Default
dsis		Array of DSI values.	required
cell_types		Array of cell type labels.	required
sorted_type_list		List of cell types in desired order.	required
known_on_off_first		Whether to sort known ON/OFF types first.	required
sort_descending		Whether to sort DSIs in descending order.	required

Returns:

Type	Description
	Tuple of prepared DSIs and cell types.

Source code in flyvision/analysis/moving_bar_responses.py

def prepare_dsi_data(
    dsis, cell_types, sorted_type_list, known_on_off_first, sort_descending
):
    """
    Prepare DSI data for plotting.

    Args:
        dsis: Array of DSI values.
        cell_types: Array of cell type labels.
        sorted_type_list: List of cell types in desired order.
        known_on_off_first: Whether to sort known ON/OFF types first.
        sort_descending: Whether to sort DSIs in descending order.

    Returns:
        Tuple of prepared DSIs and cell types.
    """
    if known_on_off_first:
        sorted_type_list = nodes_edges_utils.nodes_list_sorting_on_off_unknown(cell_types)

    if sorted_type_list is not None:
        dsis = nodes_edges_utils.sort_by_mapping_lists(
            cell_types, sorted_type_list, dsis, axis=0
        )
        cell_types = np.array(sorted_type_list)

    if sort_descending:
        medians = np.median(dsis, axis=(-2, -1))
        index = np.argsort(medians)[::-1]
        dsis = dsis[index]
        cell_types = cell_types[index]

    return dsis, cell_types

dsi_violins

dsi_violins(
    dsis,
    cell_types,
    scatter_best=False,
    scatter_all=True,
    cmap=None,
    colors=None,
    color="b",
    figsize=[10, 1],
    fontsize=6,
    showmeans=False,
    showmedians=True,
    sorted_type_list=None,
    sort_descending=False,
    known_on_off_first=True,
    scatter_kwargs={},
    **kwargs
)

Create violin plots for Direction Selectivity Index (DSI) data.

Parameters:

Name	Type	Description	Default
dsis		Array of DSI values.	required
cell_types		Array of cell type labels.	required
scatter_best		Whether to scatter the best points.	False
scatter_all		Whether to scatter all points.	True
cmap		Colormap for the violins.	None
colors		Specific colors for the violins.	None
color		Default color if colors is None and cmap is None.	'b'
figsize		Figure size.	[10, 1]
fontsize		Font size for labels.	6
showmeans		Whether to show means on violins.	False
showmedians		Whether to show medians on violins.	True
sorted_type_list		List of cell types in desired order.	None
sort_descending		Whether to sort DSIs in descending order.	False
known_on_off_first		Whether to sort known ON/OFF types first.	True
**kwargs		Additional keyword arguments for violin_groups.	{}

Returns:

Type	Description
	Tuple of (figure, axis, colors, prepared DSIs)

Source code in flyvision/analysis/moving_bar_responses.py

def dsi_violins(
    dsis,
    cell_types,
    scatter_best=False,
    scatter_all=True,
    cmap=None,
    colors=None,
    color="b",
    figsize=[10, 1],
    fontsize=6,
    showmeans=False,
    showmedians=True,
    sorted_type_list=None,
    sort_descending=False,
    known_on_off_first=True,
    scatter_kwargs={},
    **kwargs,
):
    """
    Create violin plots for Direction Selectivity Index (DSI) data.

    Args:
        dsis: Array of DSI values.
        cell_types: Array of cell type labels.
        scatter_best: Whether to scatter the best points.
        scatter_all: Whether to scatter all points.
        cmap: Colormap for the violins.
        colors: Specific colors for the violins.
        color: Default color if colors is None and cmap is None.
        figsize: Figure size.
        fontsize: Font size for labels.
        showmeans: Whether to show means on violins.
        showmedians: Whether to show medians on violins.
        sorted_type_list: List of cell types in desired order.
        sort_descending: Whether to sort DSIs in descending order.
        known_on_off_first: Whether to sort known ON/OFF types first.
        **kwargs: Additional keyword arguments for violin_groups.

    Returns:
        Tuple of (figure, axis, colors, prepared DSIs)
    """
    dsis, cell_types = prepare_dsi_data(
        dsis, cell_types, sorted_type_list, known_on_off_first, sort_descending
    )

    if len(dsis.shape) == 1:
        dsis = dsis[None, None, :]
    elif len(dsis.shape) == 2:
        dsis = dsis[:, None]

    if colors is None and cmap is None and color is not None:
        colors = (color,)

    fig, ax, colors = violin_groups(
        dsis,
        cell_types[:],
        rotation=90,
        scatter=False,
        cmap=cmap,
        colors=colors,
        fontsize=fontsize,
        figsize=figsize,
        width=0.7,
        showmeans=showmeans,
        showmedians=showmedians,
        **kwargs,
    )

    if dsis.shape[1] == 1:
        plt_utils.scatter_on_violins_with_best(
            dsis.T.squeeze(), ax, scatter_best, scatter_all, **scatter_kwargs
        )

    return fig, ax, colors, dsis

dsi_violins_on_and_off

dsi_violins_on_and_off(
    dsis,
    cell_types,
    scatter_best=False,
    scatter_all=True,
    bold_output_type_labels=False,
    output_cell_types=None,
    known_on_off_first=True,
    sorted_type_list=None,
    figsize=[10, 1],
    ylim=(0, 1),
    color_known_types=True,
    fontsize=6,
    fig=None,
    axes=None,
    **kwargs
)

Plot Direction Selectivity Index (DSI) for ON and OFF intensities.

Parameters:

Name	Type	Description	Default
dsis	DataArray	DataArray of DSI values.	required
cell_types	DataArray	DataArray of cell type labels.	required
scatter_best		Whether to scatter the best points.	False
scatter_all		Whether to scatter all points.	True
bold_output_type_labels		Whether to bold output type labels.	False
output_cell_types		Cell types to output.	None
known_on_off_first		Whether to sort known ON/OFF types first.	True
sorted_type_list		List of cell types in desired order.	None
figsize		Figure size.	[10, 1]
ylim		Y-axis limits.	(0, 1)
color_known_types		Whether to color known cell types.	True
fontsize		Font size for labels.	6
fig		Existing figure to use.	None
axes		Existing axes to use.	None
**kwargs		Additional keyword arguments for dsi_violins.	{}

Returns:

Type	Description
	Tuple of (figure, (ax1, ax2))

Source code in flyvision/analysis/moving_bar_responses.py

def dsi_violins_on_and_off(
    dsis: xr.DataArray,
    cell_types: xr.DataArray,
    scatter_best=False,
    scatter_all=True,
    bold_output_type_labels=False,
    output_cell_types=None,
    known_on_off_first=True,
    sorted_type_list=None,
    figsize=[10, 1],
    ylim=(0, 1),
    color_known_types=True,
    fontsize=6,
    fig=None,
    axes=None,
    **kwargs,
):
    """
    Plot Direction Selectivity Index (DSI) for ON and OFF intensities.

    Args:
        dsis: DataArray of DSI values.
        cell_types: DataArray of cell type labels.
        scatter_best: Whether to scatter the best points.
        scatter_all: Whether to scatter all points.
        bold_output_type_labels: Whether to bold output type labels.
        output_cell_types: Cell types to output.
        known_on_off_first: Whether to sort known ON/OFF types first.
        sorted_type_list: List of cell types in desired order.
        figsize: Figure size.
        ylim: Y-axis limits.
        color_known_types: Whether to color known cell types.
        fontsize: Font size for labels.
        fig: Existing figure to use.
        axes: Existing axes to use.
        **kwargs: Additional keyword arguments for dsi_violins.

    Returns:
        Tuple of (figure, (ax1, ax2))
    """
    if len(dsis.shape) == 2:
        dsis = dsis[None, :]

    if fig is None or axes is None:
        fig, axes = plt.subplots(2, 1, figsize=figsize, sharex=True)
        plt_utils.rm_spines(axes[0], spines=("bottom",))

    for i, intensity in enumerate([1, 0]):
        color = ON if intensity == 1 else OFF
        _, ax, *_ = dsi_violins(
            dsis=dsis.sel(intensity=intensity).values.T,
            cell_types=cell_types.values,
            color=color,
            fig=fig,
            ax=axes[i],
            fontsize=fontsize,
            sorted_type_list=sorted_type_list,
            scatter_best=scatter_best,
            scatter_all=scatter_all,
            known_on_off_first=known_on_off_first,
            **kwargs,
        )

        ax.grid(False)
        ax.set_ylim(*ylim)
        plt_utils.trim_axis(ax)
        plt_utils.set_spine_tick_params(
            ax, tickwidth=0.5, ticklength=3, ticklabelpad=2, spinewidth=0.5
        )

        if bold_output_type_labels:
            plt_utils.boldify_labels(output_cell_types, ax)

        if color_known_types:
            plt_utils.color_labels(["T4a", "T4b", "T4c", "T4d"], ON, ax)
            plt_utils.color_labels(["T5a", "T5b", "T5c", "T5d"], OFF, ax)

    # axes[0].set_xticks([])
    axes[0].set_yticks(np.arange(0, 1.2, 0.5))
    axes[1].set_yticks(np.arange(0, 1.2, 0.5))
    axes[1].invert_yaxis()

    return fig, axes

dsi_correlation_to_known

dsi_correlation_to_known(
    dsis, max_aggregate_dims=("intensity")
)

Compute the correlation between predicted DSIs and known DSIs.

Parameters:

Name	Type	Description	Default
dsis	DataArray	DataArray containing DSIs for ON and OFF intensities. Should have dimensions including ‘intensity’ and ‘neuron’, and a coordinate ‘cell_type’.	required
max_aggregate_dims		Dimensions to max-aggregate before computing correlation.	('intensity')

Returns:

Type	Description
DataArray	Correlation between predicted and known DSIs.

Note

Known DSIs are binary (0 or 1) based on whether the cell type is known to be motion-tuned.

Source code in flyvision/analysis/moving_bar_responses.py

def dsi_correlation_to_known(
    dsis: xr.DataArray, max_aggregate_dims=("intensity",)
) -> xr.DataArray:
    """
    Compute the correlation between predicted DSIs and known DSIs.

    Args:
        dsis: DataArray containing DSIs for ON and OFF intensities.
            Should have dimensions including 'intensity' and 'neuron',
            and a coordinate 'cell_type'.
        max_aggregate_dims: Dimensions to max-aggregate before computing correlation.

    Returns:
        Correlation between predicted and known DSIs.

    Note:
        Known DSIs are binary (0 or 1) based on whether the cell type
        is known to be motion-tuned.
    """
    # Ensure the 'intensity' dimension has length 2
    assert dsis.sizes['intensity'] == 2, "Dimension 'intensity' should have length 2"

    # Retrieve ground truth motion tuning information
    motion_tuning = groundtruth_utils.motion_tuning
    known_dsi_types = groundtruth_utils.known_dsi_types

    # Select dsis for known cell types
    dsis_for_known = dsis.where(dsis['cell_type'].isin(known_dsi_types), drop=True).max(
        dim=max_aggregate_dims
    )

    # Construct ground truth motion tuning array
    groundtruth_mt = xr.DataArray(
        [
            1.0 if ct in motion_tuning else 0.0
            for ct in dsis_for_known['cell_type'].values
        ],
        coords={'neuron': dsis_for_known['neuron']},
        dims=['neuron'],
    )

    # Compute correlation along 'neuron' dimension
    corr_dsi = xr.corr(dsis_for_known, groundtruth_mt, dim='neuron')

    return corr_dsi

correlation_to_known_tuning_curves

correlation_to_known_tuning_curves(dataset, absmax=False)

Compute correlation between predicted and known tuning curves.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset.	required
absmax	bool	If True, maximize magnitude of correlation regardless of sign.	False

Returns:

Type	Description
DataArray	Correlation values for each cell type.

Source code in flyvision/analysis/moving_bar_responses.py

def correlation_to_known_tuning_curves(
    dataset: xr.Dataset, absmax: bool = False
) -> xr.DataArray:
    """
    Compute correlation between predicted and known tuning curves.

    Args:
        dataset: Input dataset.
        absmax: If True, maximize magnitude of correlation regardless of sign.

    Returns:
        Correlation values for each cell type.
    """
    tuning = peak_responses(dataset)
    gt_tuning = get_known_tuning_curves(
        ["T4a", "T4b", "T4c", "T4d", "T5a", "T5b", "T5c", "T5d"], np.arange(0, 360, 30)
    )

    tuning = (
        tuning.set_index(sample=["angle", "intensity", "width", "speed"])
        .unstack("sample")
        .fillna(0.0)
        .custom.where(cell_type=["T4a", "T4b", "T4c", "T4d", "T5a", "T5b", "T5c", "T5d"])
    )

    # reset the neuron axis to make it compatible with the ground truth tuning curves
    tuning["neuron"] = np.arange(tuning.coords["neuron"].size)

    correlation = xr.corr(tuning, gt_tuning, dim="angle")
    correlation = correlation.fillna(0.0)
    if absmax:
        # take speed and width that maximize the magnitude of the correlation, regardless
        # of the sign
        argmax = np.abs(correlation).argmax(dim=("speed", "width"))
    else:
        # take speed and width that maximize the correlation as an experimentalist
        # would do
        argmax = correlation.argmax(dim=("speed", "width"))
    correlation = correlation.isel(argmax)
    return correlation

get_known_tuning_curves

get_known_tuning_curves(cell_types, angles)

Retrieve ground truth tuning curves for specified cell types.

Parameters:

Name	Type	Description	Default
cell_types	List[str]	List of cell type names.	required
angles	ndarray	Array of angles to interpolate curves to.	required

Returns:

Type	Description
DataArray	DataArray of interpolated ground truth tuning curves.

Source code in flyvision/analysis/moving_bar_responses.py

def get_known_tuning_curves(cell_types: List[str], angles: np.ndarray) -> xr.DataArray:
    """
    Retrieve ground truth tuning curves for specified cell types.

    Args:
        cell_types: List of cell type names.
        angles: Array of angles to interpolate curves to.

    Returns:
        DataArray of interpolated ground truth tuning curves.
    """
    gt_angles = np.arange(0, 360, 30)
    tuning_curves = []

    for cell_type in cell_types:
        gt_tuning = groundtruth_utils.tuning_curves[cell_type]
        interp_func = interp1d(
            gt_angles, gt_tuning, kind='cubic', fill_value="extrapolate"
        )
        gt_tuning = interp_func(angles)
        tuning_curves.append(gt_tuning)

    dataset = xr.DataArray(
        np.array(tuning_curves),
        dims=['neuron', 'angle'],
        coords={'cell_type': ("neuron", cell_types), 'angle': angles},
    )

    return dataset

preferred_direction

preferred_direction(
    dataset,
    average=True,
    norm=None,
    from_degree=None,
    to_degree=None,
)

Compute the preferred direction based on peak responses.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset.	required
average	bool	Whether to average over certain dimensions.	True
norm	DataArray	Normalization array.	None
from_degree	float	Starting degree for masking.	None
to_degree	float	Ending degree for masking.	None

Returns:

Type	Description
DataArray	Preferred direction angles in radians.

Source code in flyvision/analysis/moving_bar_responses.py

def preferred_direction(
    dataset: xr.Dataset,
    average: bool = True,
    norm: xr.DataArray = None,
    from_degree: float = None,
    to_degree: float = None,
) -> xr.DataArray:
    """
    Compute the preferred direction based on peak responses.

    Args:
        dataset: Input dataset.
        average: Whether to average over certain dimensions.
        norm: Normalization array.
        from_degree: Starting degree for masking.
        to_degree: Ending degree for masking.

    Returns:
        Preferred direction angles in radians.
    """
    view = peak_responses_angular(
        dataset, norm=norm, from_degree=from_degree, to_degree=to_degree
    )
    view = view.set_index(sample=["angle", "width", "intensity", "speed"]).unstack(
        "sample"
    )

    # Compute vector sum over 'angle'
    vector_sum = view.sum(dim='angle')
    theta_pref = np.angle(vector_sum)

    if average:
        # Sum over 'width' and 'speed' before computing angle
        vector_sum = view.sum(dim=['width', 'speed', 'angle'])
        theta_pref = np.angle(vector_sum)

    vector_sum.data = theta_pref
    return vector_sum

angular_distance_to_known

angular_distance_to_known(pds)

Compute angular distance between predicted and known preferred directions for T4/T5.

Parameters:

Name	Type	Description	Default
pds	DataArray	Preferred directions for cells.	required

Returns:

Type	Description
DataArray	Angular distances to known preferred directions.

Source code in flyvision/analysis/moving_bar_responses.py

def angular_distance_to_known(pds: xr.DataArray) -> xr.DataArray:
    """
    Compute angular distance between predicted and known preferred directions for T4/T5.

    Args:
        pds: Preferred directions for cells.

    Returns:
        Angular distances to known preferred directions.
    """
    t4s = pds.custom.where(cell_type=["T4a", "T4b", "T4c", "T4d"], intensity=1)
    t4_distances = angular_distances(t4s, np.array([np.pi, 0, np.pi / 2, 3 * np.pi / 2]))
    t5s = pds.custom.where(cell_type=["T5a", "T5b", "T5c", "T5d"], intensity=0)
    t5_distances = angular_distances(t5s, np.array([np.pi, 0, np.pi / 2, 3 * np.pi / 2]))
    # concatenate both xarrays again in the neuron dimension, drop intensity
    return xr.concat(
        [t4_distances.drop('intensity'), t5_distances.drop('intensity')], dim='neuron'
    )

angular_distances

angular_distances(x, y, upper=np.pi)

Compute angular distances between two sets of angles.

Parameters:

Name	Type	Description	Default
x	DataArray	First set of angles.	required
y	array	Second set of angles.	required
upper	float	Upper bound for distance calculation.	pi

Returns:

Type	Description
DataArray	Angular distances.

Source code in flyvision/analysis/moving_bar_responses.py

def angular_distances(x: xr.DataArray, y: np.array, upper: float = np.pi) -> xr.DataArray:
    """
    Compute angular distances between two sets of angles.

    Args:
        x: First set of angles.
        y: Second set of angles.
        upper: Upper bound for distance calculation.

    Returns:
        Angular distances.
    """
    assert x.neuron.size == len(y)
    y_da = xr.DataArray(y, dims=['neuron'], coords={'neuron': x.coords['neuron']})

    result = xr.apply_ufunc(
        simple_angle_distance,
        x,
        y_da,
        input_core_dims=[['neuron'], ['neuron']],
        output_core_dims=[['neuron']],
        vectorize=True,
        kwargs={'upper': upper},
    )

    return result

simple_angle_distance

simple_angle_distance(a, b, upper=np.pi)

Calculate element-wise angle distance between 0 and pi radians.

Parameters:

Name	Type	Description	Default
a	ndarray	First set of angles in radians.	required
b	ndarray	Second set of angles in radians.	required
upper	float	Upper bound for distance calculation.	pi

Returns:

Type	Description
ndarray	Distance between 0 and pi radians.

Source code in flyvision/analysis/moving_bar_responses.py

def simple_angle_distance(
    a: np.ndarray, b: np.ndarray, upper: float = np.pi
) -> np.ndarray:
    """
    Calculate element-wise angle distance between 0 and pi radians.

    Args:
        a: First set of angles in radians.
        b: Second set of angles in radians.
        upper: Upper bound for distance calculation.

    Returns:
        Distance between 0 and pi radians.
    """
    a = np.atleast_1d(a)
    b = np.atleast_1d(b)

    # make all angles positive between 0 and 2 * pi
    a = a % (2 * np.pi)
    b = b % (2 * np.pi)

    y = np.zeros_like(a)
    # subtract the smaller angle from the larger one
    mask = a >= b
    y[mask] = a[mask] - b[mask]
    y[~mask] = b[~mask] - a[~mask]

    # map distances between pi and 2 pi to 0 and pi
    y[y > np.pi] = 2 * np.pi - y[y > np.pi]

    # map distances between 0 and pi to 0 and upper
    return y / np.pi * upper

plot_angular_tuning

plot_angular_tuning(
    dataset,
    cell_type,
    intensity,
    figsize=(1, 1),
    fontsize=5,
    linewidth=1,
    anglepad=-7,
    xlabelpad=-1,
    groundtruth=True,
    groundtruth_linewidth=1.0,
    fig=None,
    ax=None,
    peak_responses_da=None,
    weighted_average=None,
    average_models=False,
    colors=None,
    zorder=0,
    **kwargs
)

Plot angular tuning for a specific cell type and intensity.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset.	required
cell_type	int	Neuron index to plot.	required
intensity	int	Intensity level (0 or 1).	required
figsize	Tuple[float, float]	Figure size.	(1, 1)
fontsize	int	Font size.	5
linewidth	float	Line width.	1
anglepad	float	Angle padding.	-7
xlabelpad	float	X-label padding.	-1
groundtruth	bool	Whether to plot ground truth.	True
groundtruth_linewidth	float	Line width for ground truth.	1.0
fig	Figure	Existing figure.	None
ax	Axes	Existing axes.	None
peak_responses_da	DataArray	Precomputed peak responses.	None
weighted_average	DataArray	Weights for averaging models.	None
average_models	bool	Whether to average across models.	False
colors	str	Color for the plot.	None
zorder	Union[int, Iterable]	Z-order for plotting.	0
**kwargs		Additional keyword arguments for plotting.	{}

Returns:

Type	Description
Tuple[Figure, Axes]	The figure and axes objects.

Source code in flyvision/analysis/moving_bar_responses.py

def plot_angular_tuning(
    dataset: xr.Dataset,
    cell_type: int,
    intensity: int,
    figsize: Tuple[float, float] = (1, 1),
    fontsize: int = 5,
    linewidth: float = 1,
    anglepad: float = -7,
    xlabelpad: float = -1,
    groundtruth: bool = True,
    groundtruth_linewidth: float = 1.0,
    fig: plt.Figure = None,
    ax: plt.Axes = None,
    peak_responses_da: xr.DataArray = None,
    weighted_average: xr.DataArray = None,
    average_models: bool = False,
    colors: str = None,
    zorder: Union[int, Iterable] = 0,
    **kwargs,
) -> Tuple[plt.Figure, plt.Axes]:
    """
    Plot angular tuning for a specific cell type and intensity.

    Args:
        dataset: Input dataset.
        cell_type: Neuron index to plot.
        intensity: Intensity level (0 or 1).
        figsize: Figure size.
        fontsize: Font size.
        linewidth: Line width.
        anglepad: Angle padding.
        xlabelpad: X-label padding.
        groundtruth: Whether to plot ground truth.
        groundtruth_linewidth: Line width for ground truth.
        fig: Existing figure.
        ax: Existing axes.
        peak_responses_da: Precomputed peak responses.
        weighted_average: Weights for averaging models.
        average_models: Whether to average across models.
        colors: Color for the plot.
        zorder: Z-order for plotting.
        **kwargs: Additional keyword arguments for plotting.

    Returns:
        The figure and axes objects.
    """
    if peak_responses_da is None:
        peak_responses_da = peak_responses(dataset)

    peak_responses_da = peak_responses_da.set_index(
        sample=["angle", "width", "intensity", "speed"]
    ).unstack("sample")

    # Select the specific cell type
    peak = peak_responses_da.custom.where(cell_type=cell_type, intensity=intensity)

    # Squeeze irrelevant dimensions
    # peak = peak.squeeze(dim=['width', 'intensity', 'speed'], drop=True)

    # Average over speeds
    average_tuning = peak.mean(dim=('speed', 'width'))

    # Average over models if specified
    if average_models and weighted_average is not None:
        average_tuning = average_tuning.weighted(weighted_average).mean(dim='network_id')
    elif average_models:
        average_tuning = average_tuning.mean(dim='network_id')

    color = (ON if intensity == 1 else OFF) if colors is None else colors

    average_tuning = average_tuning / average_tuning.max()

    angles = average_tuning['angle'].values
    fig, ax = polar(
        angles,
        average_tuning.data.squeeze().T,
        figsize=figsize,
        fontsize=fontsize,
        linewidth=linewidth,
        anglepad=anglepad,
        xlabelpad=xlabelpad,
        color=color,
        fig=fig,
        ax=ax,
        zorder=zorder,
        **kwargs,
    )

    if groundtruth and cell_type in groundtruth_utils.tuning_curves:
        r_gt = np.array(groundtruth_utils.tuning_curves[cell_type])
        r_gt = r_gt / np.max(np.abs(r_gt))
        theta_gt = np.arange(0, 360, 360 / len(r_gt))
        polar(
            theta_gt,
            r_gt,
            figsize=figsize,
            fontsize=fontsize,
            linewidth=groundtruth_linewidth,
            anglepad=anglepad,
            xlabelpad=xlabelpad,
            color="k",
            fig=fig,
            ax=ax,
            # **kwargs,
        )

    return fig, ax

plot_T4_tuning

plot_T4_tuning(dataset)

Plot tuning curves for T4 cells.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset.	required

Source code in flyvision/analysis/moving_bar_responses.py

def plot_T4_tuning(dataset: xr.Dataset) -> None:
    """
    Plot tuning curves for T4 cells.

    Args:
        dataset: Input dataset.
    """
    fig, axes, _ = plt_utils.get_axis_grid(
        range(4),
        projection="polar",
        aspect_ratio=4,
        figsize=[2.95, 0.83],
        wspace=0.25,
    )
    for i, cell_type in enumerate(["T4a", "T4b", "T4c", "T4d"]):
        plot_angular_tuning(
            dataset,
            cell_type,
            intensity=1,
            fig=fig,
            ax=axes[i],
            groundtruth=True,
            aggregate_models="mean",
            linewidth=1.0,
        )
        axes[i].set_xlabel(cell_type)

    for ax in axes:
        ax.xaxis.label.set_fontsize(5)
        [i.set_linewidth(0.5) for i in ax.spines.values()]
        ax.grid(True, linewidth=0.5)

plot_T5_tuning

plot_T5_tuning(dataset)

Plot tuning curves for T5 cells.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Input dataset.	required

Source code in flyvision/analysis/moving_bar_responses.py

def plot_T5_tuning(dataset: xr.Dataset) -> None:
    """
    Plot tuning curves for T5 cells.

    Args:
        dataset: Input dataset.
    """
    fig, axes, _ = plt_utils.get_axis_grid(
        range(4),
        projection="polar",
        aspect_ratio=4,
        figsize=[2.95, 0.83],
        wspace=0.25,
    )
    for i, cell_type in enumerate(["T5a", "T5b", "T5c", "T5d"]):
        plot_angular_tuning(
            dataset,
            cell_type,
            intensity=0,
            fig=fig,
            ax=axes[i],
            groundtruth=True,
            aggregate_models="mean",
            linewidth=1.0,
        )
        axes[i].set_xlabel(cell_type)

    for ax in axes:
        ax.xaxis.label.set_fontsize(5)
        [i.set_linewidth(0.5) for i in ax.spines.values()]
        ax.grid(True, linewidth=0.5)

mask_between_seconds

mask_between_seconds(
    t_start,
    t_end,
    time=None,
    t_pre=None,
    t_stim=None,
    t_post=None,
    dt=None,
)

Create a boolean mask for time values between t_start and t_end.

Parameters:

Name	Type	Description	Default
t_start	float	Start time for the mask.	required
t_end	float	End time for the mask.	required
time	ndarray	Array of time values. If None, it will be generated using other parameters.	None
t_pre	float	Time before stimulus onset.	None
t_stim	float	Stimulus duration.	None
t_post	float	Time after stimulus offset.	None
dt	float	Time step.	None

Returns:

Type	Description
ndarray	Boolean mask array.

Note

If ‘time’ is not provided, it will be generated using t_pre, t_stim, t_post, and dt.

Source code in flyvision/analysis/moving_bar_responses.py

def mask_between_seconds(
    t_start: float,
    t_end: float,
    time: np.ndarray = None,
    t_pre: float = None,
    t_stim: float = None,
    t_post: float = None,
    dt: float = None,
) -> np.ndarray:
    """
    Create a boolean mask for time values between t_start and t_end.

    Args:
        t_start: Start time for the mask.
        t_end: End time for the mask.
        time: Array of time values. If None, it will be generated using other parameters.
        t_pre: Time before stimulus onset.
        t_stim: Stimulus duration.
        t_post: Time after stimulus offset.
        dt: Time step.

    Returns:
        Boolean mask array.

    Note:
        If 'time' is not provided, it will be generated using t_pre, t_stim, t_post,
        and dt.
    """
    time = time if time is not None else np.arange(-t_pre, t_stim + t_post - dt, dt)
    return (time >= t_start) & (time <= t_end)

time_window

time_window(
    speed,
    from_column=-1.5,
    to_column=1.5,
    start=-10,
    end=11,
)

Calculate start and end time when the bar passes from_column to to_column.

Parameters:

Name	Type	Description	Default
speed	float	Speed in columns/s (5.8deg/s).	required
from_column	float	Starting column in 5.8deg units.	-1.5
to_column	float	Ending column in 5.8deg units.	1.5
start	float	Starting position in LED units (2.25deg).	-10
end	float	Ending position in LED units (2.25deg).	11

Returns:

Type	Description
tuple[float, float]	Tuple containing start and end times.

Note

The function adjusts the to_column by adding a single LED width (2.25 deg) to make it symmetric around the central column.

Source code in flyvision/analysis/moving_bar_responses.py

def time_window(
    speed: float,
    from_column: float = -1.5,
    to_column: float = 1.5,
    start: float = -10,
    end: float = 11,
) -> tuple[float, float]:
    """
    Calculate start and end time when the bar passes from_column to to_column.

    Args:
        speed: Speed in columns/s (5.8deg/s).
        from_column: Starting column in 5.8deg units.
        to_column: Ending column in 5.8deg units.
        start: Starting position in LED units (2.25deg).
        end: Ending position in LED units (2.25deg).

    Returns:
        Tuple containing start and end times.

    Note:
        The function adjusts the to_column by adding a single LED width (2.25 deg)
        to make it symmetric around the central column.
    """
    start_in_columns = start * 2.25 / 5.8  # in 5.8deg
    end_in_columns = end * 2.25 / 5.8  # in 5.8deg

    # Make it symmetric around the central column by adding a single LED width
    to_column += 2.25 / 5.8

    assert abs(start_in_columns) >= abs(from_column)
    assert abs(end_in_columns) >= abs(to_column)

    # Calculate when the edge is at the from_column
    t_start = (abs(start_in_columns) - abs(from_column)) / speed
    # Calculate when it's at the to_column
    t_end = t_start + (to_column - from_column) / speed
    return t_start, t_end

Current Analysis

flyvision.analysis.moving_edge_currents.MovingEdgeCurrentView

Represents a view of moving edge currents for analysis and visualization.

This class provides methods for analyzing and visualizing currents and responses related to moving edge stimuli in neural simulations.

Parameters:

Name	Type	Description	Default
ensemble		The ensemble of models.	required
target_type	str	The type of target cell.	required
exp_data	List[ExperimentData]	Experimental data.	required
arg_df	DataFrame | None	DataFrame containing stimulus arguments.	None
currents	Namespace | None	Currents for each source type.	None
rfs	ReceptiveFields | None	Receptive fields for the target cells.	None
time	ndarray | None	Time array for the simulation.	None
responses	ndarray | None	Responses of the target cells.	None

Attributes:

Name	Type	Description
target_type		The type of target cell.
ensemble		The ensemble of models.
config		Configuration settings.
arg_df		DataFrame containing stimulus arguments.
rfs		Receptive fields for the target cells.
exp_data		Experimental data.
source_types		Types of source cells.
time		Time array for the simulation.
currents		Currents for each source type.
responses		Responses of the target cells.

Note

This class is intended to be updated to use xarray datasets in the future.

Source code in flyvision/analysis/moving_edge_currents.py

class MovingEdgeCurrentView:
    """Represents a view of moving edge currents for analysis and visualization.

    This class provides methods for analyzing and visualizing currents and responses
    related to moving edge stimuli in neural simulations.

    Args:
        ensemble: The ensemble of models.
        target_type: The type of target cell.
        exp_data: Experimental data.
        arg_df: DataFrame containing stimulus arguments.
        currents: Currents for each source type.
        rfs: Receptive fields for the target cells.
        time: Time array for the simulation.
        responses: Responses of the target cells.

    Attributes:
        target_type: The type of target cell.
        ensemble: The ensemble of models.
        config: Configuration settings.
        arg_df: DataFrame containing stimulus arguments.
        rfs: Receptive fields for the target cells.
        exp_data: Experimental data.
        source_types: Types of source cells.
        time: Time array for the simulation.
        currents: Currents for each source type.
        responses: Responses of the target cells.

    Note:
        This class is intended to be updated to use xarray datasets in the future.
    """

    def __init__(
        self,
        ensemble,
        target_type: str,
        exp_data: List[ExperimentData],
        arg_df: pd.DataFrame | None = None,
        currents: Namespace | None = None,
        rfs: ReceptiveFields | None = None,
        time: np.ndarray | None = None,
        responses: np.ndarray | None = None,
    ):
        self.target_type = target_type
        self.ensemble = ensemble
        self.config = exp_data[0].config
        if arg_df is None:
            self.arg_df = MovingEdge(**exp_data[0].config).arg_df
        else:
            self.arg_df = arg_df
        self.rfs = rfs or reset_index(
            ReceptiveFields(target_type, ensemble[0].connectome.edges.to_df())
        )
        self.exp_data = exp_data
        self.source_types = self.rfs.source_types
        self.time = time
        self.init_currents(currents)
        self.init_time(time)
        self.init_responses(responses)

    def init_currents(self, currents: Namespace | None) -> None:
        """Initialize the currents for each source type.

        Args:
            currents: Currents for each source type.
        """
        if currents is not None:
            self.currents = currents
            return
        self.currents = Namespace()
        for source_type in self.rfs.source_types:
            # (on/off, n_models, n_angles, n_timesteps, n_input_cells)
            self.currents[source_type] = np.array(
                [
                    np.array(exp.target_data[self.target_type].source_data[source_type])
                    for exp in self.exp_data
                ],
            )

    def init_responses(self, responses: np.ndarray | None) -> None:
        """Initialize the responses of the target cells.

        Args:
            responses: Responses of the target cells.
        """
        if responses is not None:
            self.responses = responses
            return
        # (on/off, n_models, n_angles, n_timesteps)
        self.responses = np.array(
            [
                np.array(exp.target_data[self.target_type].activity_central)
                for exp in self.exp_data
            ],
        )

    def init_time(self, time: np.ndarray | None) -> None:
        """Initialize the time array for the simulation.

        Args:
            time: Time array for the simulation.
        """
        if time is not None:
            self.time = time
            return
        self.time = self.time or (
            np.arange(0, next(iter(self.currents.values())).shape[-2]) * self.config.dt
            - self.config.t_pre
        )

    @property
    def on(self) -> "MovingEdgeCurrentView":
        """Return a view of the ON responses."""
        on_index = get_stimulus_index(self.arg_df, intensity=0)
        arg_df = self.arg_df.iloc[on_index]
        return self.view(
            Namespace({
                cell_type: np.take(c, indices=on_index, axis=1)
                for cell_type, c in self.currents.items()
            }),
            responses=np.take(self.responses, indices=on_index, axis=1),
            arg_df=arg_df,
        )

    @property
    def off(self) -> "MovingEdgeCurrentView":
        """Return a view of the OFF responses."""
        off_index = get_stimulus_index(self.arg_df, intensity=0)
        arg_df = self.arg_df.iloc[off_index]
        return self.view(
            Namespace({
                cell_type: np.take(c, indices=off_index, axis=1)
                for cell_type, c in self.currents.items()
            }),
            responses=np.take(self.responses, indices=off_index, axis=1),
            arg_df=arg_df,
        )

    def divide_by_given_norm(self, norm: CellTypeArray) -> "MovingEdgeCurrentView":
        """Divide currents and responses by a given norm.

        Args:
            norm: The norm to divide by.

        Returns:
            A new view with normalized currents and responses.

        Raises:
            ValueError: If norm is not a CellTypeArray.
        """
        if not isinstance(norm, CellTypeArray):
            raise ValueError

        response_dims = np.arange(len(self.responses.shape))
        response_norm = np.expand_dims(
            norm[self.target_type].squeeze(), list(set(response_dims) - set([0]))
        )

        # divide the responses by the norm
        new_responses = self.responses[:] / response_norm

        # note: we also divide by the norm of the target cell type

        currents_dims = np.arange(len(next(iter(self.currents.values())).shape))

        currents_norm = np.expand_dims(
            norm[self.target_type].squeeze(), list(set(currents_dims) - set([0]))
        )

        # divide the currents by the norm
        new_currents = Namespace({
            cell_type: c / currents_norm for cell_type, c in self.currents.items()
        })
        return self.view(currents=new_currents, responses=new_responses)

    def at_contrast(self, contrast: float) -> "MovingEdgeCurrentView":
        """Create a new view filtered by contrast.

        Args:
            contrast: The contrast value to filter by.

        Returns:
            A new view with data filtered by the specified contrast.
        """
        contrast_index = get_stimulus_index(self.arg_df, intensity=contrast)
        arg_df = self.arg_df.iloc[contrast_index]
        return self.view(
            Namespace({
                cell_type: np.take(c, indices=contrast_index, axis=1)
                for cell_type, c in self.currents.items()
            }),
            responses=np.take(self.responses, indices=contrast_index, axis=1),
            arg_df=arg_df,
        )

    def at_angle(self, angle: float) -> "MovingEdgeCurrentView":
        """Create a new view filtered by angle.

        Args:
            angle: The angle value to filter by.

        Returns:
            A new view with data filtered by the specified angle.
        """
        angle_index = get_stimulus_index(self.arg_df, angle=angle)
        arg_df = self.arg_df.iloc[angle_index]
        return self.view(
            Namespace({
                cell_type: np.take(c, indices=angle_index, axis=1)
                for cell_type, c in self.currents.items()
            }),
            responses=np.take(self.responses, indices=angle_index, axis=1),
            arg_df=arg_df,
        )

    def at_position(
        self, u: float | None = None, v: float | None = None, central: bool = True
    ) -> "MovingEdgeCurrentView":
        """Create a new view filtered by position.

        Args:
            u: The u-coordinate.
            v: The v-coordinate.
            central: Whether to use central position.

        Returns:
            A new view with data filtered by the specified position.
        """
        rfs = at_position(self.rfs, u, v, central)
        currents = Namespace({
            cell_type: c[:, :, :, :, rfs[cell_type].index]
            for cell_type, c in self.currents.items()
        })
        return self.view(currents, rfs=rfs)

    def between_seconds(self, t_start: float, t_end: float) -> "MovingEdgeCurrentView":
        """Create a new view filtered by time range.

        Args:
            t_start: Start time in seconds.
            t_end: End time in seconds.

        Returns:
            A new view with data filtered by the specified time range.
        """
        slice = np.where((self.time >= t_start) & (self.time < t_end))[0]
        newview = self[:, :, slice, :]
        newview.time = self.time[slice]
        newview.responses = self.responses[:, :, slice]
        return newview

    def model_selection(self, mask: np.ndarray) -> "MovingEdgeCurrentView":
        """Create a new view with selected models.

        Args:
            mask: Boolean mask for model selection.

        Returns:
            A new view with selected models.
        """
        return self[mask, :, :, :]

    def __getattr__(self, key):
        return self.__getitem__(key)

    def __getitem__(self, key) -> Union["MovingEdgeCurrentView", Any]:
        # e.g. view.C3
        if isinstance(key, str) and key in self.source_types:
            return self.view(Namespace({key: self.currents[key]}))
        # e.g. view["C3", 0, 0, 0]
        elif (
            isinstance(key, Iterable)
            and isinstance(key[0], str)
            and key[0] in self.source_types
            and len(key[1:]) == self.shape
        ):
            return self.view(self.currents[key[0]][key[1:]])
        # e.g. view[index, :, :, :]
        elif isinstance(key, Iterable) and len(key) == len(self.shape):
            return self.view(
                Namespace({cell_type: c[key] for cell_type, c in self.currents.items()}),
                responses=self.responses[key[:-1]],
            )
        # view[:]
        elif key == slice(None):
            if len(self.currents) == 1:
                return next(iter(self.currents.values()))
            return self.currents
        return object.__getattribute__(self, key)

    def __repr__(self):
        cv = {ct: v.shape for ct, v in self.currents.items()}
        formatted_cv = ",\n        ".join(
            f"'{ct}': Array(shape={v})" for ct, v in cv.items()
        )
        return (
            f"{self.__class__.__name__}(\n"
            f"    ensemble={self.ensemble.name},\n"
            f"    target_type={self.target_type},\n"
            f"    currents={{\n        {formatted_cv}\n    }},\n"
            f"    rfs={self.rfs}\n"
            f")"
        )

    @property
    def shape(self):
        return next(iter(self.currents.values())).shape

    def sorting(self, average_over_models: bool = True, mode: str = "all") -> np.ndarray:
        """Sort cell types based on their contributions.

        Args:
            average_over_models: Whether to average over models.
            mode: Sorting mode ("all", "excitatory", or "inhibitory").

        Returns:
            Sorted array of cell types.

        Raises:
            ValueError: If an invalid mode is provided.
        """
        summed = self if len(self.shape) == 4 else self.sum_over_cells()
        signs = self.signs()
        if average_over_models:
            absmax = {
                k: v * signs[k]
                for k, v in valmap(
                    lambda v: np.nanmax(
                        np.abs(np.nanmean(v, axis=1, keepdims=True)),
                        axis=(0, 2, 3),
                    ),
                    summed[:],
                ).items()
            }
        else:
            # summing over on/off, angles and time to sort -- results in n_models sortings
            absmax = {
                k: v * signs[k]
                for k, v in valmap(
                    lambda v: np.nanmax(np.abs(v), axis=(0, 2, 3)), summed[:]
                ).items()
            }
        cell_types = np.array(list(absmax.keys()))
        values = np.array(list(absmax.values()))
        sorting = np.argsort(values, axis=0).T
        #         if average_over_models:
        #             # add extra dimension here for the next operation
        #             sorting = sorting[None]
        self.sorted_cell_types = cell_types[sorting[:, ::-1]]

        # return all excitatory and inhibitory from most excitatory to most inhibitory
        if mode == "all":
            return self.sorted_cell_types
        # from most excitatory to least excitatory
        elif mode == "excitatory":
            assert average_over_models
            return np.array([
                cell_type
                for cell_type in self.sorted_cell_types[0]
                if signs[cell_type] == 1
            ])
        # from most inhibitory to least inhibitory
        elif mode == "inhibitory":
            assert average_over_models
            return np.array([
                cell_type
                for cell_type in self.sorted_cell_types[0][::-1]
                if signs[cell_type] == -1
            ])
        else:
            raise ValueError(f"mode {mode}")

    def filter_cell_types_by_contribution(
        self,
        bins: int = 3,
        cut_off_edge: int = 1,
        mode: str = "above_cut_off",
        statistic: Callable = np.max,
    ) -> np.ndarray:
        """Filter cell types based on their contribution.

        Args:
            bins: Number of bins for contribution levels.
            cut_off_edge: Edge index for cut-off.
            mode: Filtering mode ("above_cut_off" or "below_cut_off").
            statistic: Function to compute the statistic.

        Returns:
            Filtered array of cell types.

        Raises:
            ValueError: If an invalid mode is provided.

        Info:
            In principle, chunks the y-axis of the current plots into excitatory and
            inhibitory parts and each of the parts into bins. All cell types with currents
            above or below, depending on the mode, the specified bin edge are discarded.
        """
        sorting = self.sorting()[0]
        signs = self.signs()
        currents = self.sum_over_cells().currents

        filtered_cell_types = []
        for sign in [1, -1]:
            # compute the std over all inputs
            values = {
                cell_type: statistic(np.abs(currents[cell_type][:]))
                for cell_type in sorting
                if signs[cell_type] == sign
            }
            # bin into three bins
            # ala (low contribution, medium contribution, high contribution)
            counts, bins = np.histogram(list(values.values()), bins=bins)
            cut_off_value = bins[cut_off_edge]
            if mode == "above_cut_off":
                filtered_cell_types.extend(
                    list(valfilter(lambda v, cut_off=cut_off_value: v >= cut_off, values))
                )
            elif mode == "below_cut_off":
                filtered_cell_types.extend(
                    list(valfilter(lambda v, cut_off=cut_off_value: v < cut_off, values))
                )
            else:
                raise ValueError(f"mode {mode}")
        return np.array(filtered_cell_types)

    def filter_source_types(
        self,
        hide_source_types: str | list | None,
        bins: int,
        edge: int,
        mode: str,
        statistic: Callable = np.max,
    ) -> np.ndarray:
        """Filter source types based on various criteria.

        Args:
            hide_source_types: Source types to hide or "auto".
            bins: Number of bins for contribution levels.
            edge: Edge index for cut-off.
            mode: Filtering mode.
            statistic: Function to compute the statistic.

        Returns:
            Filtered array of source types.
        """
        source_types = self.sorting()[0]
        if isinstance(hide_source_types, str) and hide_source_types == "auto":
            hide_source_types = self.filter_cell_types_by_contribution(
                bins=bins, cut_off_edge=edge, mode=mode, statistic=statistic
            )

        if hide_source_types is not None:
            source_types = np.array([
                source_type
                for source_type in source_types
                if source_type not in hide_source_types
            ])
        return source_types

    def signs(self) -> dict[str, float]:
        """Compute the signs of receptive fields for each source type.

        Returns:
            Dictionary of signs for each source type.
        """
        return {ct: np.mean(self.rfs[ct].sign) for ct in self.rfs.source_types}

    def sum_over_cells(self) -> "MovingEdgeCurrentView":
        """Sum currents over cells.

        Returns:
            A new view with currents summed over cells.
        """
        return self.view(
            Namespace({
                cell_type: c.sum(axis=-1) for cell_type, c in self.currents.items()
            }),
        )

    def plot_spatial_contribution(
        self,
        source_type: str,
        t_start: float,
        t_end: float,
        mode: str = "peak",
        title: str = "{source_type} :→",
        fig: plt.Figure | None = None,
        ax: plt.Axes | None = None,
        max_extent: float | None = None,
        **kwargs,
    ) -> plt.Axes:
        """Plot the spatial contribution of a source type.

        Args:
            source_type: The source type to plot.
            t_start: Start time for the plot.
            t_end: End time for the plot.
            mode: Mode for calculating values ("peak", "mean", or "std").
            title: Title format string for the plot.
            fig: Existing figure to use.
            ax: Existing axes to use.
            max_extent: Maximum extent of the spatial filter.
            **kwargs: Additional keyword arguments for plt_utils.kernel.

        Returns:
            Axes object containing the plot.
        """
        current_view = kwargs.get("current_view") or (
            self.between_seconds(t_start, t_end)  # .at_contrast(contrast).at_angle(angle)
        )

        vmin = kwargs.get("vmin") or (
            np.floor(
                min(
                    0,
                    min(
                        current.mean(axis=(0, 1, 2)).min()
                        for current in list(current_view[:].values())
                    ),
                )
                * 100
            )
            / 100
        )

        vmax = kwargs.get("vmax") or (
            np.ceil(
                max(
                    0,
                    max(
                        current.mean(axis=(0, 1, 2)).max()
                        for current in list(current_view[:].values())
                    ),
                )
                * 100
            )
            / 100
        )

        u, v = current_view.rfs[source_type][["source_u", "source_v"]].values.T
        # average over models
        # (1, n_models, 1, n_timesteps, n_models) -> (n_timesteps, n_models)
        # import pdb

        # pdb.set_trace()
        values = current_view[source_type][:].mean(axis=(0, 1))
        if mode == "peak":
            values = values[
                np.argmax(np.abs(values), axis=0), np.arange(values.shape[-1])
            ]
        elif mode == "mean":
            values = np.mean(values, axis=0)
        elif mode == "std":
            signs = self.signs()
            values = signs[source_type] * np.std(values, axis=0)
        fig, ax, _ = plots.kernel(
            u,
            v,
            values,
            fill=True,
            max_extent=max_extent or current_view.rfs.max_extent,
            label=title.format(source_type=source_type),
            labelxy="auto",
            strict_sign=False,
            fig=fig,
            ax=ax,
            **kwargs,
        )
        (xmin, ymin, xmax, ymax) = ax.dataLim.extents
        ax.set_xlim(plt_utils.get_lims((xmin, xmax), 0.01))
        ax.set_ylim(plt_utils.get_lims((ymin, ymax), 0.01))

    def plot_spatial_contribution_grid(
        self,
        t_start: float,
        t_end: float,
        max_extent: float = 3,
        mode: str = "peak",
        title: str = "{source_type} :→",
        fig: plt.Figure | None = None,
        axes: np.ndarray[plt.Axes] | None = None,
        fontsize: float = 5,
        edgewidth: float = 0.125,
        title_y: float = 0.8,
        max_figure_height_cm: float = 9.271,
        panel_height_cm: float | str = "auto",
        max_figure_width_cm: float = 2.54,
        panel_width_cm: float = 2.54,
        annotate: bool = False,
        cbar: bool = False,
        hide_source_types: str | list | None = "auto",
        hide_source_types_bins: int = 5,
        hide_source_types_cut_off_edge: int = 1,
        hide_source_types_mode: str = "below_cut_off",
        max_axes: int | None = None,
        **kwargs,
    ) -> tuple[
        plt.Figure,
        np.ndarray[plt.Axes],
        tuple[plt.Colorbar, plt.Colormap, plt.Normalize, float, float],
    ]:
        """Plot a grid of spatial contributions for different source types.

        Args:
            t_start: Start time for the plot.
            t_end: End time for the plot.
            max_extent: Maximum extent of the spatial filter.
            mode: Mode for calculating values ("peak", "mean", or "std").
            title: Title format string for each subplot.
            fig: Existing figure to use.
            axes: Existing axes to use.
            fontsize: Font size for labels and titles.
            edgewidth: Width of edges in the plot.
            title_y: Y-position of the title.
            max_figure_height_cm: Maximum figure height in centimeters.
            panel_height_cm: Height of each panel in centimeters.
            max_figure_width_cm: Maximum figure width in centimeters.
            panel_width_cm: Width of each panel in centimeters.
            annotate: Whether to annotate the plots.
            cbar: Whether to add a colorbar.
            hide_source_types: Source types to hide or "auto".
            hide_source_types_bins: Number of bins for auto-hiding.
            hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
            hide_source_types_mode: Mode for auto-hiding source types.
            max_axes: Maximum number of axes to create.
            **kwargs: Additional keyword arguments for plot_spatial_contribution.

        Returns:
            Figure, axes, and colorbar information (cbar, cmap, norm, vmin, vmax).
        """
        current_view = self.between_seconds(t_start, t_end)

        vmin = (
            np.floor(
                min(
                    0,
                    min(
                        current.mean(axis=(0, 1, 2)).min()
                        for current in list(current_view[:].values())
                    ),
                )
                * 10
            )
            / 10
        )

        vmax = (
            np.ceil(
                max(
                    0,
                    max(
                        current.mean(axis=(0, 1, 2)).max()
                        for current in list(current_view[:].values())
                    ),
                )
                * 10
            )
            / 10
        )

        source_types = self.filter_source_types(
            hide_source_types,
            bins=hide_source_types_bins,
            edge=hide_source_types_cut_off_edge,
            mode=hide_source_types_mode,
        )

        if fig is None and axes is None:
            figsize = figsize_from_n_items(
                max_axes or len(source_types),
                max_figure_height_cm=max_figure_height_cm,
                panel_height_cm=(
                    max_figure_height_cm / (max_axes or len(source_types))
                    if panel_height_cm == "auto"
                    else panel_height_cm
                ),
                max_figure_width_cm=max_figure_width_cm,
                panel_width_cm=panel_width_cm,
            )
            fig, axes = figsize.axis_grid(
                unmask_n=max_axes or len(source_types), hspace=0.1, wspace=0
            )
            if max_axes is not None and len(source_types) < max_axes:
                for ax in np.array(axes).flatten():
                    if isinstance(ax, Axes):
                        ax.axis("off")

        for i, source_type in enumerate(source_types):
            self.plot_spatial_contribution(
                source_type,
                #                 contrast,
                #                 angle,
                t_start,
                t_end,
                mode=mode,
                title=title,
                fontsize=fontsize,
                edgewidth=edgewidth,
                title_y=title_y,
                fig=fig,
                ax=axes[i],
                current_view=current_view,
                vmin=vmin,
                vmax=vmax,
                annotate=annotate,
                cbar=False,
                max_extent=max_extent or current_view.rfs.max_extent,
                **kwargs,
            )

        cmap = plt.cm.seismic
        norm = plt_utils.get_norm(vmin=vmin, vmax=vmax, midpoint=0)
        if cbar:
            cbar = plt_utils.add_colorbar_to_fig(
                fig,
                width=0.01,
                height=0.25,
                fontsize=fontsize,
                cmap=cmap,
                norm=norm,
                label=f"{mode} input currents",
                n_ticks=4,
                n_decimals=1,
            )
        return fig, axes, (cbar, cmap, norm, vmin, vmax)

    def plot_spatial_filter(
        self,
        source_type: str,
        title: str = "{source_type} :→",
        fig: plt.Figure | None = None,
        ax: plt.Axes | None = None,
        max_extent: float | None = None,
        **kwargs,
    ) -> plt.Axes:
        """Plot the spatial filter for a given source type.

        Args:
            source_type: The source type to plot.
            title: Title format string for the plot.
            fig: Existing figure to use.
            ax: Existing axes to use.
            max_extent: Maximum extent of the spatial filter.
            **kwargs: Additional keyword arguments for plt_utils.kernel.

        Returns:
            Axes object containing the plot.
        """
        filter = self.rfs

        def filter_values(rf):
            return (rf.n_syn * rf.sign).values

        vmin = kwargs.pop("vmin", None) or (
            np.floor(
                min(
                    0,
                    min(
                        min(filter_values(filter[source_type]))
                        for source_type in self.source_types
                    ),
                )
                * 100
            )
            / 100
        )

        vmax = kwargs.pop("vmax", None) or (
            np.ceil(
                max(
                    0,
                    max(
                        max(filter_values(filter[source_type]))
                        for source_type in self.source_types
                    ),
                )
                * 100
            )
            / 100
        )

        u, v = filter[source_type][["source_u", "source_v"]].values.T
        # average over models
        # (1, n_models, 1, n_timesteps, n_models) -> (n_timesteps, n_models)
        values = filter_values(filter[source_type])

        label = title.format(source_type=source_type)
        fig, ax, _ = plt_utils.kernel(
            u,
            v,
            values,
            fill=True,
            max_extent=max_extent or filter.max_extent,
            label=label,
            labelxy="auto",
            strict_sign=False,
            fig=fig,
            ax=ax,
            vmin=vmin,
            vmax=vmax,
            **kwargs,
        )
        (xmin, ymin, xmax, ymax) = ax.dataLim.extents
        ax.set_xlim(plt_utils.get_lims((xmin, xmax), 0.01))
        ax.set_ylim(plt_utils.get_lims((ymin, ymax), 0.01))
        return ax

    def plot_spatial_filter_grid(
        self,
        title: str = "{source_type} :→",
        fig: plt.Figure | None = None,
        axes: np.ndarray[plt.Axes] | None = None,
        max_extent: float | None = None,
        fontsize: float = 5,
        edgewidth: float = 0.125,
        title_y: float = 0.8,
        max_figure_height_cm: float = 9.271,
        panel_height_cm: float | str = "auto",
        max_figure_width_cm: float = 2.54,
        panel_width_cm: float = 2.54,
        annotate: bool = False,
        cbar: bool = False,
        hide_source_types: str | list | None = "auto",
        hide_source_types_bins: int = 5,
        hide_source_types_cut_off_edge: int = 1,
        hide_source_types_mode: str = "below_cut_off",
        max_axes: int | None = None,
        wspace: float = 0.0,
        hspace: float = 0.1,
        **kwargs,
    ) -> tuple[
        plt.Figure,
        np.ndarray[plt.Axes],
        tuple[plt.Colorbar, plt.Colormap, plt.Normalize, float, float],
    ]:
        """Plot a grid of spatial filters for different source types.

        Args:
            title: Title format string for each subplot.
            fig: Existing figure to use.
            axes: Existing axes to use.
            max_extent: Maximum extent of the spatial filter.
            fontsize: Font size for labels and titles.
            edgewidth: Width of edges in the plot.
            title_y: Y-position of the title.
            max_figure_height_cm: Maximum figure height in centimeters.
            panel_height_cm: Height of each panel in centimeters.
            max_figure_width_cm: Maximum figure width in centimeters.
            panel_width_cm: Width of each panel in centimeters.
            annotate: Whether to annotate the plots.
            cbar: Whether to add a colorbar.
            hide_source_types: Source types to hide or "auto".
            hide_source_types_bins: Number of bins for auto-hiding.
            hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
            hide_source_types_mode: Mode for auto-hiding source types.
            max_axes: Maximum number of axes to create.
            wspace: Width space between subplots.
            hspace: Height space between subplots.
            **kwargs: Additional keyword arguments for plot_spatial_filter.

        Returns:
            Figure, axes, and colorbar information (cbar, cmap, norm, vmin, vmax).
        """
        filter = self.rfs

        def filter_values(rf):
            return (rf.n_syn * rf.sign).values

        vmin = kwargs.pop("vmin", None) or (
            np.floor(
                min(
                    0,
                    min(
                        min(filter_values(filter[source_type]))
                        for source_type in self.source_types
                    ),
                )
                * 100
            )
            / 100
        )

        vmax = kwargs.pop("vmax", None) or (
            np.ceil(
                max(
                    0,
                    max(
                        max(filter_values(filter[source_type]))
                        for source_type in self.source_types
                    ),
                )
                * 100
            )
            / 100
        )

        source_types = self.filter_source_types(
            hide_source_types,
            bins=hide_source_types_bins,
            edge=hide_source_types_cut_off_edge,
            mode=hide_source_types_mode,
        )

        if fig is None and axes is None:
            figsize = figsize_from_n_items(
                max_axes or len(source_types),
                max_figure_height_cm=max_figure_height_cm,
                panel_height_cm=(
                    max_figure_height_cm / (max_axes or len(source_types))
                    if panel_height_cm == "auto"
                    else panel_height_cm
                ),
                max_figure_width_cm=max_figure_width_cm,
                panel_width_cm=panel_width_cm,
            )
            fig, axes = figsize.axis_grid(
                unmask_n=max_axes or len(source_types), hspace=hspace, wspace=wspace
            )
            if max_axes is not None and len(source_types) < max_axes:
                for ax in np.array(axes).flatten():
                    if isinstance(ax, Axes):
                        ax.axis("off")

        for i, source_type in enumerate(source_types):
            self.plot_spatial_filter(
                source_type,
                title=title,
                fontsize=fontsize,
                edgewidth=edgewidth,
                title_y=title_y,
                fig=fig,
                ax=axes[i],
                vmin=vmin,
                vmax=vmax,
                annotate=annotate,
                cbar=False,
                max_extent=max_extent or filter.max_extent,
                **kwargs,
            )

        cmap = plt.cm.seismic
        norm = plt_utils.get_norm(vmin=vmin, vmax=vmax, midpoint=0)
        if cbar:
            cbar = plt_utils.add_colorbar_to_fig(
                fig,
                width=0.01,
                height=0.25,
                fontsize=fontsize,
                cmap=cmap,
                norm=norm,
                label="spatial filters",
                n_ticks=4,
                n_decimals=1,
            )
        return fig, axes, (cbar, cmap, norm, vmin, vmax)

    def view(
        self,
        currents: Namespace,
        rfs: ReceptiveFields | None = None,
        time: np.ndarray | None = None,
        responses: np.ndarray | None = None,
        arg_df: pd.DataFrame | None = None,
    ) -> "MovingEdgeCurrentView":
        """
        Create a new view with the given currents, rfs, time, responses, and arg_df.

        Args:
            currents: Currents for each source type.
            rfs: Receptive fields for the target cells.
            time: Time array for the simulation.
            responses: Responses of the target cells.
            arg_df: DataFrame containing stimulus arguments.

        Returns:
            A new view with the given data.
        """
        arg_df = arg_df.reset_index(drop=True) if arg_df is not None else self.arg_df
        return MovingEdgeCurrentView(
            self.ensemble,
            self.target_type,
            self.exp_data,
            arg_df,
            currents,
            rfs if rfs is not None else self.rfs,
            time if time is not None else self.time,
            responses if responses is not None else self.responses,
        )

    def subtract_baseline(self) -> "MovingEdgeCurrentView":
        """
        Create a new view with baseline subtracted from the currents and responses.

        Returns:
            A new view with baseline subtracted data.
        """
        return self.view(
            Namespace({
                cell_type: c - np.take(c, [0], -2)
                for cell_type, c in self.currents.items()
            }),
            responses=self.responses - np.take(self.responses, [0], -1),
        )

    def subtract_mean(self) -> "MovingEdgeCurrentView":
        """
        Create a new view with mean subtracted from the currents and responses.

        Returns:
            A new view with mean subtracted data.
        """
        return self.view(
            Namespace({
                cell_type: c - np.mean(c, -2, keepdims=True)
                for cell_type, c in self.currents.items()
            }),
            responses=self.responses - np.mean(self.responses, -1, keepdims=True),
        )

    def standardize(self) -> "MovingEdgeCurrentView":
        """
        Create a new view with standardized currents and responses.

        Returns:
            A new view with standardized data.
        """
        return self.view(
            Namespace({
                cell_type: (c - np.mean(c, -2, keepdims=True))
                / (np.std(c, -2, keepdims=True) + 1e-15)
                for cell_type, c in self.currents.items()
            }),
            responses=(self.responses - np.mean(self.responses, -1, keepdims=True))
            / (np.std(self.responses, -1, keepdims=True) + 1e-15),
        )

    def standardize_over_time_and_pd_nd(
        self, t_start: float, t_end: float, pd: float
    ) -> "MovingEdgeCurrentView":
        """
        Create a new view with standardized currents and responses over time and PD/ND.

        Args:
            t_start: Start time for standardization.
            t_end: End time for standardization.
            pd: Preferred direction for standardization.

        Returns:
            A new view with standardized data.
        """
        temp = self.between_seconds(t_start, t_end).at_angle([pd, (pd - 180) % 360])
        return self.view(
            Namespace({
                cell_type: (
                    c - np.mean(temp.currents[cell_type], (-2, -3), keepdims=True)
                )
                / (np.std(temp.currents[cell_type], (-2, -3), keepdims=True) + 1e-15)
                for cell_type, c in self.currents.items()
            }),
            responses=(self.responses - np.mean(temp.responses, (-1, -2), keepdims=True))
            / (np.std(temp.responses, (-1, -2), keepdims=True) + 1e-15),
        )

    def init_colors(self, source_types: list[str]) -> None:
        """
        Initialize colors for source types.

        Args:
            source_types: List of source types.
        """
        signs = self.signs()
        signs = {cell_type: signs[cell_type] for cell_type in source_types}
        signs_reversed = {cell_type: signs[cell_type] for cell_type in source_types[::-1]}
        n_exc = len([v for v in signs.values() if v == 1])
        n_inh = len([v for v in signs.values() if v == -1])
        exc_colors_pd = cmap_iter(
            truncate_colormap(plt.cm.RdBu, minval=0.05, maxval=0.45, n=n_exc)
        )
        inh_cmap_pd = cmap_iter(
            truncate_colormap(plt.cm.RdBu_r, minval=0.05, maxval=0.45, n=n_inh)
        )
        exc_colors_nd = cmap_iter(
            truncate_colormap(plt.cm.BrBG_r, minval=0.05, maxval=0.45, n=n_exc)
        )
        inh_cmap_nd = cmap_iter(
            truncate_colormap(plt.cm.BrBG, minval=0.05, maxval=0.45, n=n_inh)
        )
        colors_pd = {}
        colors_nd = {}
        for _, (cell_type, sign) in enumerate(signs.items()):
            if sign == 1:
                # take the first half of the RdBu colormap, i.e. red
                colors_pd[cell_type] = next(exc_colors_pd)
                colors_nd[cell_type] = next(exc_colors_nd)

        for _, (cell_type, sign) in enumerate(signs_reversed.items()):
            if sign == -1:
                # take the second half of the RdBu colormap, i.e. blue
                colors_pd[cell_type] = next(inh_cmap_pd)
                colors_nd[cell_type] = next(inh_cmap_nd)
        self.colors_pd = colors_pd
        self.colors_nd = colors_nd

    def color(self, source_type: str, pd: bool = True) -> tuple[float, float, float]:
        """
        Get the color for a given source type.

        Args:
            source_type: The source type.
            pd: Whether to use PD or ND colors.

        Returns:
            The color as an RGB tuple.
        """
        if pd:
            return self.colors_pd[source_type]
        return self.colors_nd[source_type]

    def zorder(
        self,
        source_types: list[str],
        source_type: str,
        start_exc: int = 1000,
        start_inh: int = 1000,
    ) -> int:
        """
        Get the z-order for a given source type.

        Args:
            source_types: List of source types.
            source_type: The source type.
            start_exc: Starting z-order for excitatory cells.
            start_inh: Starting z-order for inhibitory cells.

        Returns:
            The z-order for the given source type.
        """
        signs = self.signs()
        signs_reversed = {cell_type: signs[cell_type] for cell_type in source_types[::-1]}

        z_order = start_exc
        for _, (cell_type, sign) in enumerate(signs.items()):
            if sign == 1:
                if cell_type == source_type:
                    return z_order
                z_order -= 10

        z_order = start_inh
        for _, (cell_type, sign) in enumerate(signs_reversed.items()):
            if sign == -1:
                if cell_type == source_type:
                    return z_order
                z_order -= 10

    def ylims(
        self, source_types: list[str] | None = None, offset: float = 0.02
    ) -> dict[str, tuple[float, float]]:
        """
        Get the y-limits for temporal contributions summed over cells.

        Args:
            source_types: List of source types to consider.
            offset: Offset for the y-limits.

        Returns:
            Y-limits for the given source types or all source types.
        """
        if source_types is not None:
            return {
                cell_type: plt_utils.get_lims(c, offset)
                for cell_type, c in self.sum_over_cells().currents.items()
                if cell_type in source_types
            }
        return plt_utils.get_lims(list(self.sum_over_cells().currents.values()), offset)

    def plot_response(
        self,
        contrast: float,
        angle: float,
        t_start: float = 0,
        t_end: float = 1,
        max_figure_height_cm: float = 1.4477,
        panel_height_cm: float = 1.4477,
        max_figure_width_cm: float = 4.0513,
        panel_width_cm: float = 4.0513,
        fontsize: float = 5,
        model_average: bool = True,
        color: tuple[float, float, float] = (0, 0, 0),
        legend: bool = False,
        hide_yaxis: bool = True,
        trim_axes: bool = True,
        quantile: float | None = None,
        scale_position: str | None = None,  # "lower left",
        scale_label: str = "{:.0f} ms",
        scale_unit: float = 1000,
        hline: bool = False,
        fig: plt.Figure | None = None,
        ax: plt.Axes | None = None,
    ):
        """
        Plot the response to a moving edge stimulus.

        Args:
            contrast: The contrast of the stimulus.
            angle: The angle of the stimulus.
            t_start: Start time for the plot.
            t_end: End time for the plot.
            max_figure_height_cm: Maximum figure height in centimeters.
            panel_height_cm: Height of each panel in centimeters.
            max_figure_width_cm: Maximum figure width in centimeters.
            panel_width_cm: Width of each panel in centimeters.
            fontsize: Font size for labels and titles.
            model_average: Whether to plot the model average.
            color: Color for the plot.
            legend: Whether to show the legend.
            hide_yaxis: Whether to hide the y-axis.
            trim_axes: Whether to trim the axes.
            quantile: Quantile for shading.
            scale_position: Position of the scale.
            scale_label: Label format for the scale.
            scale_unit: Unit for the scale.
            hline: Whether to show a horizontal line at 0.
            fig: Existing figure to use.
            ax: Existing axes to use.

        Returns:
            Figure and axes objects.
        """
        r_pd = (
            self.at_angle(angle)
            .at_contrast(contrast)
            .between_seconds(t_start, t_end)
            .responses.squeeze(axis=-2)
        )
        r_nd = (
            self.at_angle((angle - 180) % 360)
            .at_contrast(contrast)
            .between_seconds(t_start, t_end)
            .responses.squeeze(axis=-2)
        )

        if fig is None and ax is None:
            figsize = figsize_from_n_items(
                1,
                max_figure_height_cm=max_figure_height_cm,
                panel_height_cm=panel_height_cm,
                max_figure_width_cm=max_figure_width_cm,
                panel_width_cm=panel_width_cm,
            )
            fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
            ax = axes[0]

        color = [hex2color(PD), hex2color(ND)] if color is None else [color, color]

        if model_average:
            fig, ax, _, _ = plots.traces(
                [r_pd.mean(axis=0), r_nd.mean(axis=0)],
                x=self.between_seconds(t_start, t_end).time,
                color=color,
                linewidth=1,
                fontsize=fontsize,
                null_line=False,
                fig=fig,
                ax=ax,
                linestyle=["solid", "dashed"],
                legend="" if not legend else [f"{self.target_type}", "null direction"],
                scale_pos=scale_position,
                scale_label=scale_label,
                scale_unit=scale_unit,
            )
        else:
            fig, ax, _, _ = plots.traces(
                r_pd,
                x=self.between_seconds(t_start, t_end).time,
                mean_color=adapt_color_alpha(color[0], 1),
                color=adapt_color_alpha(color[0], 0.5),
                linewidth=0.25,
                zorder_traces=0,
                zorder_mean=10,
                fontsize=fontsize,
                null_line=False,
                highlight_mean=True,
                fig=fig,
                ax=ax,
            )
            plots.traces(
                r_nd,
                x=self.between_seconds(t_start, t_end).time,
                mean_color=adapt_color_alpha(color[1], 1),
                color=adapt_color_alpha(color[1], 0.5),
                linewidth=0.25,
                zorder_traces=0,
                zorder_mean=10,
                fontsize=fontsize,
                null_line=False,
                highlight_mean=True,
                fig=fig,
                linestyle="dashed",
                ax=ax,
            )
        if quantile:
            quantile_pd = np.quantile(r_pd, quantile, axis=0)
            quantile_nd = np.quantile(r_nd, quantile, axis=0)
            ax.fill_between(
                self.between_seconds(t_start, t_end).time,
                quantile_pd[0],
                quantile_pd[1],
                facecolor=adapt_color_alpha(color[0], 0.1),
                edgecolor=adapt_color_alpha(color[0], 1),
                linewidth=0.25,
            )
            ax.fill_between(
                self.between_seconds(t_start, t_end).time,
                quantile_nd[0],
                quantile_nd[1],
                facecolor=adapt_color_alpha(color[1], 0.1),
                edgecolor=adapt_color_alpha(color[1], 1),
                linewidth=0.25,
                linestyle="dashed",
            )

        if hline:
            # horizontal line at 0
            ax.axhline(0, color=(0, 0, 0, 1), linewidth=0.25, zorder=-10)

        if hide_yaxis:
            plt_utils.rm_spines(ax, ("left",))
        if trim_axes:
            plt_utils.trim_axis(ax)
        if legend:
            ax.legend(
                fontsize=fontsize,
                ncols=1,
                bbox_to_anchor=(1.05, 1),
                loc="upper left",
                borderaxespad=0.0,
            )
        return fig, ax

    def plot_response_pc_nc(
        self,
        contrast: float,
        angle: float,
        t_start: float = 0,
        t_end: float = 1,
        max_figure_height_cm: float = 1.4477,
        panel_height_cm: float = 1.4477,
        max_figure_width_cm: float = 4.0513,
        panel_width_cm: float = 4.0513,
        fontsize: float = 5,
        model_average: bool = True,
        color: tuple[float, float, float] = (0, 0, 0),
        legend: bool = False,
        hide_yaxis: bool = True,
        trim_axes: bool = True,
        quantile: float | None = None,
        scale_position: str | None = None,
        scale_label: str = "{:.0f} ms",
        scale_unit: float = 1000,
        fig: plt.Figure | None = None,
        ax: plt.Axes | None = None,
        hline: bool = False,
    ) -> tuple[plt.Figure, plt.Axes]:
        """
        Plot the response to a moving edge stimulus with positive and negative contrasts.

        Args:
            contrast: The contrast of the stimulus.
            angle: The angle of the stimulus.
            t_start: Start time for the plot.
            t_end: End time for the plot.
            max_figure_height_cm: Maximum figure height in centimeters.
            panel_height_cm: Height of each panel in centimeters.
            max_figure_width_cm: Maximum figure width in centimeters.
            panel_width_cm: Width of each panel in centimeters.
            fontsize: Font size for labels and titles.
            model_average: Whether to plot the model average.
            color: Color for the plot.
            legend: Whether to show the legend.
            hide_yaxis: Whether to hide the y-axis.
            trim_axes: Whether to trim the axes.
            quantile: Quantile for shading.
            scale_position: Position of the scale.
            scale_label: Label format for the scale.
            scale_unit: Unit for the scale.
            fig: Existing figure to use.
            ax: Existing axes to use.
            hline: Whether to show a horizontal line at 0.

        Returns:
            Figure and axes objects.
        """
        r_pc = (
            self.at_angle(angle)
            .at_contrast(contrast)
            .between_seconds(t_start, t_end)
            .responses.squeeze(axis=-2)
        )
        r_nc = (
            self.at_angle(angle)
            .at_contrast(0 if contrast == 1 else 1)
            .between_seconds(t_start, t_end)
            .responses.squeeze(axis=-2)
        )

        if fig is None and ax is None:
            figsize = figsize_from_n_items(
                1,
                max_figure_height_cm=max_figure_height_cm,
                panel_height_cm=panel_height_cm,
                max_figure_width_cm=max_figure_width_cm,
                panel_width_cm=panel_width_cm,
            )
            fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
            ax = axes[0]

        color = [hex2color(PD), hex2color(ND)] if color is None else [color, color]

        if model_average:
            fig, ax, _, _ = plots.traces(
                [r_pc.mean(axis=0), r_nc.mean(axis=0)],
                x=self.between_seconds(t_start, t_end).time,
                color=color,
                linewidth=1,
                fontsize=fontsize,
                null_line=False,
                fig=fig,
                ax=ax,
                linestyle=["solid", "dotted"],
                legend="" if not legend else [f"{self.target_type}", "null contrast"],
                scale_pos=scale_position,
                scale_label=scale_label,
                scale_unit=scale_unit,
            )
        else:
            fig, ax, _, _ = plots.traces(
                r_pc,
                x=self.between_seconds(t_start, t_end).time,
                mean_color=adapt_color_alpha(color[0], 1),
                color=adapt_color_alpha(color[0], 0.5),
                linewidth=0.25,
                zorder_traces=0,
                zorder_mean=10,
                fontsize=fontsize,
                null_line=False,
                highlight_mean=True,
                fig=fig,
                ax=ax,
            )
            plots.traces(
                r_nc,
                x=self.between_seconds(t_start, t_end).time,
                mean_color=adapt_color_alpha(color[1], 1),
                color=adapt_color_alpha(color[1], 0.5),
                linewidth=0.25,
                zorder_traces=0,
                zorder_mean=10,
                fontsize=fontsize,
                null_line=False,
                highlight_mean=True,
                fig=fig,
                linestyle="dashed",
                ax=ax,
            )
        if quantile:
            quantile_pd = np.quantile(r_pc, quantile, axis=0)
            quantile_nd = np.quantile(r_nc, quantile, axis=0)
            ax.fill_between(
                self.between_seconds(t_start, t_end).time,
                quantile_pd[0],
                quantile_pd[1],
                facecolor=adapt_color_alpha(color[0], 0.1),
                edgecolor=adapt_color_alpha(color[0], 1),
                linewidth=0.25,
            )
            ax.fill_between(
                self.between_seconds(t_start, t_end).time,
                quantile_nd[0],
                quantile_nd[1],
                facecolor=adapt_color_alpha(color[1], 0.1),
                edgecolor=adapt_color_alpha(color[1], 1),
                linewidth=0.25,
                linestyle="dashed",
            )

        # horizontal line at 0
        if hline:
            ax.axhline(0, color=(0, 0, 0, 1), linewidth=0.25, zorder=-10)

        if hide_yaxis:
            plt_utils.rm_spines(ax, ("left",))
        if trim_axes:
            plt_utils.trim_axis(ax)
        if legend:
            ax.legend(
                fontsize=fontsize,
                ncols=1,
                bbox_to_anchor=(1.05, 1),
                loc="upper left",
                borderaxespad=0.0,
            )
        return fig, ax

    def plot_temporal_contributions(
        self,
        contrast: float,
        angle: float,
        t_start: float = 0,
        t_end: float = 1,
        fontsize: float = 5,
        linewidth: float = 0.25,
        legend: bool = False,
        legend_standalone: bool = True,
        legend_figsize_cm: tuple[float, float] = (4.0572, 1),
        legend_n_rows: int | None = None,
        max_figure_height_cm: float = 3.3941,
        panel_height_cm: float = 3.3941,
        max_figure_width_cm: float = 4.0572,
        panel_width_cm: float = 4.0572,
        model_average: bool = True,
        highlight_mean: bool = True,  # only applies if model_average is False
        sum_exc_inh: bool = False,
        only_sum: bool = False,
        hide_source_types: str | list | None = "auto",
        hide_source_types_bins: int = 5,
        hide_source_types_cut_off_edge: int = 1,
        hide_source_types_mode: str = "below_cut_off",
        hide_yaxis: bool = True,
        trim_axes: bool = True,
        quantile: float | None = None,
        fig: plt.Figure | None = None,
        ax: plt.Axes | None = None,
        legend_ax: plt.Axes | None = None,
        hline: bool = True,
        legend_n_cols: int | None = None,
        baseline_color: tuple[float, float, float, float] | None = None,
        colors: dict[str, tuple[float, float, float, float]] | None = None,
    ):
        """
        Plot temporal contributions of different source types.

        Args:
            contrast: The contrast of the stimulus.
            angle: The angle of the stimulus.
            t_start: Start time for the plot.
            t_end: End time for the plot.
            fontsize: Font size for labels and titles.
            linewidth: Line width for traces.
            legend: Whether to show the legend.
            legend_standalone: Whether to create a standalone legend.
            legend_figsize_cm: Figure size for the standalone legend.
            legend_n_rows: Number of rows for the standalone legend.
            max_figure_height_cm: Maximum figure height in centimeters.
            panel_height_cm: Height of each panel in centimeters.
            max_figure_width_cm: Maximum figure width in centimeters.
            panel_width_cm: Width of each panel in centimeters.
            model_average: Whether to plot the model average.
            highlight_mean: Whether to highlight the mean trace.
            sum_exc_inh: Whether to sum excitatory and inhibitory contributions.
            only_sum: Whether to only plot the summed contributions.
            hide_source_types: Source types to hide or "auto".
            hide_source_types_bins: Number of bins for auto-hiding.
            hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
            hide_source_types_mode: Mode for auto-hiding source types.
            hide_yaxis: Whether to hide the y-axis.
            trim_axes: Whether to trim the axes.
            quantile: Quantile for shading.
            fig: Existing figure to use.
            ax: Existing axes to use.
            legend_ax: Existing axes for the standalone legend.
            hline: Whether to show a horizontal line at 0.
            legend_n_cols: Number of columns for the standalone legend.
            baseline_color: Color for the baseline.
            colors: Colors for each source type.

        Returns:
            Figure, axes, and legend axes objects.

        Example:
            ```
            view = MovingEdgeCurrentView(...)
            fig, ax = view.plot_temporal_contributions(
                contrast=1.0,
                angle=0,
                t_start=0,
                t_end=1,
                fontsize=5,
                linewidth=0.25,
                legend=True
            )
            ```
        """
        if fig is None and ax is None:
            figsize = figsize_from_n_items(
                1,
                max_figure_height_cm=max_figure_height_cm,
                panel_height_cm=panel_height_cm,
                max_figure_width_cm=max_figure_width_cm,
                panel_width_cm=panel_width_cm,
            )
            fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
            ax = axes[0]
        cv_pd = (
            self.at_contrast(contrast)
            .at_angle(angle)
            .between_seconds(t_start, t_end)
            .sum_over_cells()
        )
        cv_nd = (
            self.at_contrast(contrast)
            .at_angle((angle - 180) % 360)
            .between_seconds(t_start, t_end)
            .sum_over_cells()
        )

        source_types = (
            self.at_contrast(contrast)
            .at_angle([angle, (angle - 180) % 360])
            .between_seconds(t_start, t_end)
            .filter_source_types(
                hide_source_types,
                hide_source_types_bins,
                hide_source_types_cut_off_edge,
                hide_source_types_mode,
            )
        )

        color_source_types = (
            self.at_contrast(contrast)
            .at_angle([angle, (angle - 180) % 360])
            .between_seconds(t_start, t_end)
            .filter_source_types(
                None,
                hide_source_types_bins,
                hide_source_types_cut_off_edge,
                hide_source_types_mode,
            )
        )
        cv_pd.init_colors(color_source_types)
        cv_nd.init_colors(color_source_types)

        def plot_mean_trace(
            time, trace, label, color, zorder, linestyle="solid", ax=None, fig=None
        ):
            ax.plot(
                time,
                trace,
                label=label,
                color=color,
                zorder=zorder,
                linestyle=linestyle,
            )

        def plot_individual_traces(
            traces, time, color, zorder, label, linestyle="solid", legend=None
        ):
            if not only_sum and not model_average:
                plots.traces(
                    traces,
                    time,
                    mean_color=color,
                    color=color,
                    linewidth=linewidth,
                    zorder_traces=0,
                    zorder_mean=zorder,
                    fontsize=fontsize,
                    null_line=True,
                    highlight_mean=highlight_mean,
                    fig=fig,
                    ax=ax,
                    legend=legend or label,
                    linestyle=linestyle,
                )

        def plot_quantile(traces, time, color, zorder, linestyle="solid"):
            if quantile:
                Q = np.quantile(traces, quantile, axis=0)
                ax.fill_between(
                    time,
                    Q[0],
                    Q[1],
                    facecolor=adapt_color_alpha(color, 0.1),
                    edgecolor=color,
                    linewidth=0.25,
                    linestyle=linestyle,
                    zorder=zorder - 1,
                )

        def plot_summed_trace(time, trace, label, color, zorder, linestyle="solid"):
            if np.any(trace):
                ax.plot(
                    time,
                    trace,
                    label=label,
                    color=color,
                    zorder=zorder,
                    linestyle=linestyle,
                )

        def get_summed_traces(signs, source_types, cv_pd, cv_nd):
            # sum over cell types then average over models
            exc_pd = np.zeros(cv_pd.shape)
            inh_pd = np.zeros(cv_pd.shape)
            exc_nd = np.zeros(cv_nd.shape)
            inh_nd = np.zeros(cv_nd.shape)
            # sum over cell types
            for source_type in source_types:
                if signs[source_type] == 1:
                    exc_pd += cv_pd[source_type][:]  # (1, n_models, 1, n_timesteps)
                    exc_nd += cv_nd[source_type][:]
                else:
                    inh_pd += cv_pd[source_type][:]
                    inh_nd += cv_nd[source_type][:]
            # (n_models, n_timesteps)
            return (
                exc_pd.squeeze(),
                inh_pd.squeeze(),
                exc_nd.squeeze(),
                inh_nd.squeeze(),
            )

        for source_type in source_types:
            if model_average and not only_sum:
                # mean traces solid for PD and dashed for ND
                if baseline_color is not None:
                    color = baseline_color
                elif colors:
                    color = colors[source_type]
                else:
                    color = cv_pd.color(source_type)

                plot_mean_trace(
                    cv_pd.time,
                    cv_pd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                    source_type,
                    color,
                    cv_pd.zorder(source_types, source_type),
                    ax=ax,
                    fig=fig,
                )
                plot_mean_trace(
                    cv_nd.time,
                    cv_nd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                    source_type,
                    color,
                    linestyle="dashed",
                    zorder=cv_pd.zorder(source_types, source_type),
                    ax=ax,
                    fig=fig,
                )

            elif not model_average and not only_sum:
                # individual traces
                plot_individual_traces(
                    cv_pd[source_type][:].squeeze(axis=-2),
                    cv_pd.time,
                    cv_pd.color(source_type),
                    cv_pd.zorder(source_types, source_type),
                    source_type,
                )
                plot_individual_traces(
                    cv_nd[source_type][:].squeeze(axis=-2),
                    cv_nd.time,
                    cv_pd.color(source_type),
                    cv_pd.zorder(source_types, source_type),
                    source_type,
                    linestyle="dashed",
                    legend="null direction",
                )

            # quantiles
            plot_quantile(
                cv_pd[source_type][:].squeeze(axis=-2),
                cv_pd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                linestyle="solid",
            )
            plot_quantile(
                cv_nd[source_type][:].squeeze(axis=-2),
                cv_nd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                linestyle="dashed",
            )
        if sum_exc_inh or only_sum:
            # plot summed traces
            signs = cv_pd.signs()
            exc_pd, inh_pd, exc_nd, inh_nd = get_summed_traces(
                signs, source_types, cv_pd, cv_nd
            )
            plot_summed_trace(
                cv_pd.time,
                exc_pd.mean(axis=0),
                "excitatory",
                (0.931, 0.0, 0.0, 1.0),
                zorder=2000,
            )
            plot_quantile(
                exc_pd,
                cv_pd.time,
                (0.931, 0.0, 0.0, 1.0),
                zorder=0,
                linestyle="solid",
            )
            plot_summed_trace(
                cv_nd.time,
                exc_nd.mean(axis=0),
                "excitatory",
                (0.931, 0.0, 0.0, 1.0),
                zorder=2000,
                linestyle="dashed",
            )
            plot_quantile(
                exc_nd,
                cv_pd.time,
                (0.931, 0.0, 0.0, 1.0),
                zorder=0,
                linestyle="dashed",
            )
            plot_summed_trace(
                cv_pd.time,
                inh_pd.mean(axis=0),
                "inhibitory",
                (0.0, 0.0, 0.849, 1.0),
                zorder=2000,
            )
            plot_quantile(
                inh_pd,
                cv_pd.time,
                (0.0, 0.0, 0.849, 1.0),
                zorder=0,
                linestyle="solid",
            )
            plot_summed_trace(
                cv_nd.time,
                inh_nd.mean(axis=0),
                "inhibitory",
                (0.0, 0.0, 0.849, 1.0),
                zorder=2000,
                linestyle="dashed",
            )
            plot_quantile(
                inh_nd,
                cv_pd.time,
                (0.0, 0.0, 0.849, 1.0),
                zorder=0,
                linestyle="dashed",
            )

        if hline:
            ax.hlines(
                0,
                cv_pd.time.min(),
                cv_pd.time.max(),
                color=(0, 0, 0, 1),
                linewidth=0.25,
                zorder=-10,
            )

        if legend:
            ax.legend(
                fontsize=fontsize,
                ncols=1,
                bbox_to_anchor=(1.05, 1),
                loc="upper left",
                borderaxespad=0.0,
            )
        else:
            ax.legend().set_visible(False)

        ax.set_xlabel("time (s)", fontsize=fontsize)
        #         ax.set_ylabel("current (a.u.)", fontsize=fontsize)

        if hide_yaxis:
            plt_utils.rm_spines(ax, ("left",))

        if trim_axes:
            plt_utils.trim_axis(ax)

        if legend_standalone:
            handles, labels = ax.get_legend_handles_labels()
            nd_handle = Line2D(
                [0], [0], color="k", lw=1, label="null direction", ls="dashed"
            )
            legend_n_rows = legend_n_rows or len(labels) + 1
            # legend_n_cols = (len(labels) + 1) // legend_n_rows
            legend_fig, legend_ax = plt_utils.standalone_legend(
                [*labels[::2], "null direction"],
                None,
                [*handles[::2], nd_handle],
                fontsize=fontsize,
                n_cols=legend_n_cols,
                handlelength=2,
                columnspacing=0.8,
                labelspacing=0.25,
                figsize=cm_to_inch(legend_figsize_cm),
                fig=fig if legend_ax is not None else None,
                ax=legend_ax,
            )
            return fig, ax, legend_fig, legend_ax
        return fig, ax

    def plot_temporal_contributions_pc_nc(
        self,
        contrast: float,
        angle: float,
        t_start: float = 0,
        t_end: float = 1,
        fontsize: float = 5,
        linewidth: float = 0.25,
        legend: bool = False,
        legend_standalone: bool = True,
        legend_figsize_cm: tuple[float, float] = (4.0572, 1),
        legend_n_rows: int | None = None,
        max_figure_height_cm: float = 3.3941,
        panel_height_cm: float = 3.3941,
        max_figure_width_cm: float = 4.0572,
        panel_width_cm: float = 4.0572,
        model_average: bool = True,
        highlight_mean: bool = True,
        sum_exc_inh: bool = False,
        only_sum: bool = False,
        hide_source_types: str | list | None = "auto",
        hide_source_types_bins: int = 5,
        hide_source_types_cut_off_edge: int = 1,
        hide_source_types_mode: str = "below_cut_off",
        hide_yaxis: bool = True,
        trim_axes: bool = True,
        quantile: float | None = None,
        fig: plt.Figure | None = None,
        ax: plt.Axes | None = None,
        legend_ax: plt.Axes | None = None,
        null_linestyle: str = "dotted",
        legend_n_cols: int | None = None,
    ) -> tuple[plt.Figure, plt.Axes, plt.Figure | None, plt.Axes | None]:
        """
        Temporal contributions of different source types for positive/negative contrasts.

        Args:
            contrast: The contrast of the stimulus.
            angle: The angle of the stimulus.
            t_start: Start time for the plot.
            t_end: End time for the plot.
            fontsize: Font size for labels and titles.
            linewidth: Line width for traces.
            legend: Whether to show the legend.
            legend_standalone: Whether to create a standalone legend.
            legend_figsize_cm: Figure size for the standalone legend.
            legend_n_rows: Number of rows for the standalone legend.
            max_figure_height_cm: Maximum figure height in centimeters.
            panel_height_cm: Height of each panel in centimeters.
            max_figure_width_cm: Maximum figure width in centimeters.
            panel_width_cm: Width of each panel in centimeters.
            model_average: Whether to plot the model average.
            highlight_mean: Whether to highlight the mean trace.
            sum_exc_inh: Whether to sum excitatory and inhibitory contributions.
            only_sum: Whether to only plot the summed contributions.
            hide_source_types: Source types to hide or "auto".
            hide_source_types_bins: Number of bins for auto-hiding.
            hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
            hide_source_types_mode: Mode for auto-hiding source types.
            hide_yaxis: Whether to hide the y-axis.
            trim_axes: Whether to trim the axes.
            quantile: Quantile for shading.
            fig: Existing figure to use.
            ax: Existing axes to use.
            legend_ax: Existing axes for the standalone legend.
            null_linestyle: Linestyle for null direction traces.
            legend_n_cols: Number of columns for the standalone legend.

        Returns:
            Figure, axes, and legend axes objects.

        Example:
            ```
            view = MovingEdgeCurrentView(...)
            fig, ax = view.plot_temporal_contributions_pc_nc(
                contrast=1.0,
                angle=0,
                t_start=0,
                t_end=1,
                fontsize=5,
                linewidth=0.25,
                legend=True
            )
            ```
        """
        if fig is None and ax is None:
            figsize = figsize_from_n_items(
                1,
                max_figure_height_cm=max_figure_height_cm,
                panel_height_cm=panel_height_cm,
                max_figure_width_cm=max_figure_width_cm,
                panel_width_cm=panel_width_cm,
            )
            fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
            ax = axes[0]
        cv_pd = (
            self.at_contrast(contrast)
            .at_angle(angle)
            .between_seconds(t_start, t_end)
            .sum_over_cells()
        )
        cv_nd = (
            self.at_contrast(contrast)
            .at_angle((angle - 180) % 360)
            .between_seconds(t_start, t_end)
            .sum_over_cells()
        )

        source_types = (
            self.at_contrast(contrast)
            .at_angle([angle, (angle - 180) % 360])
            .between_seconds(t_start, t_end)
            .filter_source_types(
                hide_source_types,
                hide_source_types_bins,
                hide_source_types_cut_off_edge,
                hide_source_types_mode,
            )
        )

        color_source_types = (
            self.at_contrast(contrast)
            .at_angle([angle, (angle - 180) % 360])
            .between_seconds(t_start, t_end)
            .filter_source_types(
                None,
                hide_source_types_bins,
                hide_source_types_cut_off_edge,
                hide_source_types_mode,
            )
        )
        cv_pd.init_colors(color_source_types)
        cv_nd.init_colors(color_source_types)

        def plot_mean_trace(
            time, trace, label, color, zorder, linestyle="solid", ax=None, fig=None
        ):
            ax.plot(
                time,
                trace,
                label=label,
                color=color,
                zorder=zorder,
                linestyle=linestyle,
            )

        def plot_individual_traces(
            traces, time, color, zorder, label, linestyle="solid", legend=None
        ):
            if not only_sum and not model_average:
                plots.traces(
                    traces,
                    time,
                    mean_color=color,
                    color=color,
                    linewidth=linewidth,
                    zorder_traces=0,
                    zorder_mean=zorder,
                    fontsize=fontsize,
                    null_line=True,
                    highlight_mean=highlight_mean,
                    fig=fig,
                    ax=ax,
                    legend=legend or label,
                    linestyle=linestyle,
                )

        def plot_quantile(traces, time, color, zorder, linestyle="solid"):
            if quantile:
                Q = np.quantile(traces, quantile, axis=0)
                ax.fill_between(
                    time,
                    Q[0],
                    Q[1],
                    facecolor=adapt_color_alpha(color, 0.1),
                    edgecolor=color,
                    linewidth=0.25,
                    linestyle=linestyle,
                    zorder=zorder - 1,
                )

        def plot_summed_trace(time, trace, label, color, zorder, linestyle="solid"):
            if np.any(trace):
                ax.plot(
                    time,
                    trace,
                    label=label,
                    color=color,
                    zorder=zorder,
                    linestyle=linestyle,
                )

        def get_summed_traces(signs, source_types, cv_pd, cv_nd):
            # sum over cell types then average over models
            exc_pd = np.zeros(cv_pd.shape)
            inh_pd = np.zeros(cv_pd.shape)
            exc_nd = np.zeros(cv_nd.shape)
            inh_nd = np.zeros(cv_nd.shape)
            # sum over cell types
            for source_type in source_types:
                if signs[source_type] == 1:
                    exc_pd += cv_pd[source_type][:]  # (1, n_models, 1, n_timesteps)
                    exc_nd += cv_nd[source_type][:]
                else:
                    inh_pd += cv_pd[source_type][:]
                    inh_nd += cv_nd[source_type][:]
            # (n_models, n_timesteps)
            return (
                exc_pd.squeeze(),
                inh_pd.squeeze(),
                exc_nd.squeeze(),
                inh_nd.squeeze(),
            )

        for source_type in source_types:
            if model_average and not only_sum:
                # mean traces solid for PD and dashed for ND
                color = cv_pd.color(source_type)

                plot_mean_trace(
                    cv_pd.time,
                    cv_pd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                    source_type,
                    color,
                    cv_pd.zorder(source_types, source_type),
                    ax=ax,
                    fig=fig,
                )
                plot_mean_trace(
                    cv_nd.time,
                    cv_nd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                    source_type,
                    color,
                    linestyle=null_linestyle,
                    zorder=cv_pd.zorder(source_types, source_type),
                    ax=ax,
                    fig=fig,
                )

            elif not model_average and not only_sum:
                # individual traces
                plot_individual_traces(
                    cv_pd[source_type][:].squeeze(axis=-2),
                    cv_pd.time,
                    cv_pd.color(source_type),
                    cv_pd.zorder(source_types, source_type),
                    source_type,
                )
                plot_individual_traces(
                    cv_nd[source_type][:].squeeze(axis=-2),
                    cv_nd.time,
                    cv_pd.color(source_type),
                    cv_pd.zorder(source_types, source_type),
                    source_type,
                    linestyle=null_linestyle,
                    legend="null direction",
                )

            # quantiles
            plot_quantile(
                cv_pd[source_type][:].squeeze(axis=-2),
                cv_pd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                linestyle="solid",
            )
            plot_quantile(
                cv_nd[source_type][:].squeeze(axis=-2),
                cv_nd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                linestyle=null_linestyle,
            )
        if sum_exc_inh or only_sum:
            # plot summed traces
            signs = cv_pd.signs()
            exc_pd, inh_pd, exc_nd, inh_nd = get_summed_traces(
                signs, source_types, cv_pd, cv_nd
            )
            plot_summed_trace(
                cv_pd.time,
                exc_pd.mean(axis=0),
                "excitatory",
                (0.931, 0.0, 0.0, 1.0),
                zorder=2000,
            )
            plot_quantile(
                exc_pd,
                cv_pd.time,
                (0.931, 0.0, 0.0, 1.0),
                zorder=0,
                linestyle="solid",
            )
            plot_summed_trace(
                cv_nd.time,
                exc_nd.mean(axis=0),
                "excitatory",
                (0.931, 0.0, 0.0, 1.0),
                zorder=2000,
                linestyle=null_linestyle,
            )
            plot_quantile(
                exc_nd,
                cv_pd.time,
                (0.931, 0.0, 0.0, 1.0),
                zorder=0,
                linestyle=null_linestyle,
            )
            plot_summed_trace(
                cv_pd.time,
                inh_pd.mean(axis=0),
                "inhibitory",
                (0.0, 0.0, 0.849, 1.0),
                zorder=2000,
            )
            plot_quantile(
                inh_pd,
                cv_pd.time,
                (0.0, 0.0, 0.849, 1.0),
                zorder=0,
                linestyle="solid",
            )
            plot_summed_trace(
                cv_nd.time,
                inh_nd.mean(axis=0),
                "inhibitory",
                (0.0, 0.0, 0.849, 1.0),
                zorder=2000,
                linestyle=null_linestyle,
            )
            plot_quantile(
                inh_nd,
                cv_pd.time,
                (0.0, 0.0, 0.849, 1.0),
                zorder=0,
                linestyle=null_linestyle,
            )

        if legend:
            ax.legend(
                fontsize=fontsize,
                ncols=1,
                bbox_to_anchor=(1.05, 1),
                loc="upper left",
                borderaxespad=0.0,
            )
        else:
            ax.legend().set_visible(False)

        ax.set_xlabel("time (s)", fontsize=fontsize)
        #         ax.set_ylabel("current (a.u.)", fontsize=fontsize)

        if hide_yaxis:
            plt_utils.rm_spines(ax, ("left",))

        if trim_axes:
            plt_utils.trim_axis(ax)

        if legend_standalone:
            handles, labels = ax.get_legend_handles_labels()
            nd_handle = Line2D(
                [0], [0], color="k", lw=1, label="null direction", ls=null_linestyle
            )
            legend_n_rows = legend_n_rows or len(labels) + 1
            # legend_n_cols = (len(labels) + 1) // legend_n_rows
            legend_fig, legend_ax = plt_utils.standalone_legend(
                [*labels[::2], "null direction"],
                None,
                [*handles[::2], nd_handle],
                fontsize=fontsize,
                n_cols=legend_n_cols,
                handlelength=2,
                columnspacing=0.8,
                labelspacing=0.25,
                figsize=cm_to_inch(legend_figsize_cm),
                fig=fig if legend_ax is not None else None,
                ax=legend_ax,
            )
            return fig, ax, legend_fig, legend_ax
        return fig, ax, None, None

    def get_temporal_contributions(
        self,
        contrast: float,
        angle: float,
        t_start: float = 0,
        t_end: float = 1,
        hide_source_types: str | list | None = "auto",
        hide_source_types_bins: int = 5,
        hide_source_types_cut_off_edge: int = 1,
        hide_source_types_mode: str = "below_cut_off",
        summed_traces: bool = False,
    ) -> tuple[
        "MovingEdgeCurrentView" | np.ndarray,
        "MovingEdgeCurrentView" | np.ndarray,
        list[str],
        list[str],
    ]:
        cv_pd = (
            self.at_contrast(contrast)
            .at_angle(angle)
            .between_seconds(t_start, t_end)
            .sum_over_cells()
        )
        cv_nd = (
            self.at_contrast(contrast)
            .at_angle((angle - 180) % 360)
            .between_seconds(t_start, t_end)
            .sum_over_cells()
        )

        source_types = (
            self.at_contrast(contrast)
            .at_angle([angle, (angle - 180) % 360])
            .between_seconds(t_start, t_end)
            .filter_source_types(
                hide_source_types,
                hide_source_types_bins,
                hide_source_types_cut_off_edge,
                hide_source_types_mode,
            )
        )

        color_source_types = (
            self.at_contrast(contrast)
            .at_angle([angle, (angle - 180) % 360])
            .between_seconds(t_start, t_end)
            .filter_source_types(
                None,
                hide_source_types_bins,
                hide_source_types_cut_off_edge,
                hide_source_types_mode,
            )
        )
        cv_pd.init_colors(color_source_types)
        cv_nd.init_colors(color_source_types)

        def get_summed_traces(signs, source_types, cv_pd, cv_nd):
            # sum over cell types then average over models
            exc_pd = np.zeros(cv_pd.shape)
            inh_pd = np.zeros(cv_pd.shape)
            exc_nd = np.zeros(cv_nd.shape)
            inh_nd = np.zeros(cv_nd.shape)
            # sum over cell types
            for source_type in source_types:
                if signs[source_type] == 1:
                    exc_pd += cv_pd[source_type][:]  # (1, n_models, 1, n_timesteps)
                    exc_nd += cv_nd[source_type][:]
                else:
                    inh_pd += cv_pd[source_type][:]
                    inh_nd += cv_nd[source_type][:]
            # (n_models, n_timesteps)
            return (
                exc_pd.squeeze(),
                inh_pd.squeeze(),
                exc_nd.squeeze(),
                inh_nd.squeeze(),
            )

        if summed_traces:
            exc_pd, inh_pd, exc_nd, inh_nd = get_summed_traces(
                cv_pd.signs(), source_types, cv_pd, cv_nd
            )
            # return exc_pd, inh_pd, exc_nd, inh_nd, source_types, color_source_types
            return exc_pd, inh_pd, exc_nd, inh_nd

        return cv_pd, cv_nd, source_types, color_source_types

    def get_response(
        self,
        contrast: float,
        angle: float,
        t_start: float = 0,
        t_end: float = 1,
        model_average: bool = True,
    ) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
        r_pd = (
            self.at_angle(angle)
            .at_contrast(contrast)
            .between_seconds(t_start, t_end)
            .responses.squeeze(axis=-2)
        )
        r_nd = (
            self.at_angle((angle - 180) % 360)
            .at_contrast(contrast)
            .between_seconds(t_start, t_end)
            .responses.squeeze(axis=-2)
        )

        if model_average:
            return (
                r_pd.mean(axis=0),
                r_nd.mean(axis=0),
                self.between_seconds(t_start, t_end).time,
            )

        return r_pd, r_nd, self.between_seconds(t_start, t_end).time

on property

on

Return a view of the ON responses.

off property

off

Return a view of the OFF responses.

init_currents

init_currents(currents)

Initialize the currents for each source type.

Parameters:

Name	Type	Description	Default
currents	Namespace | None	Currents for each source type.	required

Source code in flyvision/analysis/moving_edge_currents.py

def init_currents(self, currents: Namespace | None) -> None:
    """Initialize the currents for each source type.

    Args:
        currents: Currents for each source type.
    """
    if currents is not None:
        self.currents = currents
        return
    self.currents = Namespace()
    for source_type in self.rfs.source_types:
        # (on/off, n_models, n_angles, n_timesteps, n_input_cells)
        self.currents[source_type] = np.array(
            [
                np.array(exp.target_data[self.target_type].source_data[source_type])
                for exp in self.exp_data
            ],
        )

init_responses

init_responses(responses)

Initialize the responses of the target cells.

Parameters:

Name	Type	Description	Default
responses	ndarray | None	Responses of the target cells.	required

Source code in flyvision/analysis/moving_edge_currents.py

def init_responses(self, responses: np.ndarray | None) -> None:
    """Initialize the responses of the target cells.

    Args:
        responses: Responses of the target cells.
    """
    if responses is not None:
        self.responses = responses
        return
    # (on/off, n_models, n_angles, n_timesteps)
    self.responses = np.array(
        [
            np.array(exp.target_data[self.target_type].activity_central)
            for exp in self.exp_data
        ],
    )

init_time

init_time(time)

Initialize the time array for the simulation.

Parameters:

Name	Type	Description	Default
time	ndarray | None	Time array for the simulation.	required

Source code in flyvision/analysis/moving_edge_currents.py

def init_time(self, time: np.ndarray | None) -> None:
    """Initialize the time array for the simulation.

    Args:
        time: Time array for the simulation.
    """
    if time is not None:
        self.time = time
        return
    self.time = self.time or (
        np.arange(0, next(iter(self.currents.values())).shape[-2]) * self.config.dt
        - self.config.t_pre
    )

divide_by_given_norm

divide_by_given_norm(norm)

Divide currents and responses by a given norm.

Parameters:

Name	Type	Description	Default
norm	CellTypeArray	The norm to divide by.	required

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with normalized currents and responses.

Raises:

Type	Description
ValueError	If norm is not a CellTypeArray.

Source code in flyvision/analysis/moving_edge_currents.py

def divide_by_given_norm(self, norm: CellTypeArray) -> "MovingEdgeCurrentView":
    """Divide currents and responses by a given norm.

    Args:
        norm: The norm to divide by.

    Returns:
        A new view with normalized currents and responses.

    Raises:
        ValueError: If norm is not a CellTypeArray.
    """
    if not isinstance(norm, CellTypeArray):
        raise ValueError

    response_dims = np.arange(len(self.responses.shape))
    response_norm = np.expand_dims(
        norm[self.target_type].squeeze(), list(set(response_dims) - set([0]))
    )

    # divide the responses by the norm
    new_responses = self.responses[:] / response_norm

    # note: we also divide by the norm of the target cell type

    currents_dims = np.arange(len(next(iter(self.currents.values())).shape))

    currents_norm = np.expand_dims(
        norm[self.target_type].squeeze(), list(set(currents_dims) - set([0]))
    )

    # divide the currents by the norm
    new_currents = Namespace({
        cell_type: c / currents_norm for cell_type, c in self.currents.items()
    })
    return self.view(currents=new_currents, responses=new_responses)

at_contrast

at_contrast(contrast)

Create a new view filtered by contrast.

Parameters:

Name	Type	Description	Default
contrast	float	The contrast value to filter by.	required

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with data filtered by the specified contrast.

Source code in flyvision/analysis/moving_edge_currents.py

def at_contrast(self, contrast: float) -> "MovingEdgeCurrentView":
    """Create a new view filtered by contrast.

    Args:
        contrast: The contrast value to filter by.

    Returns:
        A new view with data filtered by the specified contrast.
    """
    contrast_index = get_stimulus_index(self.arg_df, intensity=contrast)
    arg_df = self.arg_df.iloc[contrast_index]
    return self.view(
        Namespace({
            cell_type: np.take(c, indices=contrast_index, axis=1)
            for cell_type, c in self.currents.items()
        }),
        responses=np.take(self.responses, indices=contrast_index, axis=1),
        arg_df=arg_df,
    )

at_angle

at_angle(angle)

Create a new view filtered by angle.

Parameters:

Name	Type	Description	Default
angle	float	The angle value to filter by.	required

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with data filtered by the specified angle.

Source code in flyvision/analysis/moving_edge_currents.py

def at_angle(self, angle: float) -> "MovingEdgeCurrentView":
    """Create a new view filtered by angle.

    Args:
        angle: The angle value to filter by.

    Returns:
        A new view with data filtered by the specified angle.
    """
    angle_index = get_stimulus_index(self.arg_df, angle=angle)
    arg_df = self.arg_df.iloc[angle_index]
    return self.view(
        Namespace({
            cell_type: np.take(c, indices=angle_index, axis=1)
            for cell_type, c in self.currents.items()
        }),
        responses=np.take(self.responses, indices=angle_index, axis=1),
        arg_df=arg_df,
    )

at_position

at_position(u=None, v=None, central=True)

Create a new view filtered by position.

Parameters:

Name	Type	Description	Default
u	float | None	The u-coordinate.	None
v	float | None	The v-coordinate.	None
central	bool	Whether to use central position.	True

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with data filtered by the specified position.

Source code in flyvision/analysis/moving_edge_currents.py

def at_position(
    self, u: float | None = None, v: float | None = None, central: bool = True
) -> "MovingEdgeCurrentView":
    """Create a new view filtered by position.

    Args:
        u: The u-coordinate.
        v: The v-coordinate.
        central: Whether to use central position.

    Returns:
        A new view with data filtered by the specified position.
    """
    rfs = at_position(self.rfs, u, v, central)
    currents = Namespace({
        cell_type: c[:, :, :, :, rfs[cell_type].index]
        for cell_type, c in self.currents.items()
    })
    return self.view(currents, rfs=rfs)

between_seconds

between_seconds(t_start, t_end)

Create a new view filtered by time range.

Parameters:

Name	Type	Description	Default
t_start	float	Start time in seconds.	required
t_end	float	End time in seconds.	required

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with data filtered by the specified time range.

Source code in flyvision/analysis/moving_edge_currents.py

def between_seconds(self, t_start: float, t_end: float) -> "MovingEdgeCurrentView":
    """Create a new view filtered by time range.

    Args:
        t_start: Start time in seconds.
        t_end: End time in seconds.

    Returns:
        A new view with data filtered by the specified time range.
    """
    slice = np.where((self.time >= t_start) & (self.time < t_end))[0]
    newview = self[:, :, slice, :]
    newview.time = self.time[slice]
    newview.responses = self.responses[:, :, slice]
    return newview

model_selection

model_selection(mask)

Create a new view with selected models.

Parameters:

Name	Type	Description	Default
mask	ndarray	Boolean mask for model selection.	required

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with selected models.

Source code in flyvision/analysis/moving_edge_currents.py

def model_selection(self, mask: np.ndarray) -> "MovingEdgeCurrentView":
    """Create a new view with selected models.

    Args:
        mask: Boolean mask for model selection.

    Returns:
        A new view with selected models.
    """
    return self[mask, :, :, :]

sorting

sorting(average_over_models=True, mode='all')

Sort cell types based on their contributions.

Parameters:

Name	Type	Description	Default
average_over_models	bool	Whether to average over models.	True
mode	str	Sorting mode (“all”, “excitatory”, or “inhibitory”).	'all'

Returns:

Type	Description
ndarray	Sorted array of cell types.

Raises:

Type	Description
ValueError	If an invalid mode is provided.

Source code in flyvision/analysis/moving_edge_currents.py

def sorting(self, average_over_models: bool = True, mode: str = "all") -> np.ndarray:
    """Sort cell types based on their contributions.

    Args:
        average_over_models: Whether to average over models.
        mode: Sorting mode ("all", "excitatory", or "inhibitory").

    Returns:
        Sorted array of cell types.

    Raises:
        ValueError: If an invalid mode is provided.
    """
    summed = self if len(self.shape) == 4 else self.sum_over_cells()
    signs = self.signs()
    if average_over_models:
        absmax = {
            k: v * signs[k]
            for k, v in valmap(
                lambda v: np.nanmax(
                    np.abs(np.nanmean(v, axis=1, keepdims=True)),
                    axis=(0, 2, 3),
                ),
                summed[:],
            ).items()
        }
    else:
        # summing over on/off, angles and time to sort -- results in n_models sortings
        absmax = {
            k: v * signs[k]
            for k, v in valmap(
                lambda v: np.nanmax(np.abs(v), axis=(0, 2, 3)), summed[:]
            ).items()
        }
    cell_types = np.array(list(absmax.keys()))
    values = np.array(list(absmax.values()))
    sorting = np.argsort(values, axis=0).T
    #         if average_over_models:
    #             # add extra dimension here for the next operation
    #             sorting = sorting[None]
    self.sorted_cell_types = cell_types[sorting[:, ::-1]]

    # return all excitatory and inhibitory from most excitatory to most inhibitory
    if mode == "all":
        return self.sorted_cell_types
    # from most excitatory to least excitatory
    elif mode == "excitatory":
        assert average_over_models
        return np.array([
            cell_type
            for cell_type in self.sorted_cell_types[0]
            if signs[cell_type] == 1
        ])
    # from most inhibitory to least inhibitory
    elif mode == "inhibitory":
        assert average_over_models
        return np.array([
            cell_type
            for cell_type in self.sorted_cell_types[0][::-1]
            if signs[cell_type] == -1
        ])
    else:
        raise ValueError(f"mode {mode}")

filter_cell_types_by_contribution

filter_cell_types_by_contribution(
    bins=3,
    cut_off_edge=1,
    mode="above_cut_off",
    statistic=np.max,
)

Filter cell types based on their contribution.

Parameters:

Name	Type	Description	Default
bins	int	Number of bins for contribution levels.	3
cut_off_edge	int	Edge index for cut-off.	1
mode	str	Filtering mode (“above_cut_off” or “below_cut_off”).	'above_cut_off'
statistic	Callable	Function to compute the statistic.	max

Returns:

Type	Description
ndarray	Filtered array of cell types.

Raises:

Type	Description
ValueError	If an invalid mode is provided.

Info

In principle, chunks the y-axis of the current plots into excitatory and inhibitory parts and each of the parts into bins. All cell types with currents above or below, depending on the mode, the specified bin edge are discarded.

Source code in flyvision/analysis/moving_edge_currents.py

def filter_cell_types_by_contribution(
    self,
    bins: int = 3,
    cut_off_edge: int = 1,
    mode: str = "above_cut_off",
    statistic: Callable = np.max,
) -> np.ndarray:
    """Filter cell types based on their contribution.

    Args:
        bins: Number of bins for contribution levels.
        cut_off_edge: Edge index for cut-off.
        mode: Filtering mode ("above_cut_off" or "below_cut_off").
        statistic: Function to compute the statistic.

    Returns:
        Filtered array of cell types.

    Raises:
        ValueError: If an invalid mode is provided.

    Info:
        In principle, chunks the y-axis of the current plots into excitatory and
        inhibitory parts and each of the parts into bins. All cell types with currents
        above or below, depending on the mode, the specified bin edge are discarded.
    """
    sorting = self.sorting()[0]
    signs = self.signs()
    currents = self.sum_over_cells().currents

    filtered_cell_types = []
    for sign in [1, -1]:
        # compute the std over all inputs
        values = {
            cell_type: statistic(np.abs(currents[cell_type][:]))
            for cell_type in sorting
            if signs[cell_type] == sign
        }
        # bin into three bins
        # ala (low contribution, medium contribution, high contribution)
        counts, bins = np.histogram(list(values.values()), bins=bins)
        cut_off_value = bins[cut_off_edge]
        if mode == "above_cut_off":
            filtered_cell_types.extend(
                list(valfilter(lambda v, cut_off=cut_off_value: v >= cut_off, values))
            )
        elif mode == "below_cut_off":
            filtered_cell_types.extend(
                list(valfilter(lambda v, cut_off=cut_off_value: v < cut_off, values))
            )
        else:
            raise ValueError(f"mode {mode}")
    return np.array(filtered_cell_types)

filter_source_types

filter_source_types(
    hide_source_types, bins, edge, mode, statistic=np.max
)

Filter source types based on various criteria.

Parameters:

Name	Type	Description	Default
hide_source_types	str | list | None	Source types to hide or “auto”.	required
bins	int	Number of bins for contribution levels.	required
edge	int	Edge index for cut-off.	required
mode	str	Filtering mode.	required
statistic	Callable	Function to compute the statistic.	max

Returns:

Type	Description
ndarray	Filtered array of source types.

Source code in flyvision/analysis/moving_edge_currents.py

def filter_source_types(
    self,
    hide_source_types: str | list | None,
    bins: int,
    edge: int,
    mode: str,
    statistic: Callable = np.max,
) -> np.ndarray:
    """Filter source types based on various criteria.

    Args:
        hide_source_types: Source types to hide or "auto".
        bins: Number of bins for contribution levels.
        edge: Edge index for cut-off.
        mode: Filtering mode.
        statistic: Function to compute the statistic.

    Returns:
        Filtered array of source types.
    """
    source_types = self.sorting()[0]
    if isinstance(hide_source_types, str) and hide_source_types == "auto":
        hide_source_types = self.filter_cell_types_by_contribution(
            bins=bins, cut_off_edge=edge, mode=mode, statistic=statistic
        )

    if hide_source_types is not None:
        source_types = np.array([
            source_type
            for source_type in source_types
            if source_type not in hide_source_types
        ])
    return source_types

signs

signs()

Compute the signs of receptive fields for each source type.

Returns:

Type	Description
dict[str, float]	Dictionary of signs for each source type.

Source code in flyvision/analysis/moving_edge_currents.py

def signs(self) -> dict[str, float]:
    """Compute the signs of receptive fields for each source type.

    Returns:
        Dictionary of signs for each source type.
    """
    return {ct: np.mean(self.rfs[ct].sign) for ct in self.rfs.source_types}

sum_over_cells

sum_over_cells()

Sum currents over cells.

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with currents summed over cells.

Source code in flyvision/analysis/moving_edge_currents.py

def sum_over_cells(self) -> "MovingEdgeCurrentView":
    """Sum currents over cells.

    Returns:
        A new view with currents summed over cells.
    """
    return self.view(
        Namespace({
            cell_type: c.sum(axis=-1) for cell_type, c in self.currents.items()
        }),
    )

plot_spatial_contribution

plot_spatial_contribution(
    source_type,
    t_start,
    t_end,
    mode="peak",
    title="{source_type} :→",
    fig=None,
    ax=None,
    max_extent=None,
    **kwargs
)

Plot the spatial contribution of a source type.

Parameters:

Name	Type	Description	Default
source_type	str	The source type to plot.	required
t_start	float	Start time for the plot.	required
t_end	float	End time for the plot.	required
mode	str	Mode for calculating values (“peak”, “mean”, or “std”).	'peak'
title	str	Title format string for the plot.	'{source_type} :→'
fig	Figure | None	Existing figure to use.	None
ax	Axes | None	Existing axes to use.	None
max_extent	float | None	Maximum extent of the spatial filter.	None
**kwargs		Additional keyword arguments for plt_utils.kernel.	{}

Returns:

Type	Description
Axes	Axes object containing the plot.

Source code in flyvision/analysis/moving_edge_currents.py

def plot_spatial_contribution(
    self,
    source_type: str,
    t_start: float,
    t_end: float,
    mode: str = "peak",
    title: str = "{source_type} :→",
    fig: plt.Figure | None = None,
    ax: plt.Axes | None = None,
    max_extent: float | None = None,
    **kwargs,
) -> plt.Axes:
    """Plot the spatial contribution of a source type.

    Args:
        source_type: The source type to plot.
        t_start: Start time for the plot.
        t_end: End time for the plot.
        mode: Mode for calculating values ("peak", "mean", or "std").
        title: Title format string for the plot.
        fig: Existing figure to use.
        ax: Existing axes to use.
        max_extent: Maximum extent of the spatial filter.
        **kwargs: Additional keyword arguments for plt_utils.kernel.

    Returns:
        Axes object containing the plot.
    """
    current_view = kwargs.get("current_view") or (
        self.between_seconds(t_start, t_end)  # .at_contrast(contrast).at_angle(angle)
    )

    vmin = kwargs.get("vmin") or (
        np.floor(
            min(
                0,
                min(
                    current.mean(axis=(0, 1, 2)).min()
                    for current in list(current_view[:].values())
                ),
            )
            * 100
        )
        / 100
    )

    vmax = kwargs.get("vmax") or (
        np.ceil(
            max(
                0,
                max(
                    current.mean(axis=(0, 1, 2)).max()
                    for current in list(current_view[:].values())
                ),
            )
            * 100
        )
        / 100
    )

    u, v = current_view.rfs[source_type][["source_u", "source_v"]].values.T
    # average over models
    # (1, n_models, 1, n_timesteps, n_models) -> (n_timesteps, n_models)
    # import pdb

    # pdb.set_trace()
    values = current_view[source_type][:].mean(axis=(0, 1))
    if mode == "peak":
        values = values[
            np.argmax(np.abs(values), axis=0), np.arange(values.shape[-1])
        ]
    elif mode == "mean":
        values = np.mean(values, axis=0)
    elif mode == "std":
        signs = self.signs()
        values = signs[source_type] * np.std(values, axis=0)
    fig, ax, _ = plots.kernel(
        u,
        v,
        values,
        fill=True,
        max_extent=max_extent or current_view.rfs.max_extent,
        label=title.format(source_type=source_type),
        labelxy="auto",
        strict_sign=False,
        fig=fig,
        ax=ax,
        **kwargs,
    )
    (xmin, ymin, xmax, ymax) = ax.dataLim.extents
    ax.set_xlim(plt_utils.get_lims((xmin, xmax), 0.01))
    ax.set_ylim(plt_utils.get_lims((ymin, ymax), 0.01))

plot_spatial_contribution_grid

plot_spatial_contribution_grid(
    t_start,
    t_end,
    max_extent=3,
    mode="peak",
    title="{source_type} :→",
    fig=None,
    axes=None,
    fontsize=5,
    edgewidth=0.125,
    title_y=0.8,
    max_figure_height_cm=9.271,
    panel_height_cm="auto",
    max_figure_width_cm=2.54,
    panel_width_cm=2.54,
    annotate=False,
    cbar=False,
    hide_source_types="auto",
    hide_source_types_bins=5,
    hide_source_types_cut_off_edge=1,
    hide_source_types_mode="below_cut_off",
    max_axes=None,
    **kwargs
)

Plot a grid of spatial contributions for different source types.

Parameters:

Name	Type	Description	Default
t_start	float	Start time for the plot.	required
t_end	float	End time for the plot.	required
max_extent	float	Maximum extent of the spatial filter.	3
mode	str	Mode for calculating values (“peak”, “mean”, or “std”).	'peak'
title	str	Title format string for each subplot.	'{source_type} :→'
fig	Figure | None	Existing figure to use.	None
axes	ndarray[Axes] | None	Existing axes to use.	None
fontsize	float	Font size for labels and titles.	5
edgewidth	float	Width of edges in the plot.	0.125
title_y	float	Y-position of the title.	0.8
max_figure_height_cm	float	Maximum figure height in centimeters.	9.271
panel_height_cm	float | str	Height of each panel in centimeters.	'auto'
max_figure_width_cm	float	Maximum figure width in centimeters.	2.54
panel_width_cm	float	Width of each panel in centimeters.	2.54
annotate	bool	Whether to annotate the plots.	False
cbar	bool	Whether to add a colorbar.	False
hide_source_types	str | list | None	Source types to hide or “auto”.	'auto'
hide_source_types_bins	int	Number of bins for auto-hiding.	5
hide_source_types_cut_off_edge	int	Cut-off edge for auto-hiding.	1
hide_source_types_mode	str	Mode for auto-hiding source types.	'below_cut_off'
max_axes	int | None	Maximum number of axes to create.	None
**kwargs		Additional keyword arguments for plot_spatial_contribution.	{}

Returns:

Type	Description
tuple[Figure, ndarray[Axes], tuple[Colorbar, Colormap, Normalize, float, float]]	Figure, axes, and colorbar information (cbar, cmap, norm, vmin, vmax).

Source code in flyvision/analysis/moving_edge_currents.py

def plot_spatial_contribution_grid(
    self,
    t_start: float,
    t_end: float,
    max_extent: float = 3,
    mode: str = "peak",
    title: str = "{source_type} :→",
    fig: plt.Figure | None = None,
    axes: np.ndarray[plt.Axes] | None = None,
    fontsize: float = 5,
    edgewidth: float = 0.125,
    title_y: float = 0.8,
    max_figure_height_cm: float = 9.271,
    panel_height_cm: float | str = "auto",
    max_figure_width_cm: float = 2.54,
    panel_width_cm: float = 2.54,
    annotate: bool = False,
    cbar: bool = False,
    hide_source_types: str | list | None = "auto",
    hide_source_types_bins: int = 5,
    hide_source_types_cut_off_edge: int = 1,
    hide_source_types_mode: str = "below_cut_off",
    max_axes: int | None = None,
    **kwargs,
) -> tuple[
    plt.Figure,
    np.ndarray[plt.Axes],
    tuple[plt.Colorbar, plt.Colormap, plt.Normalize, float, float],
]:
    """Plot a grid of spatial contributions for different source types.

    Args:
        t_start: Start time for the plot.
        t_end: End time for the plot.
        max_extent: Maximum extent of the spatial filter.
        mode: Mode for calculating values ("peak", "mean", or "std").
        title: Title format string for each subplot.
        fig: Existing figure to use.
        axes: Existing axes to use.
        fontsize: Font size for labels and titles.
        edgewidth: Width of edges in the plot.
        title_y: Y-position of the title.
        max_figure_height_cm: Maximum figure height in centimeters.
        panel_height_cm: Height of each panel in centimeters.
        max_figure_width_cm: Maximum figure width in centimeters.
        panel_width_cm: Width of each panel in centimeters.
        annotate: Whether to annotate the plots.
        cbar: Whether to add a colorbar.
        hide_source_types: Source types to hide or "auto".
        hide_source_types_bins: Number of bins for auto-hiding.
        hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
        hide_source_types_mode: Mode for auto-hiding source types.
        max_axes: Maximum number of axes to create.
        **kwargs: Additional keyword arguments for plot_spatial_contribution.

    Returns:
        Figure, axes, and colorbar information (cbar, cmap, norm, vmin, vmax).
    """
    current_view = self.between_seconds(t_start, t_end)

    vmin = (
        np.floor(
            min(
                0,
                min(
                    current.mean(axis=(0, 1, 2)).min()
                    for current in list(current_view[:].values())
                ),
            )
            * 10
        )
        / 10
    )

    vmax = (
        np.ceil(
            max(
                0,
                max(
                    current.mean(axis=(0, 1, 2)).max()
                    for current in list(current_view[:].values())
                ),
            )
            * 10
        )
        / 10
    )

    source_types = self.filter_source_types(
        hide_source_types,
        bins=hide_source_types_bins,
        edge=hide_source_types_cut_off_edge,
        mode=hide_source_types_mode,
    )

    if fig is None and axes is None:
        figsize = figsize_from_n_items(
            max_axes or len(source_types),
            max_figure_height_cm=max_figure_height_cm,
            panel_height_cm=(
                max_figure_height_cm / (max_axes or len(source_types))
                if panel_height_cm == "auto"
                else panel_height_cm
            ),
            max_figure_width_cm=max_figure_width_cm,
            panel_width_cm=panel_width_cm,
        )
        fig, axes = figsize.axis_grid(
            unmask_n=max_axes or len(source_types), hspace=0.1, wspace=0
        )
        if max_axes is not None and len(source_types) < max_axes:
            for ax in np.array(axes).flatten():
                if isinstance(ax, Axes):
                    ax.axis("off")

    for i, source_type in enumerate(source_types):
        self.plot_spatial_contribution(
            source_type,
            #                 contrast,
            #                 angle,
            t_start,
            t_end,
            mode=mode,
            title=title,
            fontsize=fontsize,
            edgewidth=edgewidth,
            title_y=title_y,
            fig=fig,
            ax=axes[i],
            current_view=current_view,
            vmin=vmin,
            vmax=vmax,
            annotate=annotate,
            cbar=False,
            max_extent=max_extent or current_view.rfs.max_extent,
            **kwargs,
        )

    cmap = plt.cm.seismic
    norm = plt_utils.get_norm(vmin=vmin, vmax=vmax, midpoint=0)
    if cbar:
        cbar = plt_utils.add_colorbar_to_fig(
            fig,
            width=0.01,
            height=0.25,
            fontsize=fontsize,
            cmap=cmap,
            norm=norm,
            label=f"{mode} input currents",
            n_ticks=4,
            n_decimals=1,
        )
    return fig, axes, (cbar, cmap, norm, vmin, vmax)

plot_spatial_filter

plot_spatial_filter(
    source_type,
    title="{source_type} :→",
    fig=None,
    ax=None,
    max_extent=None,
    **kwargs
)

Plot the spatial filter for a given source type.

Parameters:

Name	Type	Description	Default
source_type	str	The source type to plot.	required
title	str	Title format string for the plot.	'{source_type} :→'
fig	Figure | None	Existing figure to use.	None
ax	Axes | None	Existing axes to use.	None
max_extent	float | None	Maximum extent of the spatial filter.	None
**kwargs		Additional keyword arguments for plt_utils.kernel.	{}

Returns:

Type	Description
Axes	Axes object containing the plot.

Source code in flyvision/analysis/moving_edge_currents.py

def plot_spatial_filter(
    self,
    source_type: str,
    title: str = "{source_type} :→",
    fig: plt.Figure | None = None,
    ax: plt.Axes | None = None,
    max_extent: float | None = None,
    **kwargs,
) -> plt.Axes:
    """Plot the spatial filter for a given source type.

    Args:
        source_type: The source type to plot.
        title: Title format string for the plot.
        fig: Existing figure to use.
        ax: Existing axes to use.
        max_extent: Maximum extent of the spatial filter.
        **kwargs: Additional keyword arguments for plt_utils.kernel.

    Returns:
        Axes object containing the plot.
    """
    filter = self.rfs

    def filter_values(rf):
        return (rf.n_syn * rf.sign).values

    vmin = kwargs.pop("vmin", None) or (
        np.floor(
            min(
                0,
                min(
                    min(filter_values(filter[source_type]))
                    for source_type in self.source_types
                ),
            )
            * 100
        )
        / 100
    )

    vmax = kwargs.pop("vmax", None) or (
        np.ceil(
            max(
                0,
                max(
                    max(filter_values(filter[source_type]))
                    for source_type in self.source_types
                ),
            )
            * 100
        )
        / 100
    )

    u, v = filter[source_type][["source_u", "source_v"]].values.T
    # average over models
    # (1, n_models, 1, n_timesteps, n_models) -> (n_timesteps, n_models)
    values = filter_values(filter[source_type])

    label = title.format(source_type=source_type)
    fig, ax, _ = plt_utils.kernel(
        u,
        v,
        values,
        fill=True,
        max_extent=max_extent or filter.max_extent,
        label=label,
        labelxy="auto",
        strict_sign=False,
        fig=fig,
        ax=ax,
        vmin=vmin,
        vmax=vmax,
        **kwargs,
    )
    (xmin, ymin, xmax, ymax) = ax.dataLim.extents
    ax.set_xlim(plt_utils.get_lims((xmin, xmax), 0.01))
    ax.set_ylim(plt_utils.get_lims((ymin, ymax), 0.01))
    return ax

plot_spatial_filter_grid

plot_spatial_filter_grid(
    title="{source_type} :→",
    fig=None,
    axes=None,
    max_extent=None,
    fontsize=5,
    edgewidth=0.125,
    title_y=0.8,
    max_figure_height_cm=9.271,
    panel_height_cm="auto",
    max_figure_width_cm=2.54,
    panel_width_cm=2.54,
    annotate=False,
    cbar=False,
    hide_source_types="auto",
    hide_source_types_bins=5,
    hide_source_types_cut_off_edge=1,
    hide_source_types_mode="below_cut_off",
    max_axes=None,
    wspace=0.0,
    hspace=0.1,
    **kwargs
)

Plot a grid of spatial filters for different source types.

Parameters:

Name	Type	Description	Default
title	str	Title format string for each subplot.	'{source_type} :→'
fig	Figure | None	Existing figure to use.	None
axes	ndarray[Axes] | None	Existing axes to use.	None
max_extent	float | None	Maximum extent of the spatial filter.	None
fontsize	float	Font size for labels and titles.	5
edgewidth	float	Width of edges in the plot.	0.125
title_y	float	Y-position of the title.	0.8
max_figure_height_cm	float	Maximum figure height in centimeters.	9.271
panel_height_cm	float | str	Height of each panel in centimeters.	'auto'
max_figure_width_cm	float	Maximum figure width in centimeters.	2.54
panel_width_cm	float	Width of each panel in centimeters.	2.54
annotate	bool	Whether to annotate the plots.	False
cbar	bool	Whether to add a colorbar.	False
hide_source_types	str | list | None	Source types to hide or “auto”.	'auto'
hide_source_types_bins	int	Number of bins for auto-hiding.	5
hide_source_types_cut_off_edge	int	Cut-off edge for auto-hiding.	1
hide_source_types_mode	str	Mode for auto-hiding source types.	'below_cut_off'
max_axes	int | None	Maximum number of axes to create.	None
wspace	float	Width space between subplots.	0.0
hspace	float	Height space between subplots.	0.1
**kwargs		Additional keyword arguments for plot_spatial_filter.	{}

Returns:

Type	Description
tuple[Figure, ndarray[Axes], tuple[Colorbar, Colormap, Normalize, float, float]]	Figure, axes, and colorbar information (cbar, cmap, norm, vmin, vmax).

Source code in flyvision/analysis/moving_edge_currents.py

def plot_spatial_filter_grid(
    self,
    title: str = "{source_type} :→",
    fig: plt.Figure | None = None,
    axes: np.ndarray[plt.Axes] | None = None,
    max_extent: float | None = None,
    fontsize: float = 5,
    edgewidth: float = 0.125,
    title_y: float = 0.8,
    max_figure_height_cm: float = 9.271,
    panel_height_cm: float | str = "auto",
    max_figure_width_cm: float = 2.54,
    panel_width_cm: float = 2.54,
    annotate: bool = False,
    cbar: bool = False,
    hide_source_types: str | list | None = "auto",
    hide_source_types_bins: int = 5,
    hide_source_types_cut_off_edge: int = 1,
    hide_source_types_mode: str = "below_cut_off",
    max_axes: int | None = None,
    wspace: float = 0.0,
    hspace: float = 0.1,
    **kwargs,
) -> tuple[
    plt.Figure,
    np.ndarray[plt.Axes],
    tuple[plt.Colorbar, plt.Colormap, plt.Normalize, float, float],
]:
    """Plot a grid of spatial filters for different source types.

    Args:
        title: Title format string for each subplot.
        fig: Existing figure to use.
        axes: Existing axes to use.
        max_extent: Maximum extent of the spatial filter.
        fontsize: Font size for labels and titles.
        edgewidth: Width of edges in the plot.
        title_y: Y-position of the title.
        max_figure_height_cm: Maximum figure height in centimeters.
        panel_height_cm: Height of each panel in centimeters.
        max_figure_width_cm: Maximum figure width in centimeters.
        panel_width_cm: Width of each panel in centimeters.
        annotate: Whether to annotate the plots.
        cbar: Whether to add a colorbar.
        hide_source_types: Source types to hide or "auto".
        hide_source_types_bins: Number of bins for auto-hiding.
        hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
        hide_source_types_mode: Mode for auto-hiding source types.
        max_axes: Maximum number of axes to create.
        wspace: Width space between subplots.
        hspace: Height space between subplots.
        **kwargs: Additional keyword arguments for plot_spatial_filter.

    Returns:
        Figure, axes, and colorbar information (cbar, cmap, norm, vmin, vmax).
    """
    filter = self.rfs

    def filter_values(rf):
        return (rf.n_syn * rf.sign).values

    vmin = kwargs.pop("vmin", None) or (
        np.floor(
            min(
                0,
                min(
                    min(filter_values(filter[source_type]))
                    for source_type in self.source_types
                ),
            )
            * 100
        )
        / 100
    )

    vmax = kwargs.pop("vmax", None) or (
        np.ceil(
            max(
                0,
                max(
                    max(filter_values(filter[source_type]))
                    for source_type in self.source_types
                ),
            )
            * 100
        )
        / 100
    )

    source_types = self.filter_source_types(
        hide_source_types,
        bins=hide_source_types_bins,
        edge=hide_source_types_cut_off_edge,
        mode=hide_source_types_mode,
    )

    if fig is None and axes is None:
        figsize = figsize_from_n_items(
            max_axes or len(source_types),
            max_figure_height_cm=max_figure_height_cm,
            panel_height_cm=(
                max_figure_height_cm / (max_axes or len(source_types))
                if panel_height_cm == "auto"
                else panel_height_cm
            ),
            max_figure_width_cm=max_figure_width_cm,
            panel_width_cm=panel_width_cm,
        )
        fig, axes = figsize.axis_grid(
            unmask_n=max_axes or len(source_types), hspace=hspace, wspace=wspace
        )
        if max_axes is not None and len(source_types) < max_axes:
            for ax in np.array(axes).flatten():
                if isinstance(ax, Axes):
                    ax.axis("off")

    for i, source_type in enumerate(source_types):
        self.plot_spatial_filter(
            source_type,
            title=title,
            fontsize=fontsize,
            edgewidth=edgewidth,
            title_y=title_y,
            fig=fig,
            ax=axes[i],
            vmin=vmin,
            vmax=vmax,
            annotate=annotate,
            cbar=False,
            max_extent=max_extent or filter.max_extent,
            **kwargs,
        )

    cmap = plt.cm.seismic
    norm = plt_utils.get_norm(vmin=vmin, vmax=vmax, midpoint=0)
    if cbar:
        cbar = plt_utils.add_colorbar_to_fig(
            fig,
            width=0.01,
            height=0.25,
            fontsize=fontsize,
            cmap=cmap,
            norm=norm,
            label="spatial filters",
            n_ticks=4,
            n_decimals=1,
        )
    return fig, axes, (cbar, cmap, norm, vmin, vmax)

view

view(
    currents,
    rfs=None,
    time=None,
    responses=None,
    arg_df=None,
)

Create a new view with the given currents, rfs, time, responses, and arg_df.

Parameters:

Name	Type	Description	Default
currents	Namespace	Currents for each source type.	required
rfs	ReceptiveFields | None	Receptive fields for the target cells.	None
time	ndarray | None	Time array for the simulation.	None
responses	ndarray | None	Responses of the target cells.	None
arg_df	DataFrame | None	DataFrame containing stimulus arguments.	None

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with the given data.

Source code in flyvision/analysis/moving_edge_currents.py

def view(
    self,
    currents: Namespace,
    rfs: ReceptiveFields | None = None,
    time: np.ndarray | None = None,
    responses: np.ndarray | None = None,
    arg_df: pd.DataFrame | None = None,
) -> "MovingEdgeCurrentView":
    """
    Create a new view with the given currents, rfs, time, responses, and arg_df.

    Args:
        currents: Currents for each source type.
        rfs: Receptive fields for the target cells.
        time: Time array for the simulation.
        responses: Responses of the target cells.
        arg_df: DataFrame containing stimulus arguments.

    Returns:
        A new view with the given data.
    """
    arg_df = arg_df.reset_index(drop=True) if arg_df is not None else self.arg_df
    return MovingEdgeCurrentView(
        self.ensemble,
        self.target_type,
        self.exp_data,
        arg_df,
        currents,
        rfs if rfs is not None else self.rfs,
        time if time is not None else self.time,
        responses if responses is not None else self.responses,
    )

subtract_baseline

subtract_baseline()

Create a new view with baseline subtracted from the currents and responses.

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with baseline subtracted data.

Source code in flyvision/analysis/moving_edge_currents.py

def subtract_baseline(self) -> "MovingEdgeCurrentView":
    """
    Create a new view with baseline subtracted from the currents and responses.

    Returns:
        A new view with baseline subtracted data.
    """
    return self.view(
        Namespace({
            cell_type: c - np.take(c, [0], -2)
            for cell_type, c in self.currents.items()
        }),
        responses=self.responses - np.take(self.responses, [0], -1),
    )

subtract_mean

subtract_mean()

Create a new view with mean subtracted from the currents and responses.

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with mean subtracted data.

Source code in flyvision/analysis/moving_edge_currents.py

def subtract_mean(self) -> "MovingEdgeCurrentView":
    """
    Create a new view with mean subtracted from the currents and responses.

    Returns:
        A new view with mean subtracted data.
    """
    return self.view(
        Namespace({
            cell_type: c - np.mean(c, -2, keepdims=True)
            for cell_type, c in self.currents.items()
        }),
        responses=self.responses - np.mean(self.responses, -1, keepdims=True),
    )

standardize

standardize()

Create a new view with standardized currents and responses.

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with standardized data.

Source code in flyvision/analysis/moving_edge_currents.py

def standardize(self) -> "MovingEdgeCurrentView":
    """
    Create a new view with standardized currents and responses.

    Returns:
        A new view with standardized data.
    """
    return self.view(
        Namespace({
            cell_type: (c - np.mean(c, -2, keepdims=True))
            / (np.std(c, -2, keepdims=True) + 1e-15)
            for cell_type, c in self.currents.items()
        }),
        responses=(self.responses - np.mean(self.responses, -1, keepdims=True))
        / (np.std(self.responses, -1, keepdims=True) + 1e-15),
    )

standardize_over_time_and_pd_nd

standardize_over_time_and_pd_nd(t_start, t_end, pd)

Create a new view with standardized currents and responses over time and PD/ND.

Parameters:

Name	Type	Description	Default
t_start	float	Start time for standardization.	required
t_end	float	End time for standardization.	required
pd	float	Preferred direction for standardization.	required

Returns:

Type	Description
'MovingEdgeCurrentView'	A new view with standardized data.

Source code in flyvision/analysis/moving_edge_currents.py

def standardize_over_time_and_pd_nd(
    self, t_start: float, t_end: float, pd: float
) -> "MovingEdgeCurrentView":
    """
    Create a new view with standardized currents and responses over time and PD/ND.

    Args:
        t_start: Start time for standardization.
        t_end: End time for standardization.
        pd: Preferred direction for standardization.

    Returns:
        A new view with standardized data.
    """
    temp = self.between_seconds(t_start, t_end).at_angle([pd, (pd - 180) % 360])
    return self.view(
        Namespace({
            cell_type: (
                c - np.mean(temp.currents[cell_type], (-2, -3), keepdims=True)
            )
            / (np.std(temp.currents[cell_type], (-2, -3), keepdims=True) + 1e-15)
            for cell_type, c in self.currents.items()
        }),
        responses=(self.responses - np.mean(temp.responses, (-1, -2), keepdims=True))
        / (np.std(temp.responses, (-1, -2), keepdims=True) + 1e-15),
    )

init_colors

init_colors(source_types)

Initialize colors for source types.

Parameters:

Name	Type	Description	Default
source_types	list[str]	List of source types.	required

Source code in flyvision/analysis/moving_edge_currents.py

def init_colors(self, source_types: list[str]) -> None:
    """
    Initialize colors for source types.

    Args:
        source_types: List of source types.
    """
    signs = self.signs()
    signs = {cell_type: signs[cell_type] for cell_type in source_types}
    signs_reversed = {cell_type: signs[cell_type] for cell_type in source_types[::-1]}
    n_exc = len([v for v in signs.values() if v == 1])
    n_inh = len([v for v in signs.values() if v == -1])
    exc_colors_pd = cmap_iter(
        truncate_colormap(plt.cm.RdBu, minval=0.05, maxval=0.45, n=n_exc)
    )
    inh_cmap_pd = cmap_iter(
        truncate_colormap(plt.cm.RdBu_r, minval=0.05, maxval=0.45, n=n_inh)
    )
    exc_colors_nd = cmap_iter(
        truncate_colormap(plt.cm.BrBG_r, minval=0.05, maxval=0.45, n=n_exc)
    )
    inh_cmap_nd = cmap_iter(
        truncate_colormap(plt.cm.BrBG, minval=0.05, maxval=0.45, n=n_inh)
    )
    colors_pd = {}
    colors_nd = {}
    for _, (cell_type, sign) in enumerate(signs.items()):
        if sign == 1:
            # take the first half of the RdBu colormap, i.e. red
            colors_pd[cell_type] = next(exc_colors_pd)
            colors_nd[cell_type] = next(exc_colors_nd)

    for _, (cell_type, sign) in enumerate(signs_reversed.items()):
        if sign == -1:
            # take the second half of the RdBu colormap, i.e. blue
            colors_pd[cell_type] = next(inh_cmap_pd)
            colors_nd[cell_type] = next(inh_cmap_nd)
    self.colors_pd = colors_pd
    self.colors_nd = colors_nd

color

color(source_type, pd=True)

Get the color for a given source type.

Parameters:

Name	Type	Description	Default
source_type	str	The source type.	required
pd	bool	Whether to use PD or ND colors.	True

Returns:

Type	Description
tuple[float, float, float]	The color as an RGB tuple.

Source code in flyvision/analysis/moving_edge_currents.py

def color(self, source_type: str, pd: bool = True) -> tuple[float, float, float]:
    """
    Get the color for a given source type.

    Args:
        source_type: The source type.
        pd: Whether to use PD or ND colors.

    Returns:
        The color as an RGB tuple.
    """
    if pd:
        return self.colors_pd[source_type]
    return self.colors_nd[source_type]

zorder

zorder(
    source_types,
    source_type,
    start_exc=1000,
    start_inh=1000,
)

Get the z-order for a given source type.

Parameters:

Name	Type	Description	Default
source_types	list[str]	List of source types.	required
source_type	str	The source type.	required
start_exc	int	Starting z-order for excitatory cells.	1000
start_inh	int	Starting z-order for inhibitory cells.	1000

Returns:

Type	Description
int	The z-order for the given source type.

Source code in flyvision/analysis/moving_edge_currents.py

def zorder(
    self,
    source_types: list[str],
    source_type: str,
    start_exc: int = 1000,
    start_inh: int = 1000,
) -> int:
    """
    Get the z-order for a given source type.

    Args:
        source_types: List of source types.
        source_type: The source type.
        start_exc: Starting z-order for excitatory cells.
        start_inh: Starting z-order for inhibitory cells.

    Returns:
        The z-order for the given source type.
    """
    signs = self.signs()
    signs_reversed = {cell_type: signs[cell_type] for cell_type in source_types[::-1]}

    z_order = start_exc
    for _, (cell_type, sign) in enumerate(signs.items()):
        if sign == 1:
            if cell_type == source_type:
                return z_order
            z_order -= 10

    z_order = start_inh
    for _, (cell_type, sign) in enumerate(signs_reversed.items()):
        if sign == -1:
            if cell_type == source_type:
                return z_order
            z_order -= 10

ylims

ylims(source_types=None, offset=0.02)

Get the y-limits for temporal contributions summed over cells.

Parameters:

Name	Type	Description	Default
source_types	list[str] | None	List of source types to consider.	None
offset	float	Offset for the y-limits.	0.02

Returns:

Type	Description
dict[str, tuple[float, float]]	Y-limits for the given source types or all source types.

Source code in flyvision/analysis/moving_edge_currents.py

def ylims(
    self, source_types: list[str] | None = None, offset: float = 0.02
) -> dict[str, tuple[float, float]]:
    """
    Get the y-limits for temporal contributions summed over cells.

    Args:
        source_types: List of source types to consider.
        offset: Offset for the y-limits.

    Returns:
        Y-limits for the given source types or all source types.
    """
    if source_types is not None:
        return {
            cell_type: plt_utils.get_lims(c, offset)
            for cell_type, c in self.sum_over_cells().currents.items()
            if cell_type in source_types
        }
    return plt_utils.get_lims(list(self.sum_over_cells().currents.values()), offset)

plot_response

plot_response(
    contrast,
    angle,
    t_start=0,
    t_end=1,
    max_figure_height_cm=1.4477,
    panel_height_cm=1.4477,
    max_figure_width_cm=4.0513,
    panel_width_cm=4.0513,
    fontsize=5,
    model_average=True,
    color=(0, 0, 0),
    legend=False,
    hide_yaxis=True,
    trim_axes=True,
    quantile=None,
    scale_position=None,
    scale_label="{:.0f} ms",
    scale_unit=1000,
    hline=False,
    fig=None,
    ax=None,
)

Plot the response to a moving edge stimulus.

Parameters:

Name	Type	Description	Default
contrast	float	The contrast of the stimulus.	required
angle	float	The angle of the stimulus.	required
t_start	float	Start time for the plot.	0
t_end	float	End time for the plot.	1
max_figure_height_cm	float	Maximum figure height in centimeters.	1.4477
panel_height_cm	float	Height of each panel in centimeters.	1.4477
max_figure_width_cm	float	Maximum figure width in centimeters.	4.0513
panel_width_cm	float	Width of each panel in centimeters.	4.0513
fontsize	float	Font size for labels and titles.	5
model_average	bool	Whether to plot the model average.	True
color	tuple[float, float, float]	Color for the plot.	(0, 0, 0)
legend	bool	Whether to show the legend.	False
hide_yaxis	bool	Whether to hide the y-axis.	True
trim_axes	bool	Whether to trim the axes.	True
quantile	float | None	Quantile for shading.	None
scale_position	str | None	Position of the scale.	None
scale_label	str	Label format for the scale.	'{:.0f} ms'
scale_unit	float	Unit for the scale.	1000
hline	bool	Whether to show a horizontal line at 0.	False
fig	Figure | None	Existing figure to use.	None
ax	Axes | None	Existing axes to use.	None

Returns:

Type	Description
	Figure and axes objects.

Source code in flyvision/analysis/moving_edge_currents.py

def plot_response(
    self,
    contrast: float,
    angle: float,
    t_start: float = 0,
    t_end: float = 1,
    max_figure_height_cm: float = 1.4477,
    panel_height_cm: float = 1.4477,
    max_figure_width_cm: float = 4.0513,
    panel_width_cm: float = 4.0513,
    fontsize: float = 5,
    model_average: bool = True,
    color: tuple[float, float, float] = (0, 0, 0),
    legend: bool = False,
    hide_yaxis: bool = True,
    trim_axes: bool = True,
    quantile: float | None = None,
    scale_position: str | None = None,  # "lower left",
    scale_label: str = "{:.0f} ms",
    scale_unit: float = 1000,
    hline: bool = False,
    fig: plt.Figure | None = None,
    ax: plt.Axes | None = None,
):
    """
    Plot the response to a moving edge stimulus.

    Args:
        contrast: The contrast of the stimulus.
        angle: The angle of the stimulus.
        t_start: Start time for the plot.
        t_end: End time for the plot.
        max_figure_height_cm: Maximum figure height in centimeters.
        panel_height_cm: Height of each panel in centimeters.
        max_figure_width_cm: Maximum figure width in centimeters.
        panel_width_cm: Width of each panel in centimeters.
        fontsize: Font size for labels and titles.
        model_average: Whether to plot the model average.
        color: Color for the plot.
        legend: Whether to show the legend.
        hide_yaxis: Whether to hide the y-axis.
        trim_axes: Whether to trim the axes.
        quantile: Quantile for shading.
        scale_position: Position of the scale.
        scale_label: Label format for the scale.
        scale_unit: Unit for the scale.
        hline: Whether to show a horizontal line at 0.
        fig: Existing figure to use.
        ax: Existing axes to use.

    Returns:
        Figure and axes objects.
    """
    r_pd = (
        self.at_angle(angle)
        .at_contrast(contrast)
        .between_seconds(t_start, t_end)
        .responses.squeeze(axis=-2)
    )
    r_nd = (
        self.at_angle((angle - 180) % 360)
        .at_contrast(contrast)
        .between_seconds(t_start, t_end)
        .responses.squeeze(axis=-2)
    )

    if fig is None and ax is None:
        figsize = figsize_from_n_items(
            1,
            max_figure_height_cm=max_figure_height_cm,
            panel_height_cm=panel_height_cm,
            max_figure_width_cm=max_figure_width_cm,
            panel_width_cm=panel_width_cm,
        )
        fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
        ax = axes[0]

    color = [hex2color(PD), hex2color(ND)] if color is None else [color, color]

    if model_average:
        fig, ax, _, _ = plots.traces(
            [r_pd.mean(axis=0), r_nd.mean(axis=0)],
            x=self.between_seconds(t_start, t_end).time,
            color=color,
            linewidth=1,
            fontsize=fontsize,
            null_line=False,
            fig=fig,
            ax=ax,
            linestyle=["solid", "dashed"],
            legend="" if not legend else [f"{self.target_type}", "null direction"],
            scale_pos=scale_position,
            scale_label=scale_label,
            scale_unit=scale_unit,
        )
    else:
        fig, ax, _, _ = plots.traces(
            r_pd,
            x=self.between_seconds(t_start, t_end).time,
            mean_color=adapt_color_alpha(color[0], 1),
            color=adapt_color_alpha(color[0], 0.5),
            linewidth=0.25,
            zorder_traces=0,
            zorder_mean=10,
            fontsize=fontsize,
            null_line=False,
            highlight_mean=True,
            fig=fig,
            ax=ax,
        )
        plots.traces(
            r_nd,
            x=self.between_seconds(t_start, t_end).time,
            mean_color=adapt_color_alpha(color[1], 1),
            color=adapt_color_alpha(color[1], 0.5),
            linewidth=0.25,
            zorder_traces=0,
            zorder_mean=10,
            fontsize=fontsize,
            null_line=False,
            highlight_mean=True,
            fig=fig,
            linestyle="dashed",
            ax=ax,
        )
    if quantile:
        quantile_pd = np.quantile(r_pd, quantile, axis=0)
        quantile_nd = np.quantile(r_nd, quantile, axis=0)
        ax.fill_between(
            self.between_seconds(t_start, t_end).time,
            quantile_pd[0],
            quantile_pd[1],
            facecolor=adapt_color_alpha(color[0], 0.1),
            edgecolor=adapt_color_alpha(color[0], 1),
            linewidth=0.25,
        )
        ax.fill_between(
            self.between_seconds(t_start, t_end).time,
            quantile_nd[0],
            quantile_nd[1],
            facecolor=adapt_color_alpha(color[1], 0.1),
            edgecolor=adapt_color_alpha(color[1], 1),
            linewidth=0.25,
            linestyle="dashed",
        )

    if hline:
        # horizontal line at 0
        ax.axhline(0, color=(0, 0, 0, 1), linewidth=0.25, zorder=-10)

    if hide_yaxis:
        plt_utils.rm_spines(ax, ("left",))
    if trim_axes:
        plt_utils.trim_axis(ax)
    if legend:
        ax.legend(
            fontsize=fontsize,
            ncols=1,
            bbox_to_anchor=(1.05, 1),
            loc="upper left",
            borderaxespad=0.0,
        )
    return fig, ax

plot_response_pc_nc

plot_response_pc_nc(
    contrast,
    angle,
    t_start=0,
    t_end=1,
    max_figure_height_cm=1.4477,
    panel_height_cm=1.4477,
    max_figure_width_cm=4.0513,
    panel_width_cm=4.0513,
    fontsize=5,
    model_average=True,
    color=(0, 0, 0),
    legend=False,
    hide_yaxis=True,
    trim_axes=True,
    quantile=None,
    scale_position=None,
    scale_label="{:.0f} ms",
    scale_unit=1000,
    fig=None,
    ax=None,
    hline=False,
)

Plot the response to a moving edge stimulus with positive and negative contrasts.

Parameters:

Name	Type	Description	Default
contrast	float	The contrast of the stimulus.	required
angle	float	The angle of the stimulus.	required
t_start	float	Start time for the plot.	0
t_end	float	End time for the plot.	1
max_figure_height_cm	float	Maximum figure height in centimeters.	1.4477
panel_height_cm	float	Height of each panel in centimeters.	1.4477
max_figure_width_cm	float	Maximum figure width in centimeters.	4.0513
panel_width_cm	float	Width of each panel in centimeters.	4.0513
fontsize	float	Font size for labels and titles.	5
model_average	bool	Whether to plot the model average.	True
color	tuple[float, float, float]	Color for the plot.	(0, 0, 0)
legend	bool	Whether to show the legend.	False
hide_yaxis	bool	Whether to hide the y-axis.	True
trim_axes	bool	Whether to trim the axes.	True
quantile	float | None	Quantile for shading.	None
scale_position	str | None	Position of the scale.	None
scale_label	str	Label format for the scale.	'{:.0f} ms'
scale_unit	float	Unit for the scale.	1000
fig	Figure | None	Existing figure to use.	None
ax	Axes | None	Existing axes to use.	None
hline	bool	Whether to show a horizontal line at 0.	False

Returns:

Type	Description
tuple[Figure, Axes]	Figure and axes objects.

Source code in flyvision/analysis/moving_edge_currents.py

def plot_response_pc_nc(
    self,
    contrast: float,
    angle: float,
    t_start: float = 0,
    t_end: float = 1,
    max_figure_height_cm: float = 1.4477,
    panel_height_cm: float = 1.4477,
    max_figure_width_cm: float = 4.0513,
    panel_width_cm: float = 4.0513,
    fontsize: float = 5,
    model_average: bool = True,
    color: tuple[float, float, float] = (0, 0, 0),
    legend: bool = False,
    hide_yaxis: bool = True,
    trim_axes: bool = True,
    quantile: float | None = None,
    scale_position: str | None = None,
    scale_label: str = "{:.0f} ms",
    scale_unit: float = 1000,
    fig: plt.Figure | None = None,
    ax: plt.Axes | None = None,
    hline: bool = False,
) -> tuple[plt.Figure, plt.Axes]:
    """
    Plot the response to a moving edge stimulus with positive and negative contrasts.

    Args:
        contrast: The contrast of the stimulus.
        angle: The angle of the stimulus.
        t_start: Start time for the plot.
        t_end: End time for the plot.
        max_figure_height_cm: Maximum figure height in centimeters.
        panel_height_cm: Height of each panel in centimeters.
        max_figure_width_cm: Maximum figure width in centimeters.
        panel_width_cm: Width of each panel in centimeters.
        fontsize: Font size for labels and titles.
        model_average: Whether to plot the model average.
        color: Color for the plot.
        legend: Whether to show the legend.
        hide_yaxis: Whether to hide the y-axis.
        trim_axes: Whether to trim the axes.
        quantile: Quantile for shading.
        scale_position: Position of the scale.
        scale_label: Label format for the scale.
        scale_unit: Unit for the scale.
        fig: Existing figure to use.
        ax: Existing axes to use.
        hline: Whether to show a horizontal line at 0.

    Returns:
        Figure and axes objects.
    """
    r_pc = (
        self.at_angle(angle)
        .at_contrast(contrast)
        .between_seconds(t_start, t_end)
        .responses.squeeze(axis=-2)
    )
    r_nc = (
        self.at_angle(angle)
        .at_contrast(0 if contrast == 1 else 1)
        .between_seconds(t_start, t_end)
        .responses.squeeze(axis=-2)
    )

    if fig is None and ax is None:
        figsize = figsize_from_n_items(
            1,
            max_figure_height_cm=max_figure_height_cm,
            panel_height_cm=panel_height_cm,
            max_figure_width_cm=max_figure_width_cm,
            panel_width_cm=panel_width_cm,
        )
        fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
        ax = axes[0]

    color = [hex2color(PD), hex2color(ND)] if color is None else [color, color]

    if model_average:
        fig, ax, _, _ = plots.traces(
            [r_pc.mean(axis=0), r_nc.mean(axis=0)],
            x=self.between_seconds(t_start, t_end).time,
            color=color,
            linewidth=1,
            fontsize=fontsize,
            null_line=False,
            fig=fig,
            ax=ax,
            linestyle=["solid", "dotted"],
            legend="" if not legend else [f"{self.target_type}", "null contrast"],
            scale_pos=scale_position,
            scale_label=scale_label,
            scale_unit=scale_unit,
        )
    else:
        fig, ax, _, _ = plots.traces(
            r_pc,
            x=self.between_seconds(t_start, t_end).time,
            mean_color=adapt_color_alpha(color[0], 1),
            color=adapt_color_alpha(color[0], 0.5),
            linewidth=0.25,
            zorder_traces=0,
            zorder_mean=10,
            fontsize=fontsize,
            null_line=False,
            highlight_mean=True,
            fig=fig,
            ax=ax,
        )
        plots.traces(
            r_nc,
            x=self.between_seconds(t_start, t_end).time,
            mean_color=adapt_color_alpha(color[1], 1),
            color=adapt_color_alpha(color[1], 0.5),
            linewidth=0.25,
            zorder_traces=0,
            zorder_mean=10,
            fontsize=fontsize,
            null_line=False,
            highlight_mean=True,
            fig=fig,
            linestyle="dashed",
            ax=ax,
        )
    if quantile:
        quantile_pd = np.quantile(r_pc, quantile, axis=0)
        quantile_nd = np.quantile(r_nc, quantile, axis=0)
        ax.fill_between(
            self.between_seconds(t_start, t_end).time,
            quantile_pd[0],
            quantile_pd[1],
            facecolor=adapt_color_alpha(color[0], 0.1),
            edgecolor=adapt_color_alpha(color[0], 1),
            linewidth=0.25,
        )
        ax.fill_between(
            self.between_seconds(t_start, t_end).time,
            quantile_nd[0],
            quantile_nd[1],
            facecolor=adapt_color_alpha(color[1], 0.1),
            edgecolor=adapt_color_alpha(color[1], 1),
            linewidth=0.25,
            linestyle="dashed",
        )

    # horizontal line at 0
    if hline:
        ax.axhline(0, color=(0, 0, 0, 1), linewidth=0.25, zorder=-10)

    if hide_yaxis:
        plt_utils.rm_spines(ax, ("left",))
    if trim_axes:
        plt_utils.trim_axis(ax)
    if legend:
        ax.legend(
            fontsize=fontsize,
            ncols=1,
            bbox_to_anchor=(1.05, 1),
            loc="upper left",
            borderaxespad=0.0,
        )
    return fig, ax

plot_temporal_contributions

plot_temporal_contributions(
    contrast,
    angle,
    t_start=0,
    t_end=1,
    fontsize=5,
    linewidth=0.25,
    legend=False,
    legend_standalone=True,
    legend_figsize_cm=(4.0572, 1),
    legend_n_rows=None,
    max_figure_height_cm=3.3941,
    panel_height_cm=3.3941,
    max_figure_width_cm=4.0572,
    panel_width_cm=4.0572,
    model_average=True,
    highlight_mean=True,
    sum_exc_inh=False,
    only_sum=False,
    hide_source_types="auto",
    hide_source_types_bins=5,
    hide_source_types_cut_off_edge=1,
    hide_source_types_mode="below_cut_off",
    hide_yaxis=True,
    trim_axes=True,
    quantile=None,
    fig=None,
    ax=None,
    legend_ax=None,
    hline=True,
    legend_n_cols=None,
    baseline_color=None,
    colors=None,
)

Plot temporal contributions of different source types.

Parameters:

Name	Type	Description	Default
contrast	float	The contrast of the stimulus.	required
angle	float	The angle of the stimulus.	required
t_start	float	Start time for the plot.	0
t_end	float	End time for the plot.	1
fontsize	float	Font size for labels and titles.	5
linewidth	float	Line width for traces.	0.25
legend	bool	Whether to show the legend.	False
legend_standalone	bool	Whether to create a standalone legend.	True
legend_figsize_cm	tuple[float, float]	Figure size for the standalone legend.	(4.0572, 1)
legend_n_rows	int | None	Number of rows for the standalone legend.	None
max_figure_height_cm	float	Maximum figure height in centimeters.	3.3941
panel_height_cm	float	Height of each panel in centimeters.	3.3941
max_figure_width_cm	float	Maximum figure width in centimeters.	4.0572
panel_width_cm	float	Width of each panel in centimeters.	4.0572
model_average	bool	Whether to plot the model average.	True
highlight_mean	bool	Whether to highlight the mean trace.	True
sum_exc_inh	bool	Whether to sum excitatory and inhibitory contributions.	False
only_sum	bool	Whether to only plot the summed contributions.	False
hide_source_types	str | list | None	Source types to hide or “auto”.	'auto'
hide_source_types_bins	int	Number of bins for auto-hiding.	5
hide_source_types_cut_off_edge	int	Cut-off edge for auto-hiding.	1
hide_source_types_mode	str	Mode for auto-hiding source types.	'below_cut_off'
hide_yaxis	bool	Whether to hide the y-axis.	True
trim_axes	bool	Whether to trim the axes.	True
quantile	float | None	Quantile for shading.	None
fig	Figure | None	Existing figure to use.	None
ax	Axes | None	Existing axes to use.	None
legend_ax	Axes | None	Existing axes for the standalone legend.	None
hline	bool	Whether to show a horizontal line at 0.	True
legend_n_cols	int | None	Number of columns for the standalone legend.	None
baseline_color	tuple[float, float, float, float] | None	Color for the baseline.	None
colors	dict[str, tuple[float, float, float, float]] | None	Colors for each source type.	None

Returns:

Type	Description
	Figure, axes, and legend axes objects.

Example

view = MovingEdgeCurrentView(...)
fig, ax = view.plot_temporal_contributions(
    contrast=1.0,
    angle=0,
    t_start=0,
    t_end=1,
    fontsize=5,
    linewidth=0.25,
    legend=True
)

Source code in flyvision/analysis/moving_edge_currents.py

def plot_temporal_contributions(
    self,
    contrast: float,
    angle: float,
    t_start: float = 0,
    t_end: float = 1,
    fontsize: float = 5,
    linewidth: float = 0.25,
    legend: bool = False,
    legend_standalone: bool = True,
    legend_figsize_cm: tuple[float, float] = (4.0572, 1),
    legend_n_rows: int | None = None,
    max_figure_height_cm: float = 3.3941,
    panel_height_cm: float = 3.3941,
    max_figure_width_cm: float = 4.0572,
    panel_width_cm: float = 4.0572,
    model_average: bool = True,
    highlight_mean: bool = True,  # only applies if model_average is False
    sum_exc_inh: bool = False,
    only_sum: bool = False,
    hide_source_types: str | list | None = "auto",
    hide_source_types_bins: int = 5,
    hide_source_types_cut_off_edge: int = 1,
    hide_source_types_mode: str = "below_cut_off",
    hide_yaxis: bool = True,
    trim_axes: bool = True,
    quantile: float | None = None,
    fig: plt.Figure | None = None,
    ax: plt.Axes | None = None,
    legend_ax: plt.Axes | None = None,
    hline: bool = True,
    legend_n_cols: int | None = None,
    baseline_color: tuple[float, float, float, float] | None = None,
    colors: dict[str, tuple[float, float, float, float]] | None = None,
):
    """
    Plot temporal contributions of different source types.

    Args:
        contrast: The contrast of the stimulus.
        angle: The angle of the stimulus.
        t_start: Start time for the plot.
        t_end: End time for the plot.
        fontsize: Font size for labels and titles.
        linewidth: Line width for traces.
        legend: Whether to show the legend.
        legend_standalone: Whether to create a standalone legend.
        legend_figsize_cm: Figure size for the standalone legend.
        legend_n_rows: Number of rows for the standalone legend.
        max_figure_height_cm: Maximum figure height in centimeters.
        panel_height_cm: Height of each panel in centimeters.
        max_figure_width_cm: Maximum figure width in centimeters.
        panel_width_cm: Width of each panel in centimeters.
        model_average: Whether to plot the model average.
        highlight_mean: Whether to highlight the mean trace.
        sum_exc_inh: Whether to sum excitatory and inhibitory contributions.
        only_sum: Whether to only plot the summed contributions.
        hide_source_types: Source types to hide or "auto".
        hide_source_types_bins: Number of bins for auto-hiding.
        hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
        hide_source_types_mode: Mode for auto-hiding source types.
        hide_yaxis: Whether to hide the y-axis.
        trim_axes: Whether to trim the axes.
        quantile: Quantile for shading.
        fig: Existing figure to use.
        ax: Existing axes to use.
        legend_ax: Existing axes for the standalone legend.
        hline: Whether to show a horizontal line at 0.
        legend_n_cols: Number of columns for the standalone legend.
        baseline_color: Color for the baseline.
        colors: Colors for each source type.

    Returns:
        Figure, axes, and legend axes objects.

    Example:
        ```
        view = MovingEdgeCurrentView(...)
        fig, ax = view.plot_temporal_contributions(
            contrast=1.0,
            angle=0,
            t_start=0,
            t_end=1,
            fontsize=5,
            linewidth=0.25,
            legend=True
        )
        ```
    """
    if fig is None and ax is None:
        figsize = figsize_from_n_items(
            1,
            max_figure_height_cm=max_figure_height_cm,
            panel_height_cm=panel_height_cm,
            max_figure_width_cm=max_figure_width_cm,
            panel_width_cm=panel_width_cm,
        )
        fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
        ax = axes[0]
    cv_pd = (
        self.at_contrast(contrast)
        .at_angle(angle)
        .between_seconds(t_start, t_end)
        .sum_over_cells()
    )
    cv_nd = (
        self.at_contrast(contrast)
        .at_angle((angle - 180) % 360)
        .between_seconds(t_start, t_end)
        .sum_over_cells()
    )

    source_types = (
        self.at_contrast(contrast)
        .at_angle([angle, (angle - 180) % 360])
        .between_seconds(t_start, t_end)
        .filter_source_types(
            hide_source_types,
            hide_source_types_bins,
            hide_source_types_cut_off_edge,
            hide_source_types_mode,
        )
    )

    color_source_types = (
        self.at_contrast(contrast)
        .at_angle([angle, (angle - 180) % 360])
        .between_seconds(t_start, t_end)
        .filter_source_types(
            None,
            hide_source_types_bins,
            hide_source_types_cut_off_edge,
            hide_source_types_mode,
        )
    )
    cv_pd.init_colors(color_source_types)
    cv_nd.init_colors(color_source_types)

    def plot_mean_trace(
        time, trace, label, color, zorder, linestyle="solid", ax=None, fig=None
    ):
        ax.plot(
            time,
            trace,
            label=label,
            color=color,
            zorder=zorder,
            linestyle=linestyle,
        )

    def plot_individual_traces(
        traces, time, color, zorder, label, linestyle="solid", legend=None
    ):
        if not only_sum and not model_average:
            plots.traces(
                traces,
                time,
                mean_color=color,
                color=color,
                linewidth=linewidth,
                zorder_traces=0,
                zorder_mean=zorder,
                fontsize=fontsize,
                null_line=True,
                highlight_mean=highlight_mean,
                fig=fig,
                ax=ax,
                legend=legend or label,
                linestyle=linestyle,
            )

    def plot_quantile(traces, time, color, zorder, linestyle="solid"):
        if quantile:
            Q = np.quantile(traces, quantile, axis=0)
            ax.fill_between(
                time,
                Q[0],
                Q[1],
                facecolor=adapt_color_alpha(color, 0.1),
                edgecolor=color,
                linewidth=0.25,
                linestyle=linestyle,
                zorder=zorder - 1,
            )

    def plot_summed_trace(time, trace, label, color, zorder, linestyle="solid"):
        if np.any(trace):
            ax.plot(
                time,
                trace,
                label=label,
                color=color,
                zorder=zorder,
                linestyle=linestyle,
            )

    def get_summed_traces(signs, source_types, cv_pd, cv_nd):
        # sum over cell types then average over models
        exc_pd = np.zeros(cv_pd.shape)
        inh_pd = np.zeros(cv_pd.shape)
        exc_nd = np.zeros(cv_nd.shape)
        inh_nd = np.zeros(cv_nd.shape)
        # sum over cell types
        for source_type in source_types:
            if signs[source_type] == 1:
                exc_pd += cv_pd[source_type][:]  # (1, n_models, 1, n_timesteps)
                exc_nd += cv_nd[source_type][:]
            else:
                inh_pd += cv_pd[source_type][:]
                inh_nd += cv_nd[source_type][:]
        # (n_models, n_timesteps)
        return (
            exc_pd.squeeze(),
            inh_pd.squeeze(),
            exc_nd.squeeze(),
            inh_nd.squeeze(),
        )

    for source_type in source_types:
        if model_average and not only_sum:
            # mean traces solid for PD and dashed for ND
            if baseline_color is not None:
                color = baseline_color
            elif colors:
                color = colors[source_type]
            else:
                color = cv_pd.color(source_type)

            plot_mean_trace(
                cv_pd.time,
                cv_pd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                source_type,
                color,
                cv_pd.zorder(source_types, source_type),
                ax=ax,
                fig=fig,
            )
            plot_mean_trace(
                cv_nd.time,
                cv_nd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                source_type,
                color,
                linestyle="dashed",
                zorder=cv_pd.zorder(source_types, source_type),
                ax=ax,
                fig=fig,
            )

        elif not model_average and not only_sum:
            # individual traces
            plot_individual_traces(
                cv_pd[source_type][:].squeeze(axis=-2),
                cv_pd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                source_type,
            )
            plot_individual_traces(
                cv_nd[source_type][:].squeeze(axis=-2),
                cv_nd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                source_type,
                linestyle="dashed",
                legend="null direction",
            )

        # quantiles
        plot_quantile(
            cv_pd[source_type][:].squeeze(axis=-2),
            cv_pd.time,
            cv_pd.color(source_type),
            cv_pd.zorder(source_types, source_type),
            linestyle="solid",
        )
        plot_quantile(
            cv_nd[source_type][:].squeeze(axis=-2),
            cv_nd.time,
            cv_pd.color(source_type),
            cv_pd.zorder(source_types, source_type),
            linestyle="dashed",
        )
    if sum_exc_inh or only_sum:
        # plot summed traces
        signs = cv_pd.signs()
        exc_pd, inh_pd, exc_nd, inh_nd = get_summed_traces(
            signs, source_types, cv_pd, cv_nd
        )
        plot_summed_trace(
            cv_pd.time,
            exc_pd.mean(axis=0),
            "excitatory",
            (0.931, 0.0, 0.0, 1.0),
            zorder=2000,
        )
        plot_quantile(
            exc_pd,
            cv_pd.time,
            (0.931, 0.0, 0.0, 1.0),
            zorder=0,
            linestyle="solid",
        )
        plot_summed_trace(
            cv_nd.time,
            exc_nd.mean(axis=0),
            "excitatory",
            (0.931, 0.0, 0.0, 1.0),
            zorder=2000,
            linestyle="dashed",
        )
        plot_quantile(
            exc_nd,
            cv_pd.time,
            (0.931, 0.0, 0.0, 1.0),
            zorder=0,
            linestyle="dashed",
        )
        plot_summed_trace(
            cv_pd.time,
            inh_pd.mean(axis=0),
            "inhibitory",
            (0.0, 0.0, 0.849, 1.0),
            zorder=2000,
        )
        plot_quantile(
            inh_pd,
            cv_pd.time,
            (0.0, 0.0, 0.849, 1.0),
            zorder=0,
            linestyle="solid",
        )
        plot_summed_trace(
            cv_nd.time,
            inh_nd.mean(axis=0),
            "inhibitory",
            (0.0, 0.0, 0.849, 1.0),
            zorder=2000,
            linestyle="dashed",
        )
        plot_quantile(
            inh_nd,
            cv_pd.time,
            (0.0, 0.0, 0.849, 1.0),
            zorder=0,
            linestyle="dashed",
        )

    if hline:
        ax.hlines(
            0,
            cv_pd.time.min(),
            cv_pd.time.max(),
            color=(0, 0, 0, 1),
            linewidth=0.25,
            zorder=-10,
        )

    if legend:
        ax.legend(
            fontsize=fontsize,
            ncols=1,
            bbox_to_anchor=(1.05, 1),
            loc="upper left",
            borderaxespad=0.0,
        )
    else:
        ax.legend().set_visible(False)

    ax.set_xlabel("time (s)", fontsize=fontsize)
    #         ax.set_ylabel("current (a.u.)", fontsize=fontsize)

    if hide_yaxis:
        plt_utils.rm_spines(ax, ("left",))

    if trim_axes:
        plt_utils.trim_axis(ax)

    if legend_standalone:
        handles, labels = ax.get_legend_handles_labels()
        nd_handle = Line2D(
            [0], [0], color="k", lw=1, label="null direction", ls="dashed"
        )
        legend_n_rows = legend_n_rows or len(labels) + 1
        # legend_n_cols = (len(labels) + 1) // legend_n_rows
        legend_fig, legend_ax = plt_utils.standalone_legend(
            [*labels[::2], "null direction"],
            None,
            [*handles[::2], nd_handle],
            fontsize=fontsize,
            n_cols=legend_n_cols,
            handlelength=2,
            columnspacing=0.8,
            labelspacing=0.25,
            figsize=cm_to_inch(legend_figsize_cm),
            fig=fig if legend_ax is not None else None,
            ax=legend_ax,
        )
        return fig, ax, legend_fig, legend_ax
    return fig, ax

plot_temporal_contributions_pc_nc

plot_temporal_contributions_pc_nc(
    contrast,
    angle,
    t_start=0,
    t_end=1,
    fontsize=5,
    linewidth=0.25,
    legend=False,
    legend_standalone=True,
    legend_figsize_cm=(4.0572, 1),
    legend_n_rows=None,
    max_figure_height_cm=3.3941,
    panel_height_cm=3.3941,
    max_figure_width_cm=4.0572,
    panel_width_cm=4.0572,
    model_average=True,
    highlight_mean=True,
    sum_exc_inh=False,
    only_sum=False,
    hide_source_types="auto",
    hide_source_types_bins=5,
    hide_source_types_cut_off_edge=1,
    hide_source_types_mode="below_cut_off",
    hide_yaxis=True,
    trim_axes=True,
    quantile=None,
    fig=None,
    ax=None,
    legend_ax=None,
    null_linestyle="dotted",
    legend_n_cols=None,
)

Temporal contributions of different source types for positive/negative contrasts.

Parameters:

Name	Type	Description	Default
contrast	float	The contrast of the stimulus.	required
angle	float	The angle of the stimulus.	required
t_start	float	Start time for the plot.	0
t_end	float	End time for the plot.	1
fontsize	float	Font size for labels and titles.	5
linewidth	float	Line width for traces.	0.25
legend	bool	Whether to show the legend.	False
legend_standalone	bool	Whether to create a standalone legend.	True
legend_figsize_cm	tuple[float, float]	Figure size for the standalone legend.	(4.0572, 1)
legend_n_rows	int | None	Number of rows for the standalone legend.	None
max_figure_height_cm	float	Maximum figure height in centimeters.	3.3941
panel_height_cm	float	Height of each panel in centimeters.	3.3941
max_figure_width_cm	float	Maximum figure width in centimeters.	4.0572
panel_width_cm	float	Width of each panel in centimeters.	4.0572
model_average	bool	Whether to plot the model average.	True
highlight_mean	bool	Whether to highlight the mean trace.	True
sum_exc_inh	bool	Whether to sum excitatory and inhibitory contributions.	False
only_sum	bool	Whether to only plot the summed contributions.	False
hide_source_types	str | list | None	Source types to hide or “auto”.	'auto'
hide_source_types_bins	int	Number of bins for auto-hiding.	5
hide_source_types_cut_off_edge	int	Cut-off edge for auto-hiding.	1
hide_source_types_mode	str	Mode for auto-hiding source types.	'below_cut_off'
hide_yaxis	bool	Whether to hide the y-axis.	True
trim_axes	bool	Whether to trim the axes.	True
quantile	float | None	Quantile for shading.	None
fig	Figure | None	Existing figure to use.	None
ax	Axes | None	Existing axes to use.	None
legend_ax	Axes | None	Existing axes for the standalone legend.	None
null_linestyle	str	Linestyle for null direction traces.	'dotted'
legend_n_cols	int | None	Number of columns for the standalone legend.	None

Returns:

Type	Description
tuple[Figure, Axes, Figure | None, Axes | None]	Figure, axes, and legend axes objects.

Example

view = MovingEdgeCurrentView(...)
fig, ax = view.plot_temporal_contributions_pc_nc(
    contrast=1.0,
    angle=0,
    t_start=0,
    t_end=1,
    fontsize=5,
    linewidth=0.25,
    legend=True
)

Source code in flyvision/analysis/moving_edge_currents.py

def plot_temporal_contributions_pc_nc(
    self,
    contrast: float,
    angle: float,
    t_start: float = 0,
    t_end: float = 1,
    fontsize: float = 5,
    linewidth: float = 0.25,
    legend: bool = False,
    legend_standalone: bool = True,
    legend_figsize_cm: tuple[float, float] = (4.0572, 1),
    legend_n_rows: int | None = None,
    max_figure_height_cm: float = 3.3941,
    panel_height_cm: float = 3.3941,
    max_figure_width_cm: float = 4.0572,
    panel_width_cm: float = 4.0572,
    model_average: bool = True,
    highlight_mean: bool = True,
    sum_exc_inh: bool = False,
    only_sum: bool = False,
    hide_source_types: str | list | None = "auto",
    hide_source_types_bins: int = 5,
    hide_source_types_cut_off_edge: int = 1,
    hide_source_types_mode: str = "below_cut_off",
    hide_yaxis: bool = True,
    trim_axes: bool = True,
    quantile: float | None = None,
    fig: plt.Figure | None = None,
    ax: plt.Axes | None = None,
    legend_ax: plt.Axes | None = None,
    null_linestyle: str = "dotted",
    legend_n_cols: int | None = None,
) -> tuple[plt.Figure, plt.Axes, plt.Figure | None, plt.Axes | None]:
    """
    Temporal contributions of different source types for positive/negative contrasts.

    Args:
        contrast: The contrast of the stimulus.
        angle: The angle of the stimulus.
        t_start: Start time for the plot.
        t_end: End time for the plot.
        fontsize: Font size for labels and titles.
        linewidth: Line width for traces.
        legend: Whether to show the legend.
        legend_standalone: Whether to create a standalone legend.
        legend_figsize_cm: Figure size for the standalone legend.
        legend_n_rows: Number of rows for the standalone legend.
        max_figure_height_cm: Maximum figure height in centimeters.
        panel_height_cm: Height of each panel in centimeters.
        max_figure_width_cm: Maximum figure width in centimeters.
        panel_width_cm: Width of each panel in centimeters.
        model_average: Whether to plot the model average.
        highlight_mean: Whether to highlight the mean trace.
        sum_exc_inh: Whether to sum excitatory and inhibitory contributions.
        only_sum: Whether to only plot the summed contributions.
        hide_source_types: Source types to hide or "auto".
        hide_source_types_bins: Number of bins for auto-hiding.
        hide_source_types_cut_off_edge: Cut-off edge for auto-hiding.
        hide_source_types_mode: Mode for auto-hiding source types.
        hide_yaxis: Whether to hide the y-axis.
        trim_axes: Whether to trim the axes.
        quantile: Quantile for shading.
        fig: Existing figure to use.
        ax: Existing axes to use.
        legend_ax: Existing axes for the standalone legend.
        null_linestyle: Linestyle for null direction traces.
        legend_n_cols: Number of columns for the standalone legend.

    Returns:
        Figure, axes, and legend axes objects.

    Example:
        ```
        view = MovingEdgeCurrentView(...)
        fig, ax = view.plot_temporal_contributions_pc_nc(
            contrast=1.0,
            angle=0,
            t_start=0,
            t_end=1,
            fontsize=5,
            linewidth=0.25,
            legend=True
        )
        ```
    """
    if fig is None and ax is None:
        figsize = figsize_from_n_items(
            1,
            max_figure_height_cm=max_figure_height_cm,
            panel_height_cm=panel_height_cm,
            max_figure_width_cm=max_figure_width_cm,
            panel_width_cm=panel_width_cm,
        )
        fig, axes = figsize.axis_grid(hspace=0.0, wspace=0, fontsize=fontsize)
        ax = axes[0]
    cv_pd = (
        self.at_contrast(contrast)
        .at_angle(angle)
        .between_seconds(t_start, t_end)
        .sum_over_cells()
    )
    cv_nd = (
        self.at_contrast(contrast)
        .at_angle((angle - 180) % 360)
        .between_seconds(t_start, t_end)
        .sum_over_cells()
    )

    source_types = (
        self.at_contrast(contrast)
        .at_angle([angle, (angle - 180) % 360])
        .between_seconds(t_start, t_end)
        .filter_source_types(
            hide_source_types,
            hide_source_types_bins,
            hide_source_types_cut_off_edge,
            hide_source_types_mode,
        )
    )

    color_source_types = (
        self.at_contrast(contrast)
        .at_angle([angle, (angle - 180) % 360])
        .between_seconds(t_start, t_end)
        .filter_source_types(
            None,
            hide_source_types_bins,
            hide_source_types_cut_off_edge,
            hide_source_types_mode,
        )
    )
    cv_pd.init_colors(color_source_types)
    cv_nd.init_colors(color_source_types)

    def plot_mean_trace(
        time, trace, label, color, zorder, linestyle="solid", ax=None, fig=None
    ):
        ax.plot(
            time,
            trace,
            label=label,
            color=color,
            zorder=zorder,
            linestyle=linestyle,
        )

    def plot_individual_traces(
        traces, time, color, zorder, label, linestyle="solid", legend=None
    ):
        if not only_sum and not model_average:
            plots.traces(
                traces,
                time,
                mean_color=color,
                color=color,
                linewidth=linewidth,
                zorder_traces=0,
                zorder_mean=zorder,
                fontsize=fontsize,
                null_line=True,
                highlight_mean=highlight_mean,
                fig=fig,
                ax=ax,
                legend=legend or label,
                linestyle=linestyle,
            )

    def plot_quantile(traces, time, color, zorder, linestyle="solid"):
        if quantile:
            Q = np.quantile(traces, quantile, axis=0)
            ax.fill_between(
                time,
                Q[0],
                Q[1],
                facecolor=adapt_color_alpha(color, 0.1),
                edgecolor=color,
                linewidth=0.25,
                linestyle=linestyle,
                zorder=zorder - 1,
            )

    def plot_summed_trace(time, trace, label, color, zorder, linestyle="solid"):
        if np.any(trace):
            ax.plot(
                time,
                trace,
                label=label,
                color=color,
                zorder=zorder,
                linestyle=linestyle,
            )

    def get_summed_traces(signs, source_types, cv_pd, cv_nd):
        # sum over cell types then average over models
        exc_pd = np.zeros(cv_pd.shape)
        inh_pd = np.zeros(cv_pd.shape)
        exc_nd = np.zeros(cv_nd.shape)
        inh_nd = np.zeros(cv_nd.shape)
        # sum over cell types
        for source_type in source_types:
            if signs[source_type] == 1:
                exc_pd += cv_pd[source_type][:]  # (1, n_models, 1, n_timesteps)
                exc_nd += cv_nd[source_type][:]
            else:
                inh_pd += cv_pd[source_type][:]
                inh_nd += cv_nd[source_type][:]
        # (n_models, n_timesteps)
        return (
            exc_pd.squeeze(),
            inh_pd.squeeze(),
            exc_nd.squeeze(),
            inh_nd.squeeze(),
        )

    for source_type in source_types:
        if model_average and not only_sum:
            # mean traces solid for PD and dashed for ND
            color = cv_pd.color(source_type)

            plot_mean_trace(
                cv_pd.time,
                cv_pd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                source_type,
                color,
                cv_pd.zorder(source_types, source_type),
                ax=ax,
                fig=fig,
            )
            plot_mean_trace(
                cv_nd.time,
                cv_nd[source_type][:].squeeze(axis=-2).T.mean(axis=1),
                source_type,
                color,
                linestyle=null_linestyle,
                zorder=cv_pd.zorder(source_types, source_type),
                ax=ax,
                fig=fig,
            )

        elif not model_average and not only_sum:
            # individual traces
            plot_individual_traces(
                cv_pd[source_type][:].squeeze(axis=-2),
                cv_pd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                source_type,
            )
            plot_individual_traces(
                cv_nd[source_type][:].squeeze(axis=-2),
                cv_nd.time,
                cv_pd.color(source_type),
                cv_pd.zorder(source_types, source_type),
                source_type,
                linestyle=null_linestyle,
                legend="null direction",
            )

        # quantiles
        plot_quantile(
            cv_pd[source_type][:].squeeze(axis=-2),
            cv_pd.time,
            cv_pd.color(source_type),
            cv_pd.zorder(source_types, source_type),
            linestyle="solid",
        )
        plot_quantile(
            cv_nd[source_type][:].squeeze(axis=-2),
            cv_nd.time,
            cv_pd.color(source_type),
            cv_pd.zorder(source_types, source_type),
            linestyle=null_linestyle,
        )
    if sum_exc_inh or only_sum:
        # plot summed traces
        signs = cv_pd.signs()
        exc_pd, inh_pd, exc_nd, inh_nd = get_summed_traces(
            signs, source_types, cv_pd, cv_nd
        )
        plot_summed_trace(
            cv_pd.time,
            exc_pd.mean(axis=0),
            "excitatory",
            (0.931, 0.0, 0.0, 1.0),
            zorder=2000,
        )
        plot_quantile(
            exc_pd,
            cv_pd.time,
            (0.931, 0.0, 0.0, 1.0),
            zorder=0,
            linestyle="solid",
        )
        plot_summed_trace(
            cv_nd.time,
            exc_nd.mean(axis=0),
            "excitatory",
            (0.931, 0.0, 0.0, 1.0),
            zorder=2000,
            linestyle=null_linestyle,
        )
        plot_quantile(
            exc_nd,
            cv_pd.time,
            (0.931, 0.0, 0.0, 1.0),
            zorder=0,
            linestyle=null_linestyle,
        )
        plot_summed_trace(
            cv_pd.time,
            inh_pd.mean(axis=0),
            "inhibitory",
            (0.0, 0.0, 0.849, 1.0),
            zorder=2000,
        )
        plot_quantile(
            inh_pd,
            cv_pd.time,
            (0.0, 0.0, 0.849, 1.0),
            zorder=0,
            linestyle="solid",
        )
        plot_summed_trace(
            cv_nd.time,
            inh_nd.mean(axis=0),
            "inhibitory",
            (0.0, 0.0, 0.849, 1.0),
            zorder=2000,
            linestyle=null_linestyle,
        )
        plot_quantile(
            inh_nd,
            cv_pd.time,
            (0.0, 0.0, 0.849, 1.0),
            zorder=0,
            linestyle=null_linestyle,
        )

    if legend:
        ax.legend(
            fontsize=fontsize,
            ncols=1,
            bbox_to_anchor=(1.05, 1),
            loc="upper left",
            borderaxespad=0.0,
        )
    else:
        ax.legend().set_visible(False)

    ax.set_xlabel("time (s)", fontsize=fontsize)
    #         ax.set_ylabel("current (a.u.)", fontsize=fontsize)

    if hide_yaxis:
        plt_utils.rm_spines(ax, ("left",))

    if trim_axes:
        plt_utils.trim_axis(ax)

    if legend_standalone:
        handles, labels = ax.get_legend_handles_labels()
        nd_handle = Line2D(
            [0], [0], color="k", lw=1, label="null direction", ls=null_linestyle
        )
        legend_n_rows = legend_n_rows or len(labels) + 1
        # legend_n_cols = (len(labels) + 1) // legend_n_rows
        legend_fig, legend_ax = plt_utils.standalone_legend(
            [*labels[::2], "null direction"],
            None,
            [*handles[::2], nd_handle],
            fontsize=fontsize,
            n_cols=legend_n_cols,
            handlelength=2,
            columnspacing=0.8,
            labelspacing=0.25,
            figsize=cm_to_inch(legend_figsize_cm),
            fig=fig if legend_ax is not None else None,
            ax=legend_ax,
        )
        return fig, ax, legend_fig, legend_ax
    return fig, ax, None, None