Skip to content

Pinnacle

openseize.file_io.annotations.Pinnacle

Bases: Annotations

A reader of Pinnacle Technologies© annotation csv files.

This annotation reader's 'read' method reads annotations into a list of Annotation dataclass instances. Each Annotation dataclass has the following attributes:

  • label: A string label given to an annotation.
  • time: Time, relative to recording start, in secs of an annotation.
  • duration: The duration in seconds of an annotation.
  • channel: The channel(s) an annotation was detected on.

Attributes:

Name Type Description
path

Python path instance to Pinnacle© file.
kwargs

Any valid kwarg for csv.DictReader initializer.

Examples:

>>> # read the annotations from the demo annotation file
>>> from openseize.demos import paths
>>> filepath = paths.locate('annotations_001.txt')
>>> from openseize.io.annotations import Pinnacle
>>> # read the 'rest' and 'exploring' annotations
>>> with Pinnacle(filepath, start=6) as pinnacle:
>>>     annotations = pinnacle.read(labels=['rest', 'exploring'])
>>> # get the first annotation and print it
>>> print(annotations[0])
>>> # print the first annotations duration
>>> print(annotations[0].duration)
Source code in openseize/file_io/annotations.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class Pinnacle(Annotations):
    """A reader of Pinnacle Technologies© annotation csv files.

    This annotation reader's 'read' method reads annotations into a list of
    Annotation dataclass instances. Each Annotation dataclass has the
    following attributes:

    - label: A string label given to an annotation.
    - time: Time, relative to recording start, in secs of an annotation.
    - duration: The duration in seconds of an annotation.
    - channel: The channel(s) an annotation was detected on.

    Attributes:
        path:
            Python path instance to Pinnacle© file.
        kwargs:
            Any valid kwarg for csv.DictReader initializer.

    Examples:
        >>> # read the annotations from the demo annotation file
        >>> from openseize.demos import paths
        >>> filepath = paths.locate('annotations_001.txt')
        >>> from openseize.io.annotations import Pinnacle
        >>> # read the 'rest' and 'exploring' annotations
        >>> with Pinnacle(filepath, start=6) as pinnacle:
        >>>     annotations = pinnacle.read(labels=['rest', 'exploring'])
        >>> # get the first annotation and print it
        >>> print(annotations[0])
        >>> # print the first annotations duration
        >>> print(annotations[0].duration)
    """

    def open(self,
            path: Union[str, Path],
            start: int = 0,
            delimiter: str ='\t',
            **kwargs
    ) -> Tuple[IO[str], Iterable[dict]]:
        """Opens a Pinnacle formatted CSV annotation file.

        Called by 'Annotations.__init__' to initialize this Pinnacle
        context manager.

        Args:
            path:
                A annotation file path location.
            start:
                The row number of the column headers in the file.
            delimiter:
                The string character separating columns in the file.
            **kwargs:
                Any valid keyword argument for CSV.DictReader builtin.

        Returns:
            A tuple (file_obj, DictReader) where file_obj is the open file
            instance and DictReader is the builtin csv DictReader.
        """

        # This method is called within context management (see base class)
        # pylint: disable-next=consider-using-with
        fobj = open(Path(path), encoding='utf-8')
        # advance to start row and return a reader
        # pylint: disable-next=expression-not-assigned
        [next(fobj) for _ in range(start)]
        return fobj, csv.DictReader(fobj, delimiter=delimiter, **kwargs)

    def label(self, row:  Dict[str, str]) -> str:
        """Extracts the annotation label for a row in this file."""

        return row['Annotation']

    def time(self, row: Dict[str, str]) -> float:
        """Extracts the annotation time of a row of this file."""

        return float(row['Time From Start'])

    def duration(self, row: Dict[str, str]) -> float:
        """Measures the duration of an annotation for a row in this file."""

        #compute time difference from start and stop datetime objs
        fmt = '%m/%d/%y %H:%M:%S.%f'
        start = datetime.strptime(row['Start Time'], fmt)
        stop = datetime.strptime(row['End Time'], fmt)
        return (stop - start).total_seconds()

    def channel(self, row: Dict[str, str]) -> Union[int, str]:
        """Extracts the annotation channel for a row in this file."""

        return row['Channel']

open(path, start=0, delimiter='\t', **kwargs)

Opens a Pinnacle formatted CSV annotation file.

Called by 'Annotations.init' to initialize this Pinnacle context manager.

Parameters:

Name Type Description Default
path Union[str, Path]

A annotation file path location.

 required
start int

The row number of the column headers in the file.

 0
delimiter str

The string character separating columns in the file.

 '\t'
**kwargs

Any valid keyword argument for CSV.DictReader builtin.

 {}

Returns:

Type Description
IO[str]

A tuple (file_obj, DictReader) where file_obj is the open file
Iterable[dict]

instance and DictReader is the builtin csv DictReader.
Source code in openseize/file_io/annotations.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def open(self,
        path: Union[str, Path],
        start: int = 0,
        delimiter: str ='\t',
        **kwargs
) -> Tuple[IO[str], Iterable[dict]]:
    """Opens a Pinnacle formatted CSV annotation file.

    Called by 'Annotations.__init__' to initialize this Pinnacle
    context manager.

    Args:
        path:
            A annotation file path location.
        start:
            The row number of the column headers in the file.
        delimiter:
            The string character separating columns in the file.
        **kwargs:
            Any valid keyword argument for CSV.DictReader builtin.

    Returns:
        A tuple (file_obj, DictReader) where file_obj is the open file
        instance and DictReader is the builtin csv DictReader.
    """

    # This method is called within context management (see base class)
    # pylint: disable-next=consider-using-with
    fobj = open(Path(path), encoding='utf-8')
    # advance to start row and return a reader
    # pylint: disable-next=expression-not-assigned
    [next(fobj) for _ in range(start)]
    return fobj, csv.DictReader(fobj, delimiter=delimiter, **kwargs)

label(row)

Extracts the annotation label for a row in this file.

Source code in openseize/file_io/annotations.py 
109
110
111
112
def label(self, row:  Dict[str, str]) -> str:
    """Extracts the annotation label for a row in this file."""

    return row['Annotation']

time(row)

Extracts the annotation time of a row of this file.

Source code in openseize/file_io/annotations.py 
114
115
116
117
def time(self, row: Dict[str, str]) -> float:
    """Extracts the annotation time of a row of this file."""

    return float(row['Time From Start'])

duration(row)

Measures the duration of an annotation for a row in this file.

Source code in openseize/file_io/annotations.py 
119
120
121
122
123
124
125
126
def duration(self, row: Dict[str, str]) -> float:
    """Measures the duration of an annotation for a row in this file."""

    #compute time difference from start and stop datetime objs
    fmt = '%m/%d/%y %H:%M:%S.%f'
    start = datetime.strptime(row['Start Time'], fmt)
    stop = datetime.strptime(row['End Time'], fmt)
    return (stop - start).total_seconds()

channel(row)

Extracts the annotation channel for a row in this file.

Source code in openseize/file_io/annotations.py 
128
129
130
131
def channel(self, row: Dict[str, str]) -> Union[int, str]:
    """Extracts the annotation channel for a row in this file."""

    return row['Channel']

Bases and Mixins

openseize.file_io.bases.Annotations

Bases: abc.ABC

Abstract base class for reading annotation data.

Annotation data may be stored in a variety of formats; csv files, pickled objects, etc. This ABC defines all annotation readers as context managers that read annotation files. Inheritors must override: open, label, time, duration and channel methods.

Attributes:

Name Type Description
path

Python path instance to an annotation file.
**kwargs

Any valid kwarg for concrete 'open' method.
Source code in openseize/file_io/bases.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
class Annotations(abc.ABC):
    """Abstract base class for reading annotation data.

    Annotation data may be stored in a variety of formats; csv files,
    pickled objects, etc. This ABC defines all annotation readers as context
    managers that read annotation files. Inheritors must override: open,
    label, time, duration and channel methods.

    Attributes:
        path:
            Python path instance to an annotation file.
        **kwargs:
            Any valid kwarg for concrete 'open' method.
    """

    def __init__(self, path: typing.Union[str, Path], **kwargs) -> None:
        """Initialize this Annotations reader.

        Args:
            path:
                A path location to an annotation file.
            **kwargs:
                Any valid kwarg for a subclasses 'open' method.
        """

        self.path = path
        self.kwargs = kwargs

    def __enter__(self):
        """Return this instance as target variable of this context."""

        # pylint: disable-next=attribute-defined-outside-init
        self._fobj, self._reader = self.open(self.path, **self.kwargs)
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        """Closes this instance's file obj. & propagate errors by returning
        None."""

        if self._fobj:
            self._fobj.close()

    @abc.abstractmethod
    def open(self, path: Path) -> typing.Tuple[typing.IO, typing.Iterable]:
        """Opens a file at path returning a file handle & row iterator."""

    @abc.abstractmethod
    def label(self, row: typing.Iterable) -> str:
        """Reads the annotation label at row in this file."""

    @abc.abstractmethod
    def time(self, row: typing.Iterable) -> float:
        """Reads annotation time in secs from recording start at row."""

    @abc.abstractmethod
    def duration(self, row: typing.Iterable) -> float:
        """Returns the duration of the annotated event in secs."""

    @abc.abstractmethod
    def channel(self, row: typing.Iterable) -> typing.Union[int, str]:
        """Returns the channel this annotated event was detected on."""

    def read(self,
             labels: typing.Optional[typing.Sequence[str]] = None,
    ) -> typing.List[Annotation]:
        """Reads annotations with labels to a list of Annotation instances.

        Args:
            labels:
                A sequence of annotation string labels for which Annotation
                instances will be returned. If None, return all.

        Returns:
            A list of Annotation dataclass instances (see Annotation).
        """

        labels = [labels] if isinstance(labels, str) else labels

        result = []
        names = ['label', 'time', 'duration', 'channel']
        for row in self._reader:
            ann = Annotation(*[getattr(self, name)(row) for name in names])

            if labels is None:
                result.append(ann)

            elif ann.label in labels:
                result.append(ann)

            else:
                continue

        return result

__init__(path, **kwargs)

Initialize this Annotations reader.

Parameters:

Name Type Description Default
path typing.Union[str, Path]

A path location to an annotation file.

 required
**kwargs

Any valid kwarg for a subclasses 'open' method.

 {}
Source code in openseize/file_io/bases.py 
319
320
321
322
323
324
325
326
327
328
329
330
def __init__(self, path: typing.Union[str, Path], **kwargs) -> None:
    """Initialize this Annotations reader.

    Args:
        path:
            A path location to an annotation file.
        **kwargs:
            Any valid kwarg for a subclasses 'open' method.
    """

    self.path = path
    self.kwargs = kwargs

read(labels=None)

Reads annotations with labels to a list of Annotation instances.

Parameters:

Name Type Description Default
labels typing.Optional[typing.Sequence[str]]

A sequence of annotation string labels for which Annotation instances will be returned. If None, return all.

 None

Returns:

Type Description
typing.List[Annotation]

A list of Annotation dataclass instances (see Annotation).
Source code in openseize/file_io/bases.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
def read(self,
         labels: typing.Optional[typing.Sequence[str]] = None,
) -> typing.List[Annotation]:
    """Reads annotations with labels to a list of Annotation instances.

    Args:
        labels:
            A sequence of annotation string labels for which Annotation
            instances will be returned. If None, return all.

    Returns:
        A list of Annotation dataclass instances (see Annotation).
    """

    labels = [labels] if isinstance(labels, str) else labels

    result = []
    names = ['label', 'time', 'duration', 'channel']
    for row in self._reader:
        ann = Annotation(*[getattr(self, name)(row) for name in names])

        if labels is None:
            result.append(ann)

        elif ann.label in labels:
            result.append(ann)

        else:
            continue

    return result

openseize.file_io.bases.Annotation dataclass

An object for storing a predefined set of annotation attributes that can be updated with user defined attributes after object creation.

Attributes:

Name Type Description
label str

The string name of this annotation.
time float

The time this annotation was made in seconds relative to the recording start.
duration float

The duration of this annotation in seconds.
channel Any

The string name or integer index of the channel this annotation created from.
Source code in openseize/file_io/bases.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
@dataclass
class Annotation:
    """An object for storing a predefined set of annotation attributes that
    can be updated with user defined attributes after object creation.

    Attributes:
        label (str):
            The string name of this annotation.
        time (float):
            The time this annotation was made in seconds relative to the
            recording start.
        duration (float):
            The duration of this annotation in seconds.
        channel (Any):
            The string name or integer index of the channel this annotation
            created from.
    """

    label: str
    time: float
    duration: float
    channel: typing.Any