import json
import warnings
import numpy as np
from pydantic import FilePath, validate_call
from pynwb import NWBFile
from pynwb.behavior import CompassDirection, Position, SpatialSeries
from .nvt_utils import read_data, read_header
from ....basetemporalalignmentinterface import BaseTemporalAlignmentInterface
from ....utils import DeepDict, get_base_schema
from ....utils.json_schema import _NWBMetaDataEncoder
from ....utils.path import infer_path
[docs]
class NeuralynxNvtInterface(BaseTemporalAlignmentInterface):
"""Data interface for Neuralynx NVT files. NVT files store position tracking information."""
display_name = "Neuralynx NVT"
keywords = ("position tracking",)
associated_suffixes = (".nvt",)
info = "Interface for writing Neuralynx position tracking .nvt files to NWB."
@validate_call
def __init__(
self, file_path: FilePath, *args, verbose: bool = False
): # TODO: change to * (keyword only) on or after August 2026
"""
Interface for writing Neuralynx .nvt files to nwb.
Parameters
----------
file_path : FilePath
Path to the .nvt file
verbose : bool, default: False
controls verbosity.
"""
# Handle deprecated positional arguments
if args:
parameter_names = [
"verbose",
]
num_positional_args_before_args = 1 # file_path
if len(args) > len(parameter_names):
raise TypeError(
f"__init__() takes at most {len(parameter_names) + num_positional_args_before_args + 1} positional arguments but "
f"{len(args) + num_positional_args_before_args + 1} were given. "
"Note: Positional arguments are deprecated and will be removed on or after August 2026. "
"Please use keyword arguments."
)
positional_values = dict(zip(parameter_names, args))
passed_as_positional = list(positional_values.keys())
warnings.warn(
f"Passing arguments positionally to NeuralynxNvtInterface.__init__() is deprecated "
f"and will be removed on or after August 2026. "
f"The following arguments were passed positionally: {passed_as_positional}. "
"Please use keyword arguments instead.",
FutureWarning,
stacklevel=2,
)
verbose = positional_values.get("verbose", verbose)
self.file_path = file_path
self.verbose = verbose
self._timestamps = self.get_original_timestamps()
self.header = read_header(self.file_path)
self.nvt_filename = infer_path(self.header["OriginalFileName"]).name
super().__init__(file_path=file_path)
[docs]
def get_original_timestamps(self) -> np.ndarray:
data = read_data(self.file_path)
times = data["TimeStamp"] / 1000000 # Neuralynx stores times in microseconds
times = times - times[0]
return times
[docs]
def get_timestamps(self) -> np.ndarray:
return self._timestamps
[docs]
def set_aligned_timestamps(self, aligned_timestamps: np.ndarray) -> None:
self._timestamps = aligned_timestamps
[docs]
def add_to_nwbfile(
self,
nwbfile: NWBFile,
metadata: dict | None = None,
*args, # TODO: change to * (keyword only) on or after August 2026
add_position: bool = True,
add_angle: bool | None = None,
):
"""
Add NVT data to a given in-memory NWB file
Parameters
----------
nwbfile : NWBFile
nwb file to which the recording information is to be added
metadata : dict, optional
metadata info for constructing the nwb file.
add_position : bool, default=True
add_angle : bool, optional
If None, write angle as long as it is not all 0s
"""
# Handle deprecated positional arguments
if args:
parameter_names = [
"add_position",
"add_angle",
]
num_positional_args_before_args = 2 # nwbfile, metadata
if len(args) > len(parameter_names):
raise TypeError(
f"add_to_nwbfile() takes at most {len(parameter_names) + num_positional_args_before_args} positional arguments but "
f"{len(args) + num_positional_args_before_args} were given. "
"Note: Positional arguments are deprecated and will be removed on or after August 2026. "
"Please use keyword arguments."
)
positional_values = dict(zip(parameter_names, args))
passed_as_positional = list(positional_values.keys())
warnings.warn(
f"Passing arguments positionally to NeuralynxNVTInterface.add_to_nwbfile() is deprecated "
f"and will be removed on or after August 2026. "
f"The following arguments were passed positionally: {passed_as_positional}. "
"Please use keyword arguments instead.",
FutureWarning,
stacklevel=2,
)
add_position = positional_values.get("add_position", add_position)
add_angle = positional_values.get("add_angle", add_angle)
metadata = metadata or self.get_metadata()
if isinstance(metadata, DeepDict):
metadata = metadata.to_dict()
data = read_data(self.file_path)
if add_position:
# convert to float and change <= 0 (null) to NaN
xi = data["Xloc"]
x = xi.astype(float)
x[xi <= 0] = np.nan
yi = data["Yloc"]
y = yi.astype(float)
y[yi <= 0] = np.nan
spatial_series = SpatialSeries(
name=metadata["Behavior"][self.nvt_filename]["position_name"],
data=np.c_[x, y],
reference_frame=metadata["Behavior"][self.nvt_filename]["position_reference_frame"],
unit="pixels",
conversion=1.0,
timestamps=self.get_timestamps(),
description=f"Pixel x and y coordinates from the .nvt file with header data: {json.dumps(self.header, cls=_NWBMetaDataEncoder)}",
)
nwbfile.add_acquisition(Position([spatial_series], name="NvtPosition"))
if add_angle or (add_angle is None and not np.all(data["Angle"] == 0)):
nwbfile.add_acquisition(
CompassDirection(
SpatialSeries(
name=metadata["Behavior"][self.nvt_filename]["angle_name"],
data=data["Angle"],
reference_frame=metadata["Behavior"][self.nvt_filename]["angle_reference_frame"],
unit="degrees",
conversion=1.0,
timestamps=spatial_series if add_position else self.get_timestamps(),
description=f"Angle from the .nvt file with header data: {json.dumps(self.header, cls=_NWBMetaDataEncoder)}",
),
name="NvtCompassDirection",
)
)