generated from mscheltienne/template-python
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
d3901e0
commit e51c649
Showing
8 changed files
with
546 additions
and
546 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,4 +1,4 @@ | ||
from . import io, lsl, mock, parallel | ||
from .lsl import LSLTrigger | ||
from .mock import MockTrigger | ||
from .parallel import ParallelPortTrigger | ||
from . import io, lsl, mock, parallel | ||
from .lsl import LSLTrigger | ||
from .mock import MockTrigger | ||
from .parallel import ParallelPortTrigger |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,34 +1,34 @@ | ||
from __future__ import annotations | ||
|
||
from abc import ABC, abstractmethod | ||
|
||
|
||
class BaseTrigger(ABC): | ||
"""Base trigger class.""" | ||
|
||
@abstractmethod | ||
def __init__(self): # pragma: no cover | ||
pass | ||
|
||
@abstractmethod | ||
def signal(self, value: int) -> int: | ||
"""Send a trigger value. | ||
Parameters | ||
---------- | ||
value : int | ||
Value of the trigger, between 1 and 255. | ||
""" | ||
try: | ||
value = int(value) | ||
except TypeError: | ||
raise TypeError( | ||
"The argument 'value' of a trigger must be an integer " | ||
"between 1 and 255 included." | ||
) | ||
if not (1 <= value <= 255): | ||
raise ValueError( | ||
"The argument 'value' of a trigger must be an integer " | ||
"between 1 and 255 included." | ||
) | ||
return value | ||
from __future__ import annotations | ||
|
||
from abc import ABC, abstractmethod | ||
|
||
|
||
class BaseTrigger(ABC): | ||
"""Base trigger class.""" | ||
|
||
@abstractmethod | ||
def __init__(self): # pragma: no cover | ||
pass | ||
|
||
@abstractmethod | ||
def signal(self, value: int) -> int: | ||
"""Send a trigger value. | ||
Parameters | ||
---------- | ||
value : int | ||
Value of the trigger, between 1 and 255. | ||
""" | ||
try: | ||
value = int(value) | ||
except TypeError: | ||
raise TypeError( | ||
"The argument 'value' of a trigger must be an integer " | ||
"between 1 and 255 included." | ||
) | ||
if not (1 <= value <= 255): | ||
raise ValueError( | ||
"The argument 'value' of a trigger must be an integer " | ||
"between 1 and 255 included." | ||
) | ||
return value |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,116 +1,116 @@ | ||
from __future__ import annotations | ||
|
||
from typing import TYPE_CHECKING | ||
|
||
import numpy as np | ||
|
||
from ..utils._checks import check_type | ||
from ..utils._docs import copy_doc, fill_doc | ||
from ..utils._imports import import_optional_dependency | ||
from ._base import BaseTrigger | ||
|
||
if TYPE_CHECKING: | ||
from mne_lsl.lsl import StreamInfo, StreamOutlet | ||
|
||
|
||
@fill_doc | ||
class LSLTrigger(BaseTrigger): | ||
"""Trigger sending values on an LSL outlet. | ||
Make sure you are recording the stream created by the | ||
:class:`~stimuli.trigger.LSLTrigger` alongside your data. e.g. if you use | ||
LabRecorder, update the stream list after creating the | ||
:class:`~stimuli.trigger.LSLTrigger`. | ||
.. warning:: | ||
Make sure to close the :class:`~mne_lsl.lsl.StreamOutlet` by calling the | ||
:meth:`~stimuli.trigger.LSLTrigger.close` method or by deleting the trigger | ||
after use. | ||
Parameters | ||
---------- | ||
name : str | ||
Name of the trigger displayed on the LSL network. | ||
Notes | ||
----- | ||
The :class:`~mne_lsl.lsl.StreamOutlet` created has the following properties: | ||
* Name: ``f"{name}"`` | ||
* Type: ``"Markers"`` | ||
* Number of channels: 1 | ||
* Sampling rate: Irregular | ||
* Data type: ``np.int8`` | ||
* Source ID: ``f"HNP-{name}"`` | ||
The values sent must be in the range of strictly positive integers defined | ||
by ``np.int8``, 1 to 127 included. | ||
""" | ||
|
||
def __init__(self, name: str) -> None: | ||
import_optional_dependency("mne_lsl") | ||
|
||
from mne_lsl.lsl import StreamInfo, StreamOutlet | ||
|
||
check_type(name, (str,), "name") | ||
self._name = name | ||
# create outlet | ||
self._sinfo = StreamInfo( | ||
name=name, | ||
stype="Markers", | ||
n_channels=1, | ||
sfreq=0.0, | ||
dtype="int8", | ||
source_id=f"HNP-{name}", | ||
) | ||
self._sinfo.set_channel_names(["STI"]) | ||
self._sinfo.set_channel_types(["stim"]) | ||
self._sinfo.set_channel_units(["none"]) | ||
self._outlet = StreamOutlet(self._sinfo, max_buffered=1) | ||
|
||
@copy_doc(BaseTrigger.signal) | ||
def signal(self, value: int) -> None: | ||
value = super().signal(value) | ||
if not (1 <= value <= 127): | ||
raise ValueError( | ||
"The argument 'value' of an LSL trigger must be an integer " | ||
"between 1 and 127 included." | ||
) | ||
self._outlet.push_sample(np.array([value], dtype=np.int8)) | ||
|
||
def close(self) -> None: | ||
"""Close the LSL outlet.""" | ||
if hasattr(self, "_outlet"): | ||
try: | ||
del self._outlet | ||
except Exception: # pragma: no cover | ||
pass | ||
|
||
def __del__(self) -> None: # noqa: D105 | ||
self.close() | ||
|
||
# -------------------------------------------------------------------- | ||
@property | ||
def name(self) -> str: | ||
"""Name of the trigger displayed on the LSL network. | ||
:type: str | ||
""" | ||
return self._name | ||
|
||
@property | ||
def sinfo(self) -> StreamInfo: | ||
"""Description of the trigger outlet. | ||
:type: `~mne_lsl.lsl.StreamInfo` | ||
""" | ||
return self._sinfo | ||
|
||
@property | ||
def outlet(self) -> StreamOutlet: | ||
"""Trigger outlet. | ||
:type: `~mne_lsl.lsl.StreamOutlet` | ||
""" | ||
return self._outlet | ||
from __future__ import annotations | ||
|
||
from typing import TYPE_CHECKING | ||
|
||
import numpy as np | ||
|
||
from ..utils._checks import check_type | ||
from ..utils._docs import copy_doc, fill_doc | ||
from ..utils._imports import import_optional_dependency | ||
from ._base import BaseTrigger | ||
|
||
if TYPE_CHECKING: | ||
from mne_lsl.lsl import StreamInfo, StreamOutlet | ||
|
||
|
||
@fill_doc | ||
class LSLTrigger(BaseTrigger): | ||
"""Trigger sending values on an LSL outlet. | ||
Make sure you are recording the stream created by the | ||
:class:`~stimuli.trigger.LSLTrigger` alongside your data. e.g. if you use | ||
LabRecorder, update the stream list after creating the | ||
:class:`~stimuli.trigger.LSLTrigger`. | ||
.. warning:: | ||
Make sure to close the :class:`~mne_lsl.lsl.StreamOutlet` by calling the | ||
:meth:`~stimuli.trigger.LSLTrigger.close` method or by deleting the trigger | ||
after use. | ||
Parameters | ||
---------- | ||
name : str | ||
Name of the trigger displayed on the LSL network. | ||
Notes | ||
----- | ||
The :class:`~mne_lsl.lsl.StreamOutlet` created has the following properties: | ||
* Name: ``f"{name}"`` | ||
* Type: ``"Markers"`` | ||
* Number of channels: 1 | ||
* Sampling rate: Irregular | ||
* Data type: ``np.int8`` | ||
* Source ID: ``f"HNP-{name}"`` | ||
The values sent must be in the range of strictly positive integers defined | ||
by ``np.int8``, 1 to 127 included. | ||
""" | ||
|
||
def __init__(self, name: str) -> None: | ||
import_optional_dependency("mne_lsl") | ||
|
||
from mne_lsl.lsl import StreamInfo, StreamOutlet | ||
|
||
check_type(name, (str,), "name") | ||
self._name = name | ||
# create outlet | ||
self._sinfo = StreamInfo( | ||
name=name, | ||
stype="Markers", | ||
n_channels=1, | ||
sfreq=0.0, | ||
dtype="int8", | ||
source_id=f"HNP-{name}", | ||
) | ||
self._sinfo.set_channel_names(["STI"]) | ||
self._sinfo.set_channel_types(["stim"]) | ||
self._sinfo.set_channel_units(["none"]) | ||
self._outlet = StreamOutlet(self._sinfo, max_buffered=1) | ||
|
||
@copy_doc(BaseTrigger.signal) | ||
def signal(self, value: int) -> None: | ||
value = super().signal(value) | ||
if not (1 <= value <= 127): | ||
raise ValueError( | ||
"The argument 'value' of an LSL trigger must be an integer " | ||
"between 1 and 127 included." | ||
) | ||
self._outlet.push_sample(np.array([value], dtype=np.int8)) | ||
|
||
def close(self) -> None: | ||
"""Close the LSL outlet.""" | ||
if hasattr(self, "_outlet"): | ||
try: | ||
del self._outlet | ||
except Exception: # pragma: no cover | ||
pass | ||
|
||
def __del__(self) -> None: # noqa: D105 | ||
self.close() | ||
|
||
# -------------------------------------------------------------------- | ||
@property | ||
def name(self) -> str: | ||
"""Name of the trigger displayed on the LSL network. | ||
:type: str | ||
""" | ||
return self._name | ||
|
||
@property | ||
def sinfo(self) -> StreamInfo: | ||
"""Description of the trigger outlet. | ||
:type: `~mne_lsl.lsl.StreamInfo` | ||
""" | ||
return self._sinfo | ||
|
||
@property | ||
def outlet(self) -> StreamOutlet: | ||
"""Trigger outlet. | ||
:type: `~mne_lsl.lsl.StreamOutlet` | ||
""" | ||
return self._outlet |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,21 +1,21 @@ | ||
from __future__ import annotations | ||
|
||
from ..utils._docs import copy_doc | ||
from ..utils.logs import _use_log_level, logger | ||
from ._base import BaseTrigger | ||
|
||
|
||
class MockTrigger(BaseTrigger): | ||
"""Mock trigger class. | ||
Delivered triggers are logged at the 'INFO' level. | ||
""" | ||
|
||
def __init__(self) -> None: | ||
pass | ||
|
||
@copy_doc(BaseTrigger.signal) | ||
def signal(self, value: int) -> None: | ||
value = super().signal(value) | ||
with _use_log_level("INFO"): | ||
logger.info("Mock set to %i.", value) | ||
from __future__ import annotations | ||
|
||
from ..utils._docs import copy_doc | ||
from ..utils.logs import _use_log_level, logger | ||
from ._base import BaseTrigger | ||
|
||
|
||
class MockTrigger(BaseTrigger): | ||
"""Mock trigger class. | ||
Delivered triggers are logged at the 'INFO' level. | ||
""" | ||
|
||
def __init__(self) -> None: | ||
pass | ||
|
||
@copy_doc(BaseTrigger.signal) | ||
def signal(self, value: int) -> None: | ||
value = super().signal(value) | ||
with _use_log_level("INFO"): | ||
logger.info("Mock set to %i.", value) |
Oops, something went wrong.