PhotometryData

This class is used to handle, manipulate, and analyze large trial-wise photometry datasets. Has functionality for filtering, plotting, averaging, and analyzing trials.

---

Example Usage

Reading and writing

trials = PhotometryData.read_h5ad(fpath)
trials.write_h5ad(fpath)

Filtering, averaging, and plotting

filtered = trials.filter_rows(
    trials.obs['trial_label'] != 'NoResponse',
)

avg = filtered.collapse(
    group_on=['trial_label'],
    metrics={'std' : np.std},
    data_cols=['event', 'AUC'],
    count_col='n_trials',
)

avg.plot_all(
    label_with=['trial_label', 'n_trials'],
    err_layer='std',
)

Recentering

recentered = trials.window(
    centers = trials.obs['event'],
    bounds = (-2, 5),
    event_cols=['event', 'lever1', 'lever2', 'loud_noise']
)

Area under curve

trials.area_under_curve(
    centers=0,
    bounds=(0, 6.5),
)

Export to flat formats

trials.trials_to_long_df(downsample=10)
trials.trials_to_wide_df(signal_prefix='X', downsample=20)

---

Handle and analyze trial-wise photometry time-series data.

Initialize a PhotometryData object from an AnnData object.

Parameters:

• adata (AnnData) –

  AnnData object containing time-series data.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def __init__(self, adata: ad.AnnData):
    """Initialize a `PhotometryData` object from an AnnData object.

    Args:
        adata (ad.AnnData): AnnData object containing time-series data.
    """
    assert isinstance(adata, ad.AnnData)
    self.adata = adata

from_arrays(obs, data, time_points, layers=None, metadata=None) classmethod

Construct PhotometryData from arrays and observation metadata.

Parameters:

• obs (DataFrame) –

  Per-trial observation metadata.

• data (ndarray) –

  Time-series data of shape (n_trials, n_time).

• time_points (ndarray) –

  Time axis of shape (n_time,).

• layers (dict[str, ndarray] | None, default: None ) –

  Optional named layers matching the shape of data. Defaults to None.

• metadata (dict[str, Any] | None, default: None ) –

  Optional unstructured metadata stored in .uns. Defaults to None.

Returns:

• PhotometryData ( Self ) –

  Constructed photometry data object.

Source code in pyFiberPhotometry/core/PhotometeryData.py

@classmethod
def from_arrays(
    cls,
    obs: pd.DataFrame, 
    data: np.ndarray, 
    time_points: np.ndarray,
    layers: dict[str, np.ndarray] | None = None,
    metadata: dict[str, Any] | None = None,
    ) -> Self:
    """Construct `PhotometryData` from arrays and observation metadata.

    Args:
        obs (pd.DataFrame): Per-trial observation metadata.
        data (np.ndarray): Time-series data of shape (n_trials, n_time).
        time_points (np.ndarray): Time axis of shape (n_time,).
        layers (dict[str, np.ndarray] | None, optional): Optional named
            layers matching the shape of ``data``. Defaults to ``None``.
        metadata (dict[str, Any] | None, optional): Optional unstructured
            metadata stored in ``.uns``. Defaults to ``None``.

    Returns:
        PhotometryData: Constructed photometry data object.
    """
    obs = obs.copy()
    obs.reset_index(drop=True, inplace=True)
    obs.index = obs.index.astype(str)
    var = pd.DataFrame({"t": time_points})
    var.index = var.index.astype(str)

    A = ad.AnnData(X=data, obs=obs, var=var, uns=metadata)
    for k, v in (layers or {}).items():
        assert v.shape == data.shape
        A.layers[k] = v
    return cls(A)

read_h5ad(path) classmethod

Read a PhotometryData object from an .h5ad file.

Parameters:

• path (str) –

  Path to the .h5ad file.

Returns:

• PhotometryData ( Self ) –

  Loaded PhotometryData instance.

Source code in pyFiberPhotometry/core/PhotometeryData.py

@classmethod
def read_h5ad(cls, path: str) -> Self:
    """Read a `PhotometryData` object from an `.h5ad` file.

    Args:
        path (str): Path to the `.h5ad` file.

    Returns:
        PhotometryData: Loaded `PhotometryData` instance.
    """
    return cls(ad.read_h5ad(path))

read_zarr(path) classmethod

Read a PhotometryData object from zarr storage.

Parameters:

• path (str) –

  Path to the zarr storage.

Returns:

• PhotometryData ( Self ) –

  Loaded PhotometryData instance.

Source code in pyFiberPhotometry/core/PhotometeryData.py

@classmethod
def read_zarr(cls, path: str) -> Self:
    """Read a `PhotometryData` object from zarr storage.

    Args:
        path (str): Path to the zarr storage.

    Returns:
        PhotometryData: Loaded `PhotometryData` instance.
    """
    return cls(ad.read_zarr(path))

write_h5ad(path)

Write the underlying AnnData to an .h5ad file.

Parameters:

• path (str) –

  Path to the output .h5ad file.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def write_h5ad(self, path: str) -> None:
    """Write the underlying AnnData to an `.h5ad` file.

    Args:
        path (str): Path to the output `.h5ad` file.

    Returns:
        None
    """
    self.adata.write_h5ad(path)

write_zarr(path)

Write the underlying AnnData to zarr storage.

Parameters:

• path (str) –

  Path to the output zarr storage.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def write_zarr(self, path: str) -> None:
    """Write the underlying AnnData to zarr storage.

    Args:
        path (str): Path to the output zarr storage.

    Returns:
        None
    """
    self.adata.write_zarr(path)

append_on_disk_h5ad(path, join='inner', merge='same', uns_merge='first')

Append this object's data to an existing .h5ad file on disk.

Creates the file if it does not already exist.

Parameters:

• path (str) –

  Path to the target .h5ad file.

• join (str, default: 'inner' ) –

  How to align values when concatenating. Defaults to 'inner'.

• merge (str, default: 'same' ) –

  How elements not aligned to the concatenated axis are selected. Defaults to 'same'.

• uns_merge (str, default: 'first' ) –

  How the elements of .uns are selected. Defaults to 'first'.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def append_on_disk_h5ad(self, path: str, join: str = 'inner', merge: str = 'same', uns_merge: str = 'first') -> None:
    """Append this object's data to an existing `.h5ad` file on disk.

    Creates the file if it does not already exist.

    Args:
        path (str): Path to the target `.h5ad` file.
        join (str, optional): How to align values when concatenating.
            Defaults to ``'inner'``.
        merge (str, optional): How elements not aligned to the concatenated
            axis are selected. Defaults to ``'same'``.
        uns_merge (str, optional): How the elements of ``.uns`` are
            selected. Defaults to ``'first'``.

    Returns:
        None
    """
    if not os.path.exists(path):
        self.write_h5ad(path)
        return

    # create tmp files and rename 
    base, ext = os.path.splitext(path)
    tmp_path_new = base + '_new_tmp' + ext
    self.write_h5ad(tmp_path_new)

    tmp_path_old = base + '_base_tmp' + ext
    os.rename(path, tmp_path_old)

    try:
        concat_on_disk(
            in_files=[tmp_path_old, tmp_path_new],
            out_file=path,
            axis='obs',
            join=join,
            merge=merge,
            uns_merge=uns_merge,
        )
    except Exception as e:
        os.rename(tmp_path_old, path)
        os.remove(tmp_path_new)
        raise Exception(f'In core.PhotometryData.append_on_disk_h5ad() concat_on_disk: {e}')

    os.remove(tmp_path_new)
    os.remove(tmp_path_old)
    return

downsample(factor)

Downsample the time dimension of the dataset.

Parameters:

• factor (int) –

  Downsampling factor applied along the time axis.

Returns:

• PhotometryData ( Self ) –

  New downsampled photometry data object.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def downsample(self, factor: int) -> Self:
    """Downsample the time dimension of the dataset.

    Args:
        factor (int): Downsampling factor applied along the time axis.

    Returns:
        PhotometryData: New downsampled photometry data object.
    """
    X_new = downsample_ndarray(self.adata.X, factor=factor, axis=1)
    t_new = downsample_1d(self.var['t'].to_numpy(), factor=factor)
    layers_new = {k : downsample_ndarray(v, factor=factor, axis=1) for k, v in self.adata.layers.items()}
    metadata = self.uns.copy()
    freq_cur = metadata.get('frequency', None)
    if freq_cur is not None: 
        freq_new = freq_cur / factor
        metadata.update({'frequency': freq_new})

    return type(self).from_arrays(
        obs=self.obs,
        data=X_new,
        time_points=t_new,
        layers=layers_new,
        metadata=metadata,
    )

combine_obj(to_append, inplace=False, join='inner', merge='same', uns_merge='same')

Concatenate this object with one or more PhotometryData objects.

Parameters:

• to_append (PhotometryData | list[PhotometryData]) –

  Object(s) to append.

• inplace (bool, default: False ) –

  Whether to modify the original object or return a new merged one. Defaults to False.

• join (str, default: 'inner' ) –

  How to align values when concatenating. If "outer", the union of the other axis is taken. If "inner", the intersection. Defaults to 'inner'.

• merge (str, default: 'same' ) –

  How elements not aligned to the axis being concatenated along are selected. Currently implemented strategies include: None: No elements are kept. 'same': Elements that are the same in each of the objects. 'unique': Elements for which there is only one possible value. 'first': The first element seen at each from each position. 'only': Elements that show up in only one of the objects. Defaults to 'same'.

• uns_merge (str, default: 'same' ) –

  How the elements of .uns are selected. Uses the same set of strategies as the merge argument, except applied recursively. Defaults to 'same'.

Returns:

• None | Self –

  None | PhotometryData: Combined PhotometryData or None depending on inplace.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def combine_obj(
        self, 
        to_append: "PhotometryData" | list["PhotometryData"], 
        inplace: bool = False, 
        join: str = 'inner', 
        merge: str ='same', 
        uns_merge: str = 'same'
        ) -> None | Self:
    """Concatenate this object with one or more `PhotometryData` objects.

    Args:
        to_append (PhotometryData | list[PhotometryData]): Object(s) to append.
        inplace (bool, optional): Whether to modify the original object or
            return a new merged one. Defaults to ``False``.
        join (str, optional): How to align values when concatenating. If
            ``"outer"``, the union of the other axis is taken. If
            ``"inner"``, the intersection. Defaults to ``'inner'``.
        merge (str, optional): How elements not aligned to the axis being
            concatenated along are selected. Currently implemented
            strategies include:
            None: No elements are kept.
            'same': Elements that are the same in each of the objects.
            'unique': Elements for which there is only one possible value.
            'first': The first element seen at each from each position.
            'only': Elements that show up in only one of the objects.
            Defaults to ``'same'``.
        uns_merge (str, optional): How the elements of ``.uns`` are
            selected. Uses the same set of strategies as the ``merge``
            argument, except applied recursively. Defaults to ``'same'``.

    Returns:
        None | PhotometryData: Combined `PhotometryData` or ``None``
            depending on ``inplace``.
    """
    if not isinstance(to_append, list):
        to_append = [to_append]

    adatas = [getattr(self, 'adata')] + [getattr(obj, 'adata') for obj in to_append]
    for i in range(len(adatas)):
        adatas[i].obs = adatas[i].obs.infer_objects()

    merged_adata = ad.concat(
        adatas=adatas,
        axis='obs',
        join=join,
        merge=merge,
        uns_merge=uns_merge,
        index_unique=''
    )
    merged_adata.obs.reset_index(drop=True, inplace=True) # type: ignore
    if inplace:
        self.adata = merged_adata
        return
    else:
        return type(self)(merged_adata)

collapse(group_on, method=np.nanmean, metrics={'std': np.std}, data_cols=[], collapse_cols=None, count_col='n')

Collapse trials by grouping obs and aggregating data.

Parameters:

• group_on (list[str] | None) –

  Columns in obs used to define groups. If None or [], the data is fully collapsed.

• method (Callable, default: nanmean ) –

  Aggregation function for the main X matrix. Defaults to np.nanmean.

• metrics (dict[str, Callable], default: {'std': std} ) –

  Additional named aggregation functions stored as layers. Defaults to {'std': np.std}.

• data_cols (list[str], default: [] ) –

  Observation columns to aggregate with each method.

• collapse_cols (list[str] | None, default: None ) –

  Optional columns to collapse to lists. Defaults to None.

• count_col (str | None, default: 'n' ) –

  Optional column name to store group counts. If None, no count_col is made. Defaults to 'n'.

Returns:

• PhotometryData ( Self ) –

  Collapsed photometry data object. Metrics are also applied to data_cols with the metric key appended to column name (ex: 'trial_std').

Source code in pyFiberPhotometry/core/PhotometeryData.py

def collapse(
        self,
        group_on: list[str] | None,
        method: Callable = np.nanmean,
        metrics: dict[str, Callable] = {'std':np.std},
        data_cols: list[str] = [],
        collapse_cols: list[str] | None = None,
        count_col: str | None = 'n'
        ) -> Self:
    """Collapse trials by grouping `obs` and aggregating data.

    Args:
        group_on (list[str] | None): Columns in ``obs`` used to define
            groups. If ``None`` or ``[]``, the data is fully collapsed.
        method (Callable, optional): Aggregation function for the main
            ``X`` matrix. Defaults to ``np.nanmean``.
        metrics (dict[str, Callable], optional): Additional named
            aggregation functions stored as layers. Defaults to
            ``{'std': np.std}``.
        data_cols (list[str]): Observation columns to aggregate with each method.
        collapse_cols (list[str] | None, optional): Optional columns to
            collapse to lists. Defaults to ``None``.
        count_col (str | None, optional): Optional column name to store
            group counts. If ``None``, no ``count_col`` is made.
            Defaults to ``'n'``.

    Returns:
        PhotometryData: Collapsed photometry data object. Metrics are 
            also applied to ``data_cols`` with the metric key appended
            to column name (ex: ``'trial_std'``).
    """
    obs_agg, X_agg = self._agg(method=method, group_on=group_on, data_cols=data_cols, count_col=count_col, collapse_cols=collapse_cols)

    layers = {}
    for key, func in metrics.items():
        obs_lay, X_lay = self._agg(method=func, group_on=group_on, data_cols=data_cols, count_col=count_col, collapse_cols=None)
        layers[key] = X_lay
        obs_agg = obs_agg.join(obs_lay[data_cols], rsuffix='_' + str(key))

    new_obj = type(self).from_arrays(
        obs=obs_agg,
        data=X_agg,
        time_points=self.adata.var['t'],
        layers=layers,
        metadata=self.adata.uns
    )
    return new_obj

window(centers, bounds, freq=None, series=None, event_cols=None)

Return a new PhotometryData object with windowed time series.

Parameters:

• centers (ndarray | float) –

  Specified centers of the windows.

• bounds (tuple[float, float]) –

  Lower and upper bounds of the windows relative to centers.

• freq (float | None, default: None ) –

  Frequency of series. If None, self.freq is used. Defaults to None.

• series (ndarray | None, default: None ) –

  Time-like sorted series that windows will be calculated in. If None, self.ts is used. Defaults to None.

• event_cols (list[str] | None, default: None ) –

  Columns of self.adata.obs to re-center to centers. Defaults to None.

Returns:

• PhotometryData ( Self ) –

  Windowed photometry data object.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def window(
    self,
    centers: np.ndarray | float,
    bounds: tuple[float, float],
    freq: float | None = None,
    series: np.ndarray | None = None,
    event_cols: list[str] | None = None,
    ) -> Self:
    """Return a new `PhotometryData` object with windowed time series.

    Args:
        centers (np.ndarray | float): Specified centers of the
            windows.
        bounds (tuple[float, float]): Lower and upper bounds of the
            windows relative to centers.
        freq (float | None, optional): Frequency of ``series``. If
            ``None``, ``self.freq`` is used. Defaults to ``None``.
        series (np.ndarray | None, optional): Time-like sorted series that
            windows will be calculated in. If ``None``, ``self.ts`` is
            used. Defaults to ``None``.
        event_cols (list[str] | None, optional): Columns of ``self.adata.obs``
            to re-center to ``centers``. Defaults to ``None``.

    Returns:
        PhotometryData: Windowed photometry data object.
    """
    if isinstance(centers, (float, int)): 
        centers = np.full(shape=self.X.shape[0], fill_value=centers, dtype=float)

    series = self.ts if series is None else np.asarray(series)
    freq = self.freq if freq is None else float(freq)
    centers = np.asarray(centers)

    window_idxs = self._get_window_idxs(series=series, centers=centers, bounds=bounds, freq=freq)
    row_slice = np.arange(self.X.shape[0])[:, None]

    data = self.X[row_slice, window_idxs]
    layers = {}
    for k, layer in self.adata.layers.items():
        layers[k] = layer[row_slice, window_idxs]

    time_points = reconstruct_time_points(bounds=bounds, freq=freq)
    obs = self.adata.obs.copy()
    if event_cols is not None:
        obs[event_cols] = obs[event_cols] - centers[:, None]

    out: Self = type(self).from_arrays(
        obs=obs,
        data=data,
        time_points=time_points,
        layers=layers,
        metadata=self.adata.uns,
    )
    return out

area_under_curve(centers=None, bounds=None, transformation=None)

Calculate the area under the curve.

Optionally calculate the area only on specific windows.

Parameters:

• centers (ndarray | float, default: None ) –

  Specified centers of the windows in the same units. If None the area under the whole curve is calculated. Default is None.

• bounds (tuple[float, float], default: None ) –

  Lower and upper bounds of the windows relative to centers. If None the area under the whole curve is calculated. Default is None.

• transformation (Callable | None, default: None ) –

  If not None, a transformation to apply to the singal before taking the area. Shoule always return an array of the same shape and size. Default is None.

Returns np.ndarray: an array of areas of length n_trials

Source code in pyFiberPhotometry/core/PhotometeryData.py

def area_under_curve(
        self,
        centers: np.ndarray | float | None = None,
        bounds: tuple[float, float] | None = None,
        transformation: Callable | None = None
        ) -> np.ndarray:
    '''Calculate the area under the curve.

    Optionally calculate the area only on specific windows.

    Args:
        centers (np.ndarray | float): Specified centers of the
            windows in the same units. If ``None`` the area under the
            whole curve is calculated. Default is ``None``.
        bounds (tuple[float, float]): Lower and upper bounds of the
            windows relative to centers. If ``None`` the area under the
            whole curve is calculated. Default is ``None``.
        transformation (Callable | None): If not ``None``, a transformation
            to apply to the singal before taking the area. Shoule always
            return an array of the same shape and size. Default is ``None``.

    Returns
    np.ndarray: an array of areas of length n_trials
    '''
    # window signals if necessary
    if centers is None or bounds is None:
        y = self.X
        x = self.ts
    else:
        windows = self.window(centers=centers, bounds=bounds)
        y = windows.X
        x = windows.ts

    # transform signals if specified
    if transformation is not None:
        y = transformation(y)

    return np.trapezoid(y=y, x=x, axis=1)

get_layer(key)

Get a layer from the underlying AnnData object.

Parameters:

• key (str) –

  Layer name.

Returns:

• ndarray –

  np.ndarray: Requested layer data.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def get_layer(self, key: str) -> np.ndarray:
    """Get a layer from the underlying AnnData object.

    Args:
        key (str): Layer name.

    Returns:
        np.ndarray: Requested layer data.
    """
    return cast(np.ndarray, self.adata.layers[key])

filter_rows(mask, inplace=False)

Filter rows (trials) using a boolean mask.

Parameters:

• mask (ndarray) –

  Boolean array of length n_trials.

• inplace (bool, default: False ) –

  If True, modify in place. If False, return a new object. Defaults to False.

Returns:

• None | Self –

  None | PhotometryData: New filtered object if inplace is False, else None.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def filter_rows(self, mask: np.ndarray, inplace: bool = False) -> None | Self:
    """Filter rows (trials) using a boolean mask.

    Args:
        mask (np.ndarray): Boolean array of length n_trials.
        inplace (bool, optional): If ``True``, modify in place. If
            ``False``, return a new object. Defaults to ``False``.

    Returns:
        None | PhotometryData: New filtered object if ``inplace`` is
            ``False``, else ``None``.
    """
    if inplace:
        self.adata = self.adata[mask, :].copy()
    else:
        return type(self)(self.adata[mask, :].copy())

add_obs_columns(add_from, keys=None)

Add columns to obs from a dictionary.

Parameters:

• add_from (dict[str, Any]) –

  Mapping from column names to values.

• keys (list[str] | None, default: None ) –

  Keys from add_from to add. Defaults to all keys.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def add_obs_columns(self, add_from: dict[str, Any], keys: list[str] | None = None) -> None:
    """Add columns to `obs` from a dictionary.

    Args:
        add_from (dict[str, Any]): Mapping from column names to values.
        keys (list[str] | None, optional): Keys from ``add_from`` to add.
            Defaults to all keys.

    Returns:
        None
    """
    keys = list(add_from.keys()) if keys is None else keys
    for k in keys:
        self.adata.obs[k] = add_from.get(k, None)

add_metadata(add_from, keys=None)

Add entries to the .uns metadata dictionary.

Parameters:

• add_from (dict[str, Any]) –

  Mapping from keys to metadata values.

• keys (list[str] | None, default: None ) –

  Keys from add_from to add. Defaults to all keys.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def add_metadata(self, add_from: dict[str, Any], keys: list[str] | None = None) -> None:
    """Add entries to the `.uns` metadata dictionary.

    Args:
        add_from (dict[str, Any]): Mapping from keys to metadata values.
        keys (list[str] | None, optional): Keys from ``add_from`` to add.
            Defaults to all keys.

    Returns:
        None
    """
    keys = list(add_from.keys()) if keys is None else keys
    for k in keys:
        self.adata.uns[k] = add_from.get(k, None)

drop_obs_columns(to_drop)

Drop observation columns from obs.

Parameters:

• to_drop (list[str]) –

  Column names to drop.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def drop_obs_columns(self, to_drop: list[str]) -> None:
    """Drop observation columns from `obs`.

    Args:
        to_drop (list[str]): Column names to drop.

    Returns:
        None
    """
    self.adata.obs = self.obs.drop(to_drop, errors='ignore', axis=1)

get_text_value_counts(col)

Get a string summary of value counts for a column in obs.

Parameters:

• col (str) –

  Column name in obs.

Returns:

• str ( str ) –

  Comma-separated summary of value counts.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def get_text_value_counts(self, col: str) -> str:
    """Get a string summary of value counts for a column in `obs`.

    Args:
        col (str): Column name in ``obs``.

    Returns:
        str: Comma-separated summary of value counts.
    """
    vc = self.adata.obs[col].value_counts(dropna=False)
    return ", ".join(f"{k}: {v}" for k, v in vc.items())

trials_to_long_df(layer=None, err_layer=None, obs_cols=None, downsample=10)

Translate trial data to long DataFrame format.

Mostly useful for graphing.

Parameters:

• layer (str | None, default: None ) –

  Which layer to export. If None, self.X is used. Defaults to None.

• err_layer (str | None, default: None ) –

  Which layer, if any, to add as an error column. Defaults to None.

• obs_cols (list[str] | None, default: None ) –

  Which columns from obs to include. Defaults to None.

• downsample (int | None, default: 10 ) –

  How much to downsample the time series data, if at all. Defaults to 10.

Returns:

• DataFrame –

  pd.DataFrame: DataFrame containing trial_id, selected obs_cols, time_idx, signal values, time, and err_layer in long format.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def trials_to_long_df(
        self, 
        layer: str | None = None,
        err_layer: str | None = None,
        obs_cols: list[str] | None = None, 
        downsample: int | None = 10,
        ) -> pd.DataFrame:
    """Translate trial data to long DataFrame format.

    Mostly useful for graphing.

    Args:
        layer (str | None, optional): Which layer to export. If ``None``,
            ``self.X`` is used. Defaults to ``None``.
        err_layer (str | None, optional): Which layer, if any, to add as
            an error column. Defaults to ``None``.
        obs_cols (list[str] | None, optional): Which columns from ``obs``
            to include. Defaults to ``None``.
        downsample (int | None, optional): How much to downsample the time
            series data, if at all. Defaults to 10.

    Returns:
        pd.DataFrame: DataFrame containing ``trial_id``, selected
            ``obs_cols``, ``time_idx``, signal values, time, and
            ``err_layer`` in long format.
    """
    obj = self if downsample is None else self.downsample(downsample)
    X = obj.adata.layers[layer] if layer is not None else obj.adata.X

    def _make_long_fast(arr: np.ndarray, obj: PhotometryData, obs_cols: list[str] | None = None, value_name: str = 'signal') -> pd.DataFrame:
            n_trials, n_time = arr.shape

            trial_ids = obj.adata.obs_names.to_numpy()

            out = {
                "trial_id": np.repeat(trial_ids, n_time),
                "time_idx": np.tile(np.arange(n_time), n_trials),
                value_name: np.asarray(arr).reshape(-1),
            }

            if obs_cols is not None:
                obs_df = obj.adata.obs[obs_cols]
                for col in obs_cols:
                    out[col] = np.repeat(obs_df[col].to_numpy(), n_time)

            return pd.DataFrame(out)    

    long_signal = _make_long_fast(X, obj=obj, obs_cols=obs_cols)

    long_signal["time_idx"] = long_signal["time_idx"].astype(int)
    long_signal["time"] = obj.ts[long_signal["time_idx"]]

    if err_layer is not None:
        E = obj.adata.layers[err_layer]
        long_err = _make_long_fast(E, obj=obj, obs_cols=None, value_name=err_layer)
        long_signal = long_signal.join(long_err.filter([err_layer]), how='left')

    return long_signal

trials_to_wide_df(layer=None, obs_cols=None, signal_prefix='photometry', downsample=10)

Translate trial data to wide DataFrame format.

Mostly useful for exporting to analysis.FMM module.

Parameters:

• layer (str | None, default: None ) –

  Which layer to export. If None, self.X is used. Defaults to None.

• obs_cols (list[str] | None, default: None ) –

  Which columns from obs to include. Defaults to None, which includes all.

• downsample (int | optional, default: 10 ) –

  How much to downsample the time series data, if at all. Defaults to 10.

Returns:

• DataFrame –

  pd.DataFrame: DataFrame containing obs_cols and signal timepoints

• DataFrame –

  in columns signal_prefix.1 ... signal_prefix.n_times.

Source code in pyFiberPhotometry/core/PhotometeryData.py

def trials_to_wide_df(
        self,
        layer: str | None = None,
        obs_cols: list[str] | None = None,
        signal_prefix: str = 'photometry',
        downsample: int = 10,
        ) -> pd.DataFrame:
    '''Translate trial data to wide DataFrame format.

    Mostly useful for exporting to ``analysis.FMM`` module.

    Args:
        layer (str | None, optional): Which layer to export. If ``None``,
            ``self.X`` is used. Defaults to ``None``.
        obs_cols (list[str] | None, optional): Which columns from ``obs``
            to include. Defaults to ``None``, which includes all.
        downsample (int | optional): How much to downsample the time
            series data, if at all. Defaults to ``10``.

    Returns:
        pd.DataFrame: DataFrame containing ``obs_cols`` and signal timepoints
        in columns ``signal_prefix``.1 ... ``signal_prefix``.n_times.
    '''
    if layer is None:
        signal = downsample_ndarray(self.X, downsample, axis=1)
    else:
        signal = downsample_ndarray(self.get_layer(layer), downsample, axis=1)

    sig_columns = signal_prefix + '.' + np.arange(1, signal.shape[1] + 1).astype(str)
    sig_df = pd.DataFrame(signal, columns=sig_columns)

    obs_df = self.obs.copy() if obs_cols is None else self.obs.filter(obs_cols).copy()

    sig_df.reset_index(drop=True, inplace=True)
    obs_df.reset_index(drop=True, inplace=True)

    return pd.concat([obs_df, sig_df], axis=1)

plot_line(i, ax=None, label_with=None, err_layer=None, plt_kwargs={})

Plot a single trial time series with an optional error band.

Parameters:

• i (int) –

  Trial index to plot.

• ax (Axes | None, default: None ) –

  Axis to plot on. Creates a new one if None. Defaults to None.

• label_with (list[str] | None, default: None ) –

  obs columns used to build the legend label. Defaults to None.

• err_layer (str | None, default: None ) –

  Optional layer name providing per-timepoint error. Defaults to None.

• plt_kwargs (dict, default: {} ) –

  Additional keyword arguments passed to plt.plot. Defaults to {}.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def plot_line(
        self, i: int, 
        ax: matplotlib.axes.Axes | None = None, 
        label_with: list[str] | None = None, 
        err_layer: str | None = None,
        plt_kwargs: dict = {},
    ) -> None:
    """Plot a single trial time series with an optional error band.

    Args:
        i (int): Trial index to plot.
        ax (matplotlib.axes.Axes | None, optional): Axis to plot on.
            Creates a new one if ``None``. Defaults to ``None``.
        label_with (list[str] | None, optional): ``obs`` columns used to
            build the legend label. Defaults to ``None``.
        err_layer (str | None, optional): Optional layer name providing
            per-timepoint error. Defaults to ``None``.
        plt_kwargs (dict, optional): Additional keyword arguments passed to
            ``plt.plot``. Defaults to ``{}``.

    Returns:
        None
    """
    if ax is None: fig, ax = plt.subplots()

    x = self.var['t'].to_numpy()
    y = self.X[i, :]
    row = self.obs.iloc[i]

    if label_with is not None:
        vals = row[label_with].astype(str).to_list()
        name_val_pairs = [k + ': ' + v for k, v in zip(label_with, vals)]
        label = ', '.join(name_val_pairs)
    else:
        label = None

    ax.plot(x, y, label=label, **plt_kwargs)
    if err_layer is not None:
        yerr = self.adata.layers[err_layer][i]
        ax.fill_between(x, y + yerr, y - yerr, alpha=0.3)

plot_set(subset, ax=None, title=None, label_with=None, err_layer=None, plt_kwargs={})

Plot a set of trials given a boolean or index subset.

Parameters:

• subset (list[bool | int]) –

  Boolean mask or indices selecting trials.

• ax (Axes, default: None ) –

  Axis to plot on. Creates a new one if None. Defaults to None.

• title (str, default: None ) –

  Optional title for the axis. Defaults to None.

• label_with (list[str], default: None ) –

  obs columns used to build legend labels. Defaults to None.

• err_layer (str, default: None ) –

  Optional layer name providing per-timepoint error. Defaults to None.

• plt_kwargs (dict, default: {} ) –

  Additional keyword arguments passed to plot_line. Defaults to {}.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def plot_set(
        self, 
        subset: list[bool | int], 
        ax: matplotlib.axes.Axes = None,
        title: str = None, 
        label_with: list[str] = None, 
        err_layer: str = None, 
        plt_kwargs: dict = {},
        ) -> None:
    """Plot a set of trials given a boolean or index subset.

    Args:
        subset (list[bool | int]): Boolean mask or indices selecting trials.
        ax (matplotlib.axes.Axes, optional): Axis to plot on. Creates a
            new one if ``None``. Defaults to ``None``.
        title (str, optional): Optional title for the axis. Defaults to
            ``None``.
        label_with (list[str], optional): ``obs`` columns used to build
            legend labels. Defaults to ``None``.
        err_layer (str, optional): Optional layer name providing
            per-timepoint error. Defaults to ``None``.
        plt_kwargs (dict, optional): Additional keyword arguments passed to
            `plot_line`. Defaults to ``{}``.

    Returns:
        None
    """
    idxs = np.arange(self.n_trials)[subset]
    if ax is None: fig, ax = plt.subplots()
    for i in idxs:
        self.plot_line(i, ax=ax, label_with=label_with, err_layer=err_layer, plt_kwargs=plt_kwargs)
    if label_with is not None: plt.legend()
    if title is not None: ax.set_title(title)

plot_groups(group_on, label_with=None, err_layer=None, save_dir=None, save_ext='.png', save_dpi=140, plt_kwargs={}, ax=None)

Plot trials grouped by observation columns.

Parameters:

• group_on (list[str]) –

  Obs columns used to define groups.

• label_with (list[str] | None, default: None ) –

  obs columns used to build legend labels. Defaults to None.

• err_layer (str | None, default: None ) –

  Optional layer name providing per-timepoint error. Defaults to None.

• save_dir (str | None, default: None ) –

  Optional output directory to save figures to. Defaults to None.

• save_ext (str, default: '.png' ) –

  File extension used when saving images. Defaults to '.png'.

• save_dpi (int, default: 140 ) –

  DPI used when saving images. Defaults to 140.

• plt_kwargs (dict, default: {} ) –

  Additional keyword arguments passed to plot_set. Defaults to {}.

• ax (optional, default: None ) –

  Axis to plot on. Defaults to None.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def plot_groups(
        self, 
        group_on: list[str], 
        label_with: list[str] | None = None, 
        err_layer: str | None = None, 
        save_dir: str | None = None,
        save_ext: str = '.png',
        save_dpi: int = 140, 
        plt_kwargs: dict = {},
        ax = None,
        ) -> None:
    """Plot trials grouped by observation columns.

    Args:
        group_on (list[str]): Obs columns used to define groups.
        label_with (list[str] | None, optional): ``obs`` columns used to
            build legend labels. Defaults to ``None``.
        err_layer (str | None, optional): Optional layer name providing
            per-timepoint error. Defaults to ``None``.
        save_dir (str | None, optional): Optional output directory to save
            figures to. Defaults to ``None``.
        save_ext (str, optional): File extension used when saving images.
            Defaults to ``'.png'``.
        save_dpi (int, optional): DPI used when saving images. Defaults to
            ``140``.
        plt_kwargs (dict, optional): Additional keyword arguments passed to
            `plot_set`. Defaults to ``{}``.
        ax (optional): Axis to plot on. Defaults to ``None``.

    Returns:
        None
    """
    groups = self.obs.groupby(group_on).indices
    for gkey, idxs in groups.items():
        if not isinstance(gkey, tuple): gkey=[gkey]
        title = ', '.join([f'{name}: {val}' for name, val in zip(group_on, gkey)])
        self.plot_set(subset=idxs, label_with=label_with, title=title, err_layer=err_layer, plt_kwargs=plt_kwargs, ax=ax)
        if save_dir is not None:
            file_name = '_'.join([f'{name}-{val}' for name, val in zip(group_on, gkey)]) + save_ext
            plt.savefig(os.path.join(save_dir, file_name), dpi=save_dpi)
        plt.show()

plot_all(label_with=None, err_layer=None, plt_kwargs={}, ax=None)

Plot all trials.

Parameters:

• label_with (list[str] | None, default: None ) –

  obs columns used to build legend labels. Defaults to None.

• err_layer (str | None, default: None ) –

  Optional layer name providing per-timepoint error. Defaults to None.

• plt_kwargs (dict, default: {} ) –

  Additional keyword arguments passed to plot_set. Defaults to {}.

• ax (optional, default: None ) –

  Axis to plot on. Defaults to None.

Returns:

• None –

  None

Source code in pyFiberPhotometry/core/PhotometeryData.py

def plot_all(
        self,
        label_with: list[str] | None = None, 
        err_layer: str | None = None, 
        plt_kwargs: dict = {},
        ax = None,
        ) -> None:
    """Plot all trials.

    Args:
        label_with (list[str] | None, optional): ``obs`` columns used to
            build legend labels. Defaults to ``None``.
        err_layer (str | None, optional): Optional layer name providing
            per-timepoint error. Defaults to ``None``.
        plt_kwargs (dict, optional): Additional keyword arguments passed to
            `plot_set`. Defaults to ``{}``.
        ax (optional): Axis to plot on. Defaults to ``None``.

    Returns:
        None
    """
    idxs = np.arange(self.n_trials, dtype=int)
    self.plot_set(subset=idxs, label_with=label_with, err_layer=err_layer, plt_kwargs=plt_kwargs, ax=ax)
    plt.show()