161 lines
5.2 KiB
Python
161 lines
5.2 KiB
Python
import logging
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
|
|
import nixio
|
|
import rlxnix as rlx
|
|
import typer
|
|
from IPython import embed
|
|
from rich.console import Console
|
|
|
|
from oephys2nix.logging import setup_logging
|
|
from oephys2nix.stimulus_recreation import StimulusToNix
|
|
from oephys2nix.tonix import RawToNix
|
|
|
|
app = typer.Typer()
|
|
log = logging.getLogger(__name__)
|
|
console = Console()
|
|
|
|
|
|
@app.command()
|
|
def convert(
|
|
data_path: Path = typer.Argument(
|
|
...,
|
|
help="The source directory containing the Open Ephys data.",
|
|
exists=True,
|
|
file_okay=False,
|
|
dir_okay=True,
|
|
readable=True,
|
|
resolve_path=True,
|
|
),
|
|
no_ttl: bool = typer.Option(False, help="For recordings that did not have a ttl pulse"),
|
|
overwrite: bool = typer.Option(default=True, help="Overwrites nix file"),
|
|
debug: bool = typer.Option(default=True, help="Shows more information and plots the results"),
|
|
verbose: Annotated[int, typer.Option("--verbose", "-v", count=True)] = 0,
|
|
) -> None:
|
|
"""Combines open ephys data with relacs data from data_path to a new nix file."""
|
|
|
|
setup_logging(logging.getLogger("oephys2nix"), verbosity=verbose)
|
|
log.info(f"Selected data_path is {data_path}")
|
|
open_ephys_data_paths = list(Path(data_path).rglob("*open-ephys"))
|
|
if not open_ephys_data_paths:
|
|
log.error("Did not find any open-ephys data")
|
|
raise typer.Exit()
|
|
|
|
relacs_data_paths = list(Path(data_path).rglob("*relacs/*.nix"))
|
|
if not relacs_data_paths:
|
|
log.error("Did not find any relacs data")
|
|
raise typer.Exit()
|
|
|
|
if len(open_ephys_data_paths) != len(relacs_data_paths):
|
|
log.error(
|
|
f"Missmatch in data directory found open-ephys data {len(open_ephys_data_paths)} "
|
|
f"and relacs data {len(relacs_data_paths)}"
|
|
)
|
|
log.error("Please check if both are present")
|
|
raise typer.Exit()
|
|
|
|
for open_ephys, relacs in zip(open_ephys_data_paths, relacs_data_paths, strict=True):
|
|
nix_file_name = relacs.parent.name.split("_")[0] + "-recording.nix"
|
|
|
|
console.print(
|
|
f"Converting [bold cyan]{open_ephys.name}[/bold cyan] and "
|
|
f"[bold magenta]{relacs.name}[/bold magenta] -> "
|
|
f"[green]{nix_file_name}[/green]"
|
|
)
|
|
|
|
nix_path = relacs.parent.parent / nix_file_name
|
|
|
|
if nix_path.exists and nix_path.is_file():
|
|
if overwrite:
|
|
log.warning("Overwriting nix file")
|
|
nix_file = nixio.File(str(nix_path), nixio.FileMode.Overwrite)
|
|
else:
|
|
log.error("Converted nix file already exits, and Overwrite is not enabled")
|
|
raise typer.Exit()
|
|
else:
|
|
log.debug("Creating nix file")
|
|
nix_file = nixio.File(str(nix_path), nixio.FileMode.Overwrite)
|
|
|
|
nix_file.close()
|
|
|
|
log.info("Appending raw data from relacs and open-ephys to new nix-file")
|
|
to_nix = RawToNix(open_ephys, str(relacs), str(nix_path))
|
|
to_nix.append_section()
|
|
to_nix.append_fish_lines()
|
|
to_nix.append_relacs_lines()
|
|
to_nix.append_raw_data()
|
|
to_nix.close()
|
|
log.info("Finished!")
|
|
|
|
log.info("Starting with stimulus recreation")
|
|
stim = StimulusToNix(open_ephys, str(relacs), str(nix_path))
|
|
stim.create_repros_automatically()
|
|
stim.print_table()
|
|
|
|
|
|
@app.command()
|
|
def timeline(
|
|
data_path: Path = typer.Argument(
|
|
...,
|
|
help="The source directory containing a dataset.",
|
|
exists=True,
|
|
file_okay=False,
|
|
dir_okay=True,
|
|
readable=True,
|
|
resolve_path=True,
|
|
),
|
|
) -> None:
|
|
"""Plot the timeline for a given dataset."""
|
|
dataset = list(Path(data_path).rglob("*.nix"))
|
|
|
|
if not dataset:
|
|
log.error("Did not find any dataset")
|
|
raise typer.Exit()
|
|
if len(dataset) > 1:
|
|
log.info(f"Found multiple datasets {len(dataset)} taking the first one")
|
|
dataset = rlx.Dataset(str(dataset[0]))
|
|
dataset.plot_timeline()
|
|
dataset.close()
|
|
|
|
|
|
@app.command()
|
|
def plot(
|
|
data_path: Path = typer.Argument(
|
|
...,
|
|
help="The source directory containing the open-ephys and relacs data.",
|
|
exists=True,
|
|
file_okay=False,
|
|
dir_okay=True,
|
|
readable=True,
|
|
resolve_path=True,
|
|
),
|
|
) -> None:
|
|
"""Plot the stimulus, TTL and the relacs stimulus.
|
|
|
|
You can run this if you converted already the nix-file.
|
|
"""
|
|
relacs_data_paths = list(Path(data_path).rglob("*relacs/*.nix"))
|
|
if not relacs_data_paths:
|
|
log.error("Did not find any relacs data")
|
|
raise typer.Exit()
|
|
|
|
open_ephys_data_paths = list(Path(data_path).rglob("*open-ephys"))
|
|
if not open_ephys_data_paths:
|
|
log.error("Did not find any open-ephys data")
|
|
raise typer.Exit()
|
|
|
|
nix_file_name = relacs_data_paths[0].parent.name.split("_")[0] + "-recording.nix"
|
|
nix_path = relacs_data_paths[0].parent.parent / nix_file_name
|
|
if not nix_path.is_file:
|
|
log.error("Did not find any converted nix file")
|
|
raise typer.Exit()
|
|
|
|
stim = StimulusToNix(open_ephys_data_paths[0], str(relacs_data_paths[0]), str(nix_path))
|
|
stim.plot_stimulus()
|
|
stim.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app()
|