Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 156 additions & 25 deletions src/nwb2bids/bids_models/_channels.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import pynwb
import typing_extensions

from ._model_utils import _get_present_fields
from ..bids_models._base_metadata_model import BaseMetadataContainerModel, BaseMetadataModel
from ..notifications import Notification

Expand Down Expand Up @@ -40,26 +41,149 @@ def _infer_scalar_field(


class Channel(BaseMetadataModel):
name: str
electrode_name: str
type: str = "n/a"
units: str = "V"
sampling_frequency: float | None = None
low_cutoff: float | None = None
high_cutoff: float | None = None
reference: str | None = None
notch: str | None = None
channel_label: str | None = None
stream_id: str | None = None
description: str | None = None
software_filter_types: str | None = None
status: typing.Literal["good", "bad"] | None = None
status_description: str | None = None
gain: float | None = None
time_offset: float | None = None
time_reference_channel: str | None = None
ground: str | None = None
recording_mode: str | None = None
name: str = pydantic.Field(
description="Label of the channel.",
title="Channel name",
)
electrode_name: str = pydantic.Field(
description=(
"Name of the electrode contact point. The value MUST match a name entry in the corresponding "
"*_electrodes.tsv file, linking this channel to its associated electrode contact point. "
"For channels not associated with an electrode, use n/a."
),
title="Electrode name",
)
type: str = pydantic.Field(
description=(
"Type of channel; MUST use the channel types listed below. Note that the type MUST be in upper-case."
),
title="Type",
default="n/a",
)
units: str = pydantic.Field(
description=(
"Physical unit of the value represented in this channel, for example, V for Volt, or fT/cm for femto Tesla "
"per centimeter (see Units)."
),
title="Units",
default="V",
)
sampling_frequency: float | None = pydantic.Field(
description="Sampling rate of the channel in Hz.",
title="Sampling frequency",
default=None,
)
low_cutoff: float | None = pydantic.Field(
description=(
"Frequencies used for the high-pass filter applied to the channel in Hz. If no high-pass "
"filter applied, use n/a."
),
title="Low cutoff",
default=None,
)
high_cutoff: float | None = pydantic.Field(
description=(
"Frequencies used for the low-pass filter applied to the channel in Hz. "
"If no low-pass filter applied, use n/a. Note that hardware anti-aliasing in A/D conversion of "
"all MEG/EEG/EMG electronics applies a low-pass filter; specify its frequency here if applicable."
),
title="High cutoff",
default=None,
)
reference: str | None = pydantic.Field(
description=(
"The reference for the given channel. When the reference is an electrode in *_electrodes.tsv, "
'use the name of that electrode. If a corresponding electrode is not applicable, use "n/a".'
),
title="Reference",
default=None,
)
notch: str | None = pydantic.Field(
description=(
"Frequencies used for the notch filter applied to the channel, in Hz. "
"If notch filters are applied at multiple frequencies, these frequencies MAY be specified as a list, "
"for example, [60, 120, 180]. If no notch filter was applied, use n/a."
),
title="Notch",
default=None,
)
channel_label: str | None = pydantic.Field(
description=(
"Human readable identifier. Use this name to specify the content of signals not generated by electrodes. "
"For example, 'DAQ internal synchronization signals', 'behavioral signals', 'behavioral cues'."
),
title="Channel label",
default=None,
)
stream_id: str | None = pydantic.Field(
description="Data stream of the recording the signal.",
title="Stream ID",
default=None,
)
description: str | None = pydantic.Field(
description="Brief free-text description of the channel, or other information of interest.",
title="Description",
default=None,
)
software_filter_types: str | None = pydantic.Field(
description=(
"The types of software filters applied to this channel. "
"The Levels for this column SHOULD be defined in the accompanying *_channels.json file, "
"mapping each filter type key to its description. Use n/a if no software filters were "
"applied to this channel."
),
title="Software filter types",
default=None,
)
status: typing.Literal["good", "bad", "n/a"] | None = pydantic.Field(
description=(
"Data quality observed on the channel. "
"A channel is considered bad if its data quality is compromised by excessive noise. "
"If quality is unknown, then a value of n/a may be used. Description of noise type SHOULD "
"be provided in [status_description]."
),
title="Status",
default=None,
)
status_description: str | None = pydantic.Field(
description=(
"Freeform text description of noise or artifact affecting data quality on the channel. "
"It is meant to explain why the channel was declared bad in the status column."
),
title="Status description",
default=None,
)
gain: float | None = pydantic.Field(
description=(
"Amplification factor applied from signal detection at the electrode to the signal stored in the data file."
Comment thread
CodyCBakerPhD marked this conversation as resolved.
"If no gain factor is provided it is assumed to be 1."
),
title="Gain",
default=None,
)
time_offset: float | None = pydantic.Field(
description="Time shift between signal of this channel to a reference channel in seconds.",
title="Time offset",
default=None,
)
time_reference_channel: str | None = pydantic.Field(
description="Name of the channel that is used for time alignment of signals.",
title="Time reference channel",
default=None,
)
ground: str | None = pydantic.Field(
description=(
"Information on the ground. For example, 'chamber screw', 'head post', 'ear clip'. "
"Only should be used to optionally override the global ground in the _ecephys.json or _icephys.json file."
),
title="Ground",
default=None,
)
recording_mode: str | None = pydantic.Field(
description="The mode of recording for patch clamp datasets (for example, voltage clamp, current clamp).",
title="Recording mode",
default=None,
)


class ChannelTable(BaseMetadataContainerModel):
Expand Down Expand Up @@ -246,9 +370,16 @@ def to_json(self, file_path: str | pathlib.Path) -> None:
file_path : path
The path to the output JSON file.
"""
file_path = pathlib.Path(file_path)

present_non_additional_fields = _get_present_fields(models=self.channels)

json_content: dict[str, dict[str, typing.Any]] = collections.defaultdict(dict)
for field in present_non_additional_fields:
if title := getattr(Channel.model_fields[field], "title"):
json_content[field]["LongName"] = title
if description := getattr(Channel.model_fields[field], "description"):
json_content[field]["Description"] = description

with file_path.open(mode="w") as file_stream:
json.dump(
obj=dict(), # TODO
fp=file_stream,
indent=4,
)
json.dump(obj=json_content, fp=file_stream, indent=4)
132 changes: 112 additions & 20 deletions src/nwb2bids/bids_models/_electrodes.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import collections
import json
import pathlib
import typing
Expand All @@ -8,28 +9,112 @@
import pynwb
import typing_extensions

from ._model_utils import _get_present_fields
from ..bids_models._base_metadata_model import BaseMetadataContainerModel, BaseMetadataModel
from ..notifications import Notification

_NULL_LOCATION_PLACEHOLDERS = {"", "unknown", "no location", "N/A"}


class Electrode(BaseMetadataModel):
name: str
probe_name: str
x: float = numpy.nan
y: float = numpy.nan
z: float = numpy.nan
hemisphere: str = "n/a"
impedance: float = numpy.nan # in kOhms
shank_id: str = "n/a"
size: float | None = None # in square micrometers
electrode_shape: str | None = None
material: str | None = None
location: str | None = None
pipette_solution: str | None = None
internal_pipette_diameter: float | None = None # in micrometers
external_pipette_diameter: float | None = None # in micrometers
name: str = pydantic.Field(
description="Name of the electrode contact point.",
title="Electrode name",
)
probe_name: str = pydantic.Field(
description=(
"A unique identifier of the probe, can be identical with the device_serial_number. "
"The value MUST match a probe_name entry in the corresponding *_probes.tsv file, linking this electrode "
"to its associated probe. For electrodes not associated with a probe, use n/a."
),
title="Probe name",
)
x: float = pydantic.Field(
description=(
"Recorded position along the x-axis. When no space-<label> entity is used in the filename, "
"the position along the local width-axis relative to the probe origin "
"(see coordinate_reference_point in *_probes.tsv) in micrometers (um). "
"When a space-<label> entity is used in the filename, the position relative to the origin of the "
"coordinate system along the first axis. Units are specified by MicroephysCoordinateUnits in the "
"corresponding *_coordsystem.json file."
),
title="x",
default=numpy.nan,
)
y: float = pydantic.Field(
description=(
"Recorded position along the y-axis. When no space-<label> entity is used in the filename, "
"the position along the local height-axis relative to the probe origin "
"(see coordinate_reference_point in *_probes.tsv) in micrometers (um). "
"When a space-<label> entity is used in the filename, the position relative to the origin of the "
"coordinate system along the second axis. Units are specified by MicroephysCoordinateUnits in the "
"corresponding *_coordsystem.json file."
),
title="y",
default=numpy.nan,
)
z: float = pydantic.Field(
description=(
"Recorded position along the z-axis.For 2D electrode localizations, this SHOULD be a column of n/a values. "
Comment thread
asmacdo marked this conversation as resolved.
Outdated
"When no space-<label> entity is used in the filename, the position along the local depth-axis relative to "
"the probe origin (see coordinate_reference_point in *_probes.tsv) in micrometers (um). "
"When a space-<label> entity is used in the filename, the position relative to the origin of the "
"coordinate system along the third axis. Units are specified by MicroephysCoordinateUnits in the "
"corresponding *_coordsystem.json file. For 2D electrode localizations (for example, when the "
"coordinate system is Pixels), this SHOULD be a column of n/a values."
),
title="z",
default=numpy.nan,
)
hemisphere: typing.Literal["L", "R", "n/a"] = pydantic.Field(
description="The hemisphere in which the electrode is placed.", title="Hemisphere", default="n/a"
)
impedance: float = pydantic.Field(
description="Impedance of the electrode, units MUST be in kOhm.", title="Impedance", default=numpy.nan
)
shank_id: str = pydantic.Field(
description=(
"A unique identifier to specify which shank of the probe the electrode is on. "
"This is useful for spike sorting when the electrodes are on a multi-shank probe."
),
title="Shank ID",
default="n/a",
)
size: float | None = pydantic.Field(
description="Surface area of the electrode, units MUST be in um^2.",
title="Size",
default=None,
)
electrode_shape: str | None = pydantic.Field(
description="Description of the shape of the electrode (for example, square, circle).",
title="Electrode shape",
default=None,
)
material: str | None = pydantic.Field(
description="Material of the electrode (for example, Tin, Ag/AgCl, Gold).",
title="Material",
default=None,
)
location: str | None = pydantic.Field(
description="An indication on the location of the electrode (for example, cortical layer 3, CA1).",
title="Location",
default=None,
)
pipette_solution: str | None = pydantic.Field(
description="The solution used to fill the pipette.",
title="Pipette solution",
default=None,
)
internal_pipette_diameter: float | None = pydantic.Field(
description="The internal diameter of the pipette in micrometers.",
title="Internal pipette diameter",
default=None,
)
external_pipette_diameter: float | None = pydantic.Field(
description="The external diameter of the pipette in micrometers.",
title="External pipette diameter",
default=None,
)

def __eq__(self, other: typing_extensions.Self) -> bool:
if not isinstance(other, Electrode):
Expand Down Expand Up @@ -182,9 +267,16 @@ def to_json(self, file_path: str | pathlib.Path) -> None:
file_path : path
The path to the output JSON file.
"""
file_path = pathlib.Path(file_path)

present_non_additional_fields = _get_present_fields(models=self.electrodes)

json_content: dict[str, dict[str, typing.Any]] = collections.defaultdict(dict)
for field in present_non_additional_fields:
if title := getattr(Electrode.model_fields[field], "title"):
json_content[field]["LongName"] = title
if description := getattr(Electrode.model_fields[field], "description"):
json_content[field]["Description"] = description

Comment thread
CodyCBakerPhD marked this conversation as resolved.
with file_path.open(mode="w") as file_stream:
json.dump(
obj=dict(), # TODO
fp=file_stream,
indent=4,
)
json.dump(obj=json_content, fp=file_stream, indent=4)
19 changes: 19 additions & 0 deletions src/nwb2bids/bids_models/_model_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
import typing

import pydantic
import pydantic_core


def _get_present_fields(models: typing.Sequence[pydantic.BaseModel]) -> set[str]:
"""Return a set of field names that are present or required in the model."""
present_non_additional_fields = {
field
for model in models
for field, value in model.model_dump().items()
if field in model.model_fields
and (
value is not None
or model.model_fields.get(field, pydantic.fields.FieldInfo()).default is pydantic_core.PydanticUndefined
)
}
return present_non_additional_fields
Loading
Loading