Source code for allensdk.brain_observatory.ecephys.stimulus_table.visualization.view_blocks
import argparse
import itertools as it
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
[docs]def build_colormap(table, existing_map={}, base_colors=sns.color_palette("pastel")):
colormap = {}
unique_names = table["stimulus_name"].unique()
color_iterator = iter(base_colors)
for un in unique_names:
if un in existing_map:
colormap[un] = existing_map[un]
continue
if isinstance(un, float) and np.isnan(un):
un = "spontaneous_activity"
colormap[un] = next(color_iterator)
return colormap
[docs]def get_blocks(table):
changes = np.where(np.diff(table["stimulus_block"].values))[0] + 1
changes = np.sort(np.unique(np.concatenate([changes, [0, table.shape[0]]])))
blocks = []
for ii, (low, high) in enumerate(zip(changes[:-1], changes[1:])):
block = table.iloc[low:high, :]
recorded_blocks = np.unique(block["stimulus_block"].values)
if len(recorded_blocks) > 1:
raise ValueError(
"expected one recorded block per block, found: {}".format(
recorded_blocks
)
)
else:
recorded_block = recorded_blocks[0]
start = block["Start"].values[0]
end = block["End"].values[-1]
names = np.unique(block["stimulus_name"].values)
if len(names) > 1:
raise ValueError("expected one name per block, found: {}".format(names))
else:
name = names[0]
indices = np.unique(block["stimulus_index"].values)
if len(indices) > 1:
raise ValueError("expected one index per block, found: {}".format(indices))
else:
index = indices[0]
if isinstance(name, float) and np.isnan(name):
name = "spontaneous_activity"
blocks.append({"name": name, "index": index, "start": start, "end": end})
return blocks
[docs]def plot_blocks(blocks, colormap):
fig, ax = plt.subplots(figsize=(9, 9))
used = set([])
max_time = -np.inf
handles = []
labels = []
for block in blocks:
handle = ax.axvspan(
block["start"],
block["end"],
facecolor=colormap[block["name"]],
alpha=1.0,
linestyle="-",
edgecolor="black",
)
if not block["name"] in used:
labels.append(block["name"])
handles.append(handle)
max_time = max([max_time, block["end"]])
used.add(block["name"])
ax.set_xlim([0, max_time])
ax.get_yaxis().set_visible(False)
ax.set_xlabel("time (s)")
plt.legend(handles, labels)
[docs]def main(table_csv_path):
table = pd.read_csv(table_csv_path)
colormap = build_colormap(table)
blocks = get_blocks(table)
plot_blocks(blocks, colormap)
plt.show()
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument(
"table_csv_path", type=str, help="filesystem path to stimulus table csv"
)
args = parser.parse_args()
main(args.table_csv_path)