NeuroChaT Examples¶
Try NeuroChaT with OSF data¶
"""Downloads an NWB file from neurochat's OSF."""
import urllib.request
import os
from neurochat.nc_data import NData
def main(url, file_name, verbose=False):
"""
This demonstrates example hdf5 usage.
url should be the download url of a hdf5 file.
file_name is a local disk path to store that file in.
if verbose is true, information about the hdf5 file is printed.
"""
# Fetch a file from OSF if not available on disk
if not os.path.exists(file_name):
print("Downloading file from {} to {}".format(
url, file_name))
urllib.request.urlretrieve(url, file_name)
else:
print("Using {}".format(file_name))
if verbose:
from skm_pyutils.py_print import print_h5
print_h5(file_name)
# Set up the h5 paths
spike_path = "/processing/Shank/7"
pos_path = "/processing/Behavioural/Position"
lfp_path = "/processing/Neural Continuous/LFP/eeg"
# HDF requires filename + path_in_hdf5
# This function just does that
def to_hdf_path(x):
return file_name + "+" + x
# Load in that data
ndata = NData()
ndata.set_data_format("NWB")
ndata.set_spatial_file(to_hdf_path(pos_path))
ndata.set_spike_file(to_hdf_path(spike_path))
ndata.set_lfp_file(to_hdf_path(lfp_path))
ndata.load()
# Choose the unit number from those available
print("Units are:", ndata.get_unit_list())
unit_no = int(input("Unit to use:\n").strip())
ndata.set_unit_no(unit_no)
print("Loaded:", ndata)
# Perform analysis
ndata.place()
ndata.wave_property()
# print(ndata.get_results()["Spatial Skaggs"])
# print(ndata.get_results()["Mean Spiking Freq"])
print(ndata.get_results(spaces_to_underscores=True))
print(ndata.get_results(spaces_to_underscores=False))
if __name__ == "__main__":
home = os.path.abspath(os.path.expanduser("~"))
file_name = os.path.join(home, "neurochat_temp", "example.hdf")
if not os.path.isdir(os.path.dirname(file_name)):
inp = input("Will create directory {}, is this ok? (y/n) ".format(
os.path.dirname(file_name))).lower().strip()
if inp != "y":
"Ok, quitting"
exit(0)
else:
os.makedirs(os.path.dirname(file_name), exist_ok=True)
url = 'https://osf.io/89t7g/download/'
verbose = False
main(url, file_name, verbose=verbose)
Load data into NeuroChaT¶
"""Can set up this file to provide example data for tests."""
import os
from neurochat.nc_data import NData
def load_data():
dir = r'C:\Users\smartin5\recording_example'
spike_file = os.path.join(dir, "010416b-LS3-50Hz10V5ms.2")
pos_file = os.path.join(dir, "010416b-LS3-50Hz10V5ms_2.txt")
lfp_file = os.path.join(dir, "010416b-LS3-50Hz10V5ms.eeg")
unit_no = 7
ndata = NData()
ndata.set_spike_file(spike_file)
ndata.set_spatial_file(pos_file)
ndata.set_lfp_file(lfp_file)
ndata.load()
ndata.set_unit_no(unit_no)
return ndata
if __name__ == "__main__":
load_data()
Run NeuroChaT with a configuration file¶
import os
import logging
from neurochat.nc_config import Configuration
from neurochat.nc_control import NeuroChaT
def main(config_loc):
nc = NeuroChaT()
config = Configuration()
config.set_config_file(config_loc)
config.load_config()
nc.set_configuration(config)
nc.run()
excel_file = nc._pdf_file[:-3] + "xlsx"
try:
results = nc.get_results()
results.to_excel(excel_file)
logging.info("Analysis results exported to: " +
excel_file.rstrip("\n\r").split(os.sep)[-1])
except:
logging.error('Failed to export results!')
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
mpl_logger = logging.getLogger("matplotlib")
mpl_logger.setLevel(level=logging.WARNING)
# Set up path to the ncfg file
root = r"C:\Users\smartin5\Neuroscience\NC_cfgs"
name = "all_spat.ncfg"
config_loc = os.path.join(root, name)
main(config_loc)
Convert a SpikeInterface SortingExtractor into NWB¶
import os
import matplotlib.pyplot as plt
from neurochat.nc_spike import NSpike
from neurochat.nc_hdf import Nhdf
import neurochat.nc_plot as nc_plot
import logging
def spikeinterface_test(folder_name):
"""An example sorting extractor, which loads from Phy."""
import spikeinterface.extractors as se
to_exclude = ["mua", "noise"]
sorting = se.PhySortingExtractor(
folder_name, exclude_cluster_groups=to_exclude, load_waveforms=True,
verbose=False)
return sorting
def read_hdf(hdf_path, verbose=False, group=3):
"""
Read the NWB at hdf_path into NeuroChaT.
Parameters
----------
hdf_path : str
Path to the hdf5 file
verbose : bool, optional.
Defaults to False, indicates whether to print information.
group : int, optional.
Defaults to 3, indicates the group in the hdf5 file to use.
Returns
-------
NSpike
The loaded NSpike object.
"""
if verbose:
from skm_pyutils.py_print import print_h5
print_h5(hdf_path)
spike_file = hdf_path + "+/processing/Shank/" + str(group)
spike = NSpike()
spike.set_system("NWB")
spike.set_filename(spike_file)
spike.load()
unit_no = spike.get_unit_list()[0]
spike.set_unit_no(unit_no)
if verbose:
print(spike)
return spike
if __name__ == '__main__':
"""Can set whether to write or read hdf5 here, and also the paths."""
to_write = True
to_read = True
logging.basicConfig(level=logging.INFO)
mpl_logger = logging.getLogger("matplotlib")
mpl_logger.setLevel(level=logging.WARNING)
if to_write:
write_name = r"D:\Ham_Data\Batch_2\A9_CAR-SA1\CAR-SA1_20191130_1_PreBox\phy_klusta"
from neurochat.nc_control import NeuroChaT
plot_waveforms = False
sorting = spikeinterface_test(write_name)
NeuroChaT.sortingextractor_to_nwb(
sorting, plot_waveforms=plot_waveforms)
if to_read:
read_name = r"D:\Ham_Data\Batch_2\A9_CAR-SA1\CAR-SA1_20191130_1_PreBox\CAR-SA1_2019-11-30_PreBox_shuff_NC_NWB.hdf5"
spike = read_hdf(read_name)
print(spike)
Plot the waveform of a cell¶
"""Simple example to plot the waveform of a cell."""
from load_example_data import load_data
import matplotlib.pyplot as plt
import neurochat.nc_plot as nc_plot
def main():
ndata = load_data()
results = ndata.wave_property()
nc_plot.wave_property(results)
plt.savefig("out.png")
print(ndata.get_results())
if __name__ == "__main__":
main()
Measure cluster similarity¶
"""Compare a set of clusters from two Axona recordings."""
import os
from neurochat.nc_datacontainer import NDataContainer
import neurochat.nc_containeranalysis as nca
from neurochat.nc_clust import NClust
from neurochat.nc_utils import log_exception
import numpy as np
def compare_two(file1, file2, index):
"""
Compare spike file1 against spike file2.
Cluster similarity is calculated.
index is used for file naming, and saves to:
output1.csv if index is 1, for example.
"""
nclust1 = NClust()
nclust2 = NClust()
nclust1.load(file1, "Axona")
nclust2.load(file2, "Axona")
units1 = nclust1.get_unit_list()
units2 = nclust2.get_unit_list()
bc_matrix = np.zeros(shape=(len(units1), len(units2)), dtype=np.float32)
hd_matrix = np.zeros(shape=(len(units1), len(units2)), dtype=np.float32)
for i, unit1 in enumerate(units1):
for j, unit2 in enumerate(units2):
try:
bc, hd = nclust1.cluster_similarity(nclust2, unit1, unit2)
print("({}, {}): BC {:.2f}, HD {:.2f}".format(
i + 1, j + 1, bc, hd))
bc_matrix[i, j] = bc
hd_matrix[i, j] = hd
except Exception as e:
log_exception(e, "at ({}, {})".format(i, j))
bc_matrix[i, j] = np.nan
hd_matrix[i, j] = np.nan
out_filename = "output{}.csv".format(index)
with open(out_filename, "w") as f:
# Save the BC
# for i in range(len(units1)):
# out_str = ""
# for j in range(len(units2)):
# out_str += str(bc_matrix[i, j]) + ","
# out_str = out_str[:-1] + "\n"
# f.write(out_str)
# Save the HD
for i in range(len(units1)):
out_str = ""
for j in range(len(units2)):
out_str += str(hd_matrix[i, j]) + ","
out_str = out_str[:-1] + "\n"
f.write(out_str)
if __name__ == "__main__":
dir = r'C:\Users\smartin5\Recordings\recording_example'
file1 = os.path.join(dir, "010416b-LS3-50Hz10V5ms.1")
file2 = os.path.join(dir, "010416b-LS3-50Hz10V5ms.2")
compare_two(file1, file2, 1)
file1 = os.path.join(dir, "010416b-LS3-50Hz10V5ms.1")
file2 = os.path.join(dir, "010416b-LS3-50Hz10V5ms.2")
compare_two(file1, file2, 2)