diff --git a/THIRD-PARTY-NOTICES.md b/THIRD-PARTY-NOTICES.md index d86d36a..40d63bd 100644 --- a/THIRD-PARTY-NOTICES.md +++ b/THIRD-PARTY-NOTICES.md @@ -37,6 +37,36 @@ Catch2 (https://github.com/catchorg/Catch2), Boost Software License --- +## pylsl (MIT License) + +Python binding, vendored as a component at `apps/pylsl/`. Retains its +upstream MIT license, including the security integration added here; the notice +is also kept at `apps/pylsl/LICENSE`. + +``` +Copyright (c) 2012-2018 Christian A. Kothe + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. +``` + +--- + ## asio (Boost Software License 1.0) Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com) diff --git a/apps/pylsl/.gitignore b/apps/pylsl/.gitignore new file mode 100644 index 0000000..ecc95a7 --- /dev/null +++ b/apps/pylsl/.gitignore @@ -0,0 +1,21 @@ +/dist/ +/.idea/ +/src/pylsl.egg-info/ +/build/ +__pycache__ +*.so +*.so.* +*.pyc +*.dll +*.dylib +*.cprof +*.png +/wheelhouse/ +/src/pylsl/include +/src/pylsl/share +.DS_Store + +uv.lock +/src/pylsl/__version__.py +/src/pylsl/lib/lslver.exe +liblsl.zip diff --git a/apps/pylsl/COMPONENT.md b/apps/pylsl/COMPONENT.md new file mode 100644 index 0000000..f7541dc --- /dev/null +++ b/apps/pylsl/COMPONENT.md @@ -0,0 +1,28 @@ +# pylsl component + +Security-enabled Python binding for Lab Streaming Layer, vendored into this +monorepo as a component. + +## Licensing + +This component retains its upstream license (MIT, Copyright (c) 2012-2018 +Christian A. Kothe); see `LICENSE` in this directory. The security integration +added here is contributed under that same MIT license. It consists of `ctypes` +interface declarations and two stream properties that call the public liblsl C +API, and contains no cryptographic implementation, so the proprietary Secure LSL +terms do not apply to it. See COMPONENT LICENSING POLICY in the repository-root +`LICENSE`. + +## Upstream tracking + +- Upstream: https://github.com/labstreaminglayer/pylsl +- Based on upstream commit: `c90f623` +- Security integration: interface declarations plus `StreamInfo.security_enabled` + and `StreamInfo.security_fingerprint` + +## Graceful degradation + +The security symbols are resolved inside `try`/`except AttributeError` and +recorded in `_security_api_available` / `_stream_security_api_available`. Linked +against a stock `liblsl` that does not export them, the binding imports and +behaves exactly as upstream; the security properties simply report unavailable. diff --git a/apps/pylsl/LICENSE b/apps/pylsl/LICENSE new file mode 100644 index 0000000..0995c11 --- /dev/null +++ b/apps/pylsl/LICENSE @@ -0,0 +1,19 @@ +Copyright (c) 2012-2018 Christian A. Kothe + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. diff --git a/apps/pylsl/MANIFEST.in b/apps/pylsl/MANIFEST.in new file mode 100644 index 0000000..1931c39 --- /dev/null +++ b/apps/pylsl/MANIFEST.in @@ -0,0 +1,7 @@ +include README.md +include LICENSE + +# If using Python 2.6 or less, then have to include package data, even though +# it's already declared in setup.py +include src/pylsl/lib/liblsl*.* +include src/pylsl/lib/lsl*.* diff --git a/apps/pylsl/README.md b/apps/pylsl/README.md new file mode 100644 index 0000000..b3d043c --- /dev/null +++ b/apps/pylsl/README.md @@ -0,0 +1,86 @@ +# pylsl + +![publish workflow](https://github.com/labstreaminglayer/pylsl/actions/workflows/publish-to-pypi.yml/badge.svg) +[![PyPI version](https://badge.fury.io/py/pylsl.svg)](https://badge.fury.io/py/pylsl) + +This is the Python interface to the [Lab Streaming Layer (LSL)](https://github.com/sccn/labstreaminglayer). +LSL is an overlay network for real-time exchange of time series between applications, +most often used in research environments. LSL has clients for many other languages +and platforms that are compatible with each other. + +Let us know if you encounter any bugs (ideally using the issue tracker on +the GitHub project). + +# Installation + +## Prerequisites + +On all non-Windows platforms and for some Windows-Python combinations, you must first obtain a liblsl shared library. See the [liblsl repo documentation](https://github.com/sccn/liblsl) for further details. + +## Get pylsl from PyPI + +* `pip install pylsl` + +## Get pylsl from source + +This should only be necessary if you need to modify or debug pylsl. + +* Download the pylsl source: `git clone https://github.com/labstreaminglayer/pylsl.git && cd pylsl` +* From the `pylsl` working directory, run `pip install .`. + * Note: You can use `pip install -e .` to install while keeping the files in-place. This is convenient for developing pylsl. + +# Usage + +See the examples in src/pylsl/examples. Note that these can be run directly from the commandline with (e.g.) `python -m pylsl.examples.{name-of-example}`. + +You can get a list of the examples with `python -c "import pylsl.examples; help(pylsl.examples)"` + +## liblsl loading + +`pylsl` will search for `liblsl` first at the filepath specified by an environment variable named `PYLSL_LIB`, then in the package directory (default location for Windows), then finally in normal system library folders. + +If the shared object is not installed onto a standard search path (or it is but can't be found for some [other bug](https://github.com/labstreaminglayer/pylsl/issues/48)), then we recommend that you copy it to the pylsl installed module path's `lib` subfolder. i.e. `{path/to/env/}site-packages/pylsl/lib`. + +* The `site-packages/pylsl` path will only exist _after_ you install `pylsl` in your Python environment. +* You may have to create the `lib` subfolder. +* Use `python -m site` to find the "site-packages" path. +* Use `cp -L` on platforms that use symlinks. + +Alternatively, you can use an environment variable. Set the `PYLSL_LIB` environment variable to the location of the library or set `LD_LIBRARY_PATH` to the folder containing the library. For example, + +1. `PYLSL_LIB=/usr/local/lib/liblsl.so python -m pylsl.examples.{name-of-example}`, or +2. `LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/local/lib python -m pylsl.examples.{name-of-example}` + +# For maintainers + +## Continuous Integration + +pylsl uses continuous integration and distribution. GitHub Actions will upload a new release to pypi whenever a Release is created in GitHub. +Before creating the GitHub release, be sure to bump the version number in `pylsl/version.py` and consider updating the liblsl dependency +in `.github/workflows/publish-to-pypi.yml`. + +### Linux Binaries Deprecated + +We recently stopped building binary wheels for Linux. In practice, the `manylinux` dependencies were often incompatible with real systems. + +## Manual Distribution + +1. Manual way: + 1. `rm -Rf build dist *.egg-info` + 1. `python setup.py sdist bdist_wheel` + 1. Additional steps on Linux: + * `auditwheel repair dist/*.whl -w dist` + * `rm dist/*-linux_x86_64.whl` + 1. `twine upload dist/*` +1. For conda + 1. build liblsl: `conda build ../liblsl/` + 1. `conda build .` + +# Known Issues with Multithreading on Linux + +* At least for some versions of pylsl, it has been reported that running on Linux one cannot call ``pylsl`` functions from a thread that is not the main thread. This has been reported to cause access violations, and can occur during pulling from an inlet, and also from accessing an inlets info structure in a thread. +* Recent tests with multithreading (especially when safeguarding library calls with locks) using Python 3.7.6. with pylsl 1.14 on Linux Mint 20 suggest that this issue is solved, or at least depends on your machine. See https://github.com/labstreaminglayer/pylsl/issues/29 + +# Acknowledgments + +Pylsl was primarily written by Christian Kothe while at Swartz Center for Computational Neuroscience, UCSD. The LSL project was funded by the Army Research Laboratory under Cooperative Agreement Number W911NF-10-2-0022 as well as through NINDS grant 3R01NS047293-06S1. pylsl is maintained primarily by Chadwick Boulay. Thanks for contributions, bug reports, and suggestions go to Bastian Venthur, David Medine, Clemens Brunner, and Matthew Grivich. diff --git a/apps/pylsl/pyproject.toml b/apps/pylsl/pyproject.toml new file mode 100644 index 0000000..9c568ea --- /dev/null +++ b/apps/pylsl/pyproject.toml @@ -0,0 +1,68 @@ +[project] +name = "pylsl" +description = "Python library for importing XDF (Extensible Data Format)" +authors = [ + { name = "Christian Kothe", email = "christian.kothe@intheon.io" }, + { name = "Chadwick Boulay", email = "chadwick.boulay@gmail.com" } +] +#license = {file = "LICENSE"} # Bug in setuptools. https://github.com/astral-sh/uv/issues/9513 +readme = "README.md" +requires-python = ">=3.9" +dynamic = ["version"] +keywords = [ + "networking", + "LSL", + "Lab Streaming Layer", + "labstreaminglayer", + "data", + "acquisition", + "stream" +] +classifiers = [ + "Development Status :: 5 - Production/Stable", + "Intended Audience :: Developers", + "Intended Audience :: Science/Research", + "Topic :: System :: Networking", + "Topic :: Scientific/Engineering", + "License :: OSI Approved :: MIT License", + "Operating System :: Microsoft :: Windows", + "Operating System :: POSIX :: Linux", + "Operating System :: MacOS", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", +] +dependencies = [ + "numpy>=1.21,<3", +] + +[project.urls] +Repository = "https://github.com/labstreaminglayer/pylsl" +Issues = "https://github.com/labstreaminglayer/pylsl/issues" + +[project.optional-dependencies] +examples = [ + "pyqtgraph>=0.13.7", +] +#Changelog = "https://github.com/labstreaminglayer/pylsl/blob/main/CHANGELOG.md" + +[dependency-groups] +dev = [ + "pytest>=8.3.4", + "ruff>=0.8.2", +] + +[build-system] +requires = ["setuptools>=64", "setuptools-scm>=8"] +build-backend = "setuptools.build_meta" + +[tool.setuptools] +license-files = [] + +[tool.setuptools_scm] +version_file = "src/pylsl/__version__.py" + +[tool.setuptools.package-data] +pylsl = ["lib/*.dll"] diff --git a/apps/pylsl/setup.py b/apps/pylsl/setup.py new file mode 100644 index 0000000..bb0db20 --- /dev/null +++ b/apps/pylsl/setup.py @@ -0,0 +1,32 @@ +import sys + +from setuptools import setup +# from setuptools.dist import Distribution + + +# class BinaryDistribution(Distribution): +# """Distribution which always forces a binary package with platform name""" +# def has_ext_modules(foo): +# return sys.platform.startswith("win") + +try: + from wheel.bdist_wheel import bdist_wheel as _bdist_wheel + + class bdist_wheel(_bdist_wheel): + def finalize_options(self): + super().finalize_options() + self.root_is_pure = not sys.platform.startswith("win") + + def get_tag(self): + python, abi, plat = _bdist_wheel.get_tag(self) + # We don't contain any python source + python, abi = "py2.py3", "none" + return python, abi, plat +except ImportError: + bdist_wheel = None + + +setup( + # distclass=BinaryDistribution, + cmdclass={"bdist_wheel": bdist_wheel}, +) diff --git a/apps/pylsl/src/pylsl/__init__.py b/apps/pylsl/src/pylsl/__init__.py new file mode 100644 index 0000000..8d11d93 --- /dev/null +++ b/apps/pylsl/src/pylsl/__init__.py @@ -0,0 +1,48 @@ +"""Python API for the lab streaming layer. + +The lab streaming layer provides a set of functions to make instrument data +accessible in real time within a lab network. From there, streams can be +picked up by recording programs, viewing programs or custom experiment +applications that access data streams in real time. + +The API covers two areas: +- The "push API" allows to create stream outlets and to push data (regular + or irregular measurement time series, event data, coded audio/video frames, + etc.) into them. +- The "pull API" allows to create stream inlets and read time-synched + experiment data from them (for recording, viewing or experiment control). + +""" + +from .__version__ import __version__ as __version__ +from .resolve import ContinuousResolver as ContinuousResolver +from .resolve import resolve_streams as resolve_streams +from .resolve import resolve_bypred as resolve_bypred +from .resolve import resolve_byprop as resolve_byprop +from .info import StreamInfo as StreamInfo +from .inlet import StreamInlet as StreamInlet +from .outlet import StreamOutlet as StreamOutlet +from .util import IRREGULAR_RATE as IRREGULAR_RATE +from .util import FOREVER as FOREVER +from .util import proc_none as proc_none +from .util import proc_clocksync as proc_clocksync +from .util import proc_dejitter as proc_dejitter +from .util import proc_monotonize as proc_monotonize +from .util import proc_threadsafe as proc_threadsafe +from .util import proc_ALL as proc_ALL +from .util import protocol_version as protocol_version +from .util import library_version as library_version +from .util import library_info as library_info +from .util import local_clock as local_clock +from .util import is_secure_build as is_secure_build +from .util import base_version as base_version +from .util import security_version as security_version +from .util import full_version as full_version +from .util import check_security as check_security +from .lib import cf_int8 as cf_int8 +from .lib import cf_int16 as cf_int16 +from .lib import cf_int32 as cf_int32 +from .lib import cf_int64 as cf_int64 +from .lib import cf_float32 as cf_float32 +from .lib import cf_double64 as cf_double64 +from .lib import cf_string as cf_string diff --git a/apps/pylsl/src/pylsl/examples/GetTimeCorrection.py b/apps/pylsl/src/pylsl/examples/GetTimeCorrection.py new file mode 100644 index 0000000..76f4fdf --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/GetTimeCorrection.py @@ -0,0 +1,25 @@ +"""Example program to show how to read a multi-channel time series from LSL.""" + +import time + +from pylsl import StreamInlet, resolve_byprop + + +def main(): + # first resolve an EEG stream on the lab network + print("looking for an EEG stream...") + streams = resolve_byprop("type", "EEG") + info = streams[0] + + # create a new inlet to read from the stream + inlet = StreamInlet(info) + + print("Connected to outlet " + info.name() + "@" + info.hostname()) + while True: + offset = inlet.time_correction() + print("Offset: " + str(offset)) + time.sleep(1) + + +if __name__ == "__main__": + main() diff --git a/apps/pylsl/src/pylsl/examples/HandleMetadata.py b/apps/pylsl/src/pylsl/examples/HandleMetadata.py new file mode 100644 index 0000000..8c9f80f --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/HandleMetadata.py @@ -0,0 +1,61 @@ +"""Example program that shows how to attach meta-data to a stream, and how to +later on retrieve the meta-data again at the receiver side.""" + +import time + +import numpy as np +from pylsl import StreamInfo, StreamInlet, StreamOutlet, resolve_byprop + + +def main(): + # create a new StreamInfo object which shall describe our stream + info = StreamInfo("MetaTester", "EEG", 8, 100, "float32", "myuid56872") + + # now attach some meta-data (in accordance with XDF format, + # see also https://github.com/sccn/xdf/wiki/Meta-Data) + chns = info.desc().append_child("channels") + ch_labels = ["C3", "C4", "Cz", "FPz", "POz", "CPz", "O1", "O2"] + for label in ch_labels: + ch = chns.append_child("channel") + ch.append_child_value("label", label) + ch.append_child_value("unit", "microvolts") + ch.append_child_value("type", "EEG") + info.desc().append_child_value("manufacturer", "SCCN") + cap = info.desc().append_child("cap") + cap.append_child_value("name", "EasyCap") + cap.append_child_value("size", "54") + cap.append_child_value("labelscheme", "10-20") + + # create outlet for the stream + outlet = StreamOutlet(info) + + # Send a sample into the outlet... + dummy_sample = np.arange(len(ch_labels), dtype=np.float32) + outlet.push_sample(dummy_sample) + + # === the following could run on another computer === + + # first we resolve a stream whose name is MetaTester (note that there are + # other ways to query a stream, too - for instance by content-type) + results = resolve_byprop("name", "MetaTester") + + # open an inlet so we can read the stream's data (and meta-data) + inlet = StreamInlet(results[0]) + + # get the full stream info (including custom meta-data) and dissect it + info = inlet.info() + print("The stream's XML meta-data is: ") + print(info.as_xml()) + print("The manufacturer is: %s" % info.desc().child_value("manufacturer")) + print("Cap circumference is: %s" % info.desc().child("cap").child_value("size")) + print("The channel labels are as follows:") + ch = info.desc().child("channels").child("channel") + for k in range(info.channel_count()): + print(" " + ch.child_value("label")) + ch = ch.next_sibling() + + time.sleep(3) + + +if __name__ == "__main__": + main() diff --git a/apps/pylsl/src/pylsl/examples/PerformanceTest.py b/apps/pylsl/src/pylsl/examples/PerformanceTest.py new file mode 100644 index 0000000..3d954b2 --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/PerformanceTest.py @@ -0,0 +1,423 @@ +import random +import time + +import numpy as np + +from pylsl import ( + StreamInfo, + StreamInlet, + StreamOutlet, + local_clock, + proc_clocksync, + proc_dejitter, + proc_monotonize, + resolve_bypred, + resolve_byprop, +) + +try: + from pyfftw.interfaces.numpy_fft import irfft + # Performs much better than numpy's fftpack +except ImportError: + from numpy.fft import irfft +try: + import sys + + import pyqtgraph as pg + + haspyqtgraph = True +except ImportError: + haspyqtgraph = False + + +# The code for pink noise generation is taken from +# https://github.com/python-acoustics/python-acoustics/blob/master/acoustics/generator.py +# which is distributed under the BSD license. +def ms(x): + """Mean value of signal `x` squared. + :param x: Dynamic quantity. + :returns: Mean squared of `x`. + """ + return (np.abs(x) ** 2.0).mean() + + +def normalize(y, x=None): + """normalize power in y to a (standard normal) white noise signal. + Optionally normalize to power in signal `x`. + #The mean power of a Gaussian with :math:`\\mu=0` and :math:`\\sigma=1` is 1. + """ + # return y * np.sqrt( (np.abs(x)**2.0).mean() / (np.abs(y)**2.0).mean() ) + if x is not None: + x = ms(x) + else: + x = 1.0 + return y * np.sqrt(x / ms(y)) + # return y * np.sqrt( 1.0 / (np.abs(y)**2.0).mean() ) + + +def pink(N): + """ + Pink noise. + + :param N: Amount of samples. + + Pink noise has equal power in bands that are proportionally wide. + Power density decreases with 3 dB per octave. + + """ + # This method uses the filter with the following coefficients. + # b = np.array([0.049922035, -0.095993537, 0.050612699, -0.004408786]) + # a = np.array([1, -2.494956002, 2.017265875, -0.522189400]) + # return lfilter(B, A, np.random.randn(N)) + # Another way would be using the FFT + # x = np.random.randn(N) + # X = rfft(x) / N + uneven = N % 2 + X = np.random.randn(N // 2 + 1 + uneven) + 1j * np.random.randn(N // 2 + 1 + uneven) + S = np.sqrt(np.arange(len(X)) + 1.0) # +1 to avoid divide by zero + y = (irfft(X / S)).real + if uneven: + y = y[:-1] + return normalize(y) + + +class PinkNoiseGenerator(object): + def __init__(self, nSampsPerBlock=1024): + self.N = nSampsPerBlock + self.uneven = self.N % 2 + lenX = self.N // 2 + 1 + self.uneven + self.S = np.sqrt(np.arange(lenX) + 1.0) + + def generate(self): + X = np.random.randn(self.N // 2 + 1 + self.uneven) + 1j * np.random.randn( + self.N // 2 + 1 + self.uneven + ) + y = (irfft(X / self.S)).real + if self.uneven: + y = y[:-1] + return normalize(y) + + +class BetaGeneratorOutlet(object): + def __init__( + self, + Fs=2**14, + FreqBeta=20.0, + AmpBeta=100.0, + AmpNoise=20.0, + NCyclesPerChunk=4, + channels=["RAW1", "SPK1", "RAW2", "SPK2", "RAW3", "SPK3"], + ): + """ + :param Fs: Sampling rate + :param FreqBeta: Central frequency of beta band + :param AmpBeta: Amplitude of beta (uV) + :param AmpNoise: Amplitude of pink noise (uV) + :param NCyclesPerChunk: Minimum number of cycles of beta in a chunk. + :param channels: List of channel names + """ + # Saved arguments + self.FreqBeta = FreqBeta + self.AmpBeta = AmpBeta # Amplitude of Beta (uV) + self.AmpNoise = AmpNoise # Amplitude of pink noise + self.channels = channels + # Derived variables + chunk_dur = NCyclesPerChunk / self.FreqBeta # Duration, in sec, of one chunk + chunk_len = int(Fs * chunk_dur) # Number of samples in a chunk + self.tvec = 1.0 * (np.arange(chunk_len) + 1) / Fs # time vector for chunk (sec) + # Pink noise generator + self.pinkNoiseGen = PinkNoiseGenerator(nSampsPerBlock=chunk_len) + + # Create a stream of fake 'raw' data + raw_info = StreamInfo( + name="BetaGen", + type="EEG", + channel_count=len(self.channels), + nominal_srate=Fs, + channel_format="float32", + source_id="betagen1234", + ) + raw_xml = raw_info.desc() + chans = raw_xml.append_child("channels") + for channame in self.channels: + chn = chans.append_child("channel") + chn.append_child_value("label", channame) + chn.append_child_value("unit", "microvolts") + chn.append_child_value("type", "generated") + self.eeg_outlet = StreamOutlet(raw_info) + print("Created outlet with name BetaGen and type EEG") + + self.last_time = local_clock() + + def update(self, task={"phase": "precue", "class": 1}): + # Convert phase and class_id into beta_amp + if task["phase"] in ["cue", "go"]: + beta_amp = 0 if task["class"] == 3 else self.AmpBeta + else: + beta_amp = self.AmpBeta / 5.0 + + this_tvec = self.tvec + self.last_time # Sample times + # Put the signal together + this_sig = self.AmpNoise * np.asarray( + self.pinkNoiseGen.generate(), dtype=np.float32 + ) # Start with some pink noise + this_sig += beta_amp * np.sin( + this_tvec * 2 * np.pi * self.FreqBeta + ) # Add our beta signal + this_sig = np.atleast_2d(this_sig).T * np.ones( + (1, len(self.channels)), dtype=np.float32 + ) # Tile across channels + + time_to_sleep = max(0, this_tvec[-1] - local_clock()) + time.sleep(time_to_sleep) + + print( + "Beta outlet pushing signal with shape {},{} and Beta amp {}".format( + this_sig.shape[0], this_sig.shape[1], beta_amp + ) + ) + self.eeg_outlet.push_chunk(this_sig, timestamp=this_tvec[-1]) + + self.last_time = local_clock() + + +class BetaInlet(object): + def __init__(self): + print("looking for an EEG stream...") + streams = resolve_byprop("type", "EEG") + + # create a new inlet to read from the stream + proc_flags = proc_clocksync | proc_dejitter | proc_monotonize + self.inlet = StreamInlet(streams[0], processing_flags=proc_flags) + + # The following is an example of how to read stream info + stream_info = self.inlet.info() + stream_Fs = stream_info.nominal_srate() + stream_xml = stream_info.desc() + chans_xml = stream_xml.child("channels") + chan_xml_list = [] + ch = chans_xml.child("channel") + while ch.name() == "channel": + chan_xml_list.append(ch) + ch = ch.next_sibling("channel") + self.channel_names = [ch_xml.child_value("label") for ch_xml in chan_xml_list] + print( + "Reading from inlet named {} with channels {} sending data at {} Hz".format( + stream_info.name(), self.channel_names, stream_Fs + ) + ) + + def update(self): + max_samps = 3276 * 2 + data = np.nan * np.ones((max_samps, len(self.channel_names)), dtype=np.float32) + _, timestamps = self.inlet.pull_chunk(max_samples=max_samps, dest_obj=data) + data = data[: len(timestamps), :] + print("Beta inlet retrieved {} samples.".format(len(timestamps))) + return data, np.asarray(timestamps) + + +class MarkersGeneratorOutlet(object): + phases = { + "precue": {"next": "cue", "duration": 1.0}, + "cue": {"next": "go", "duration": 0.5}, + "go": {"next": "evaluate", "duration": 5.0}, + "evaluate": {"next": "precue", "duration": 0.1}, + } + + def __init__( + self, + class_list=[1, 3], + classes_rand=True, + target_list=[1, 2], + targets_rand=True, + ): + """ + + :param class_list: A list of integers comprising different class ids. Default: [1, 3] + :param classes_rand: If True, classes are chosen randomly from list. If False, the list is cycled. Default: True + :param target_list: A list of integers comprising different target ids. Default: [1, 2] + :param targets_rand: If True, targets are chosen randomly from list. If False, the list is cycled. Default: True + """ + stream_name = "GeneratedCentreOutMarkers" + stream_type = "Markers" + outlet_info = StreamInfo( + name=stream_name, + type=stream_type, + channel_count=1, + nominal_srate=0, + channel_format="string", + source_id="centreoutmarkergen1234", + ) + outlet_xml = outlet_info.desc() + channels_xml = outlet_xml.append_child("channels") + chan_xml = channels_xml.append_child("channel") + chan_xml.append_child_value("label", "EventMarkers") + chan_xml.append_child_value("type", "generated") + self.outlet = StreamOutlet(outlet_info) + print( + "Created outlet with name {} and type {}".format(stream_name, stream_type) + ) + + self.class_list = class_list + self.classes_rand = classes_rand + self.target_list = target_list + self.targets_rand = targets_rand + self.next_transition = -1 + self.in_phase = "evaluate" + self.trial_ix = 0 + self.class_id = self.class_list[0] + self.target_id = self.target_list[0] + + def update(self): + now = local_clock() + if now > self.next_transition: + # Transition phase + self.in_phase = self.phases[self.in_phase]["next"] + self.next_transition = now + self.phases[self.in_phase]["duration"] + + # Send markers + out_string = "undefined" + if self.in_phase == "precue": + # transition from evaluate to precue + # print("Previous class_id: {}, target_id: {}".format(self.class_id, self.target_id)) + self.trial_ix += 1 + self.target_id = ( + random.choice(self.target_list) + if self.targets_rand + else self.target_list[ + (self.target_list.index(self.target_id) + 1) + % len(self.target_list) + ] + ) + self.class_id = ( + random.choice(self.class_list) + if self.classes_rand + else self.class_list[ + (self.class_list.index(self.class_id) + 1) + % len(self.class_list) + ] + ) + # print("New class_id: {}, target_id: {}".format(self.class_id, self.target_id)) + out_string = "NewTrial {}, Class {}, Target {}".format( + self.trial_ix, self.class_id, self.target_id + ) + elif self.in_phase == "cue": + # transition from precue to cue + out_string = "TargetCue, Class {}, Target {}".format( + self.class_id, self.target_id + ) + elif self.in_phase == "go": + # transition from cue to go + out_string = "GoCue, Class {}, Target {}".format( + self.class_id, self.target_id + ) + elif self.in_phase == "evaluate": + # transition from go to evaluate + hit_string = "Hit" if random.randint(0, 1) == 1 else "Miss" + out_string = "{}, Class {}, Target {}".format( + hit_string, self.class_id, self.target_id + ) + print("Marker outlet pushing string: {}".format(out_string)) + self.outlet.push_sample( + [ + out_string, + ] + ) + + return True + return False + + +class MarkerInlet(object): + def __init__(self): + self.task = {"phase": "precue", "class": 1, "target": 1} + print("Looking for stream with type Markers") + streams = resolve_bypred("type='Markers'", minimum=1) + proc_flags = 0 # Marker events are relatively rare. No need to post-process. + self.inlet = StreamInlet(streams[0], processing_flags=proc_flags) + # The following is an example of how to read stream info + stream_info = self.inlet.info() + # stream_Fs = stream_info.nominal_srate() + stream_xml = stream_info.desc() + chans_xml = stream_xml.child("channels") + chan_xml_list = [] + ch = chans_xml.child("channel") + while ch.name() == "channel": + chan_xml_list.append(ch) + ch = ch.next_sibling("channel") + stream_ch_names = [ch_xml.child_value("label") for ch_xml in chan_xml_list] + print( + "Reading from inlet named {} with channels {}".format( + stream_info.name(), stream_ch_names + ) + ) + + def update(self): + marker_samples, marker_timestamps = self.inlet.pull_chunk(timeout=0.0) + if marker_timestamps: + [phase_str, class_str, targ_str] = marker_samples[-1][0].split(", ") + if phase_str in ["TargetCue"]: + self.task["phase"] = "cue" + elif phase_str in ["GoCue"]: + self.task["phase"] = "go" + elif phase_str in ["Miss", "Hit"]: + self.task["phase"] = "evaluate" + elif phase_str[:8] == "NewTrial": + self.task["phase"] = "precue" + else: + print(phase_str) + self.task["class"] = int(class_str.split(" ")[1]) + self.task["target"] = int(targ_str.split(" ")[1]) + print("Marker inlet updated with task {}".format(self.task)) + + +betaGen = BetaGeneratorOutlet() +markerGen = MarkersGeneratorOutlet() +betaIn = BetaInlet() +markerIn = MarkerInlet() + +if haspyqtgraph: + qapp = pg.QtGui.QApplication(sys.argv) + qwindow = pg.plot() + qwindow.clear() + qwindow.parent().setWindowTitle("pylsl PerformanceTest") + + +def update(): + markerGen.update() + markerIn.update() + betaGen.update(task=markerIn.task) # Rate-limiting step. Will time.sleep as needed. + signal, tvec = betaIn.update() + + if haspyqtgraph: + plot = qwindow.getPlotItem() + graphs = plot.listDataItems() + if not graphs: + # create graphs + for i in range(signal.shape[1]): + plot.plot(tvec, signal[:, i]) + else: + # update graphs + for i in range(signal.shape[1]): + graphs[i].setData(signal[:, i], x=tvec) + + +if __name__ == "__main__": + """ + python3 -m cProfile -o pylsl.cprof PerformanceTest.py + gprof2dot -f pstats pylsl.cprof | dot -Tpng -o pylsl_prof.png + """ + try: + if haspyqtgraph: + timer = pg.QtCore.QTimer() + timer.timeout.connect(update) + timer.start(1) # Delay not needed because update has time.sleep + if (sys.flags.interactive != 1) or not hasattr(pg.QtCore, "PYQT_VERSION"): + sys.exit(pg.QtGui.QApplication.instance().exec_()) + else: + while True: + update() + + except KeyboardInterrupt: + # No cleanup necessary? + pass diff --git a/apps/pylsl/src/pylsl/examples/README.md b/apps/pylsl/src/pylsl/examples/README.md new file mode 100644 index 0000000..ebb8cc2 --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/README.md @@ -0,0 +1,10 @@ +If pylsl was installed with `pip install pylsl` (recommended), then you can run each of these examples with `python -m pylsl.examples.name_of_example`. + +There are few other noteworthy uses of pylsl in the wild. These might give you some inspiration or direct examples of how to use pylsl in your project. + +* [mne-realtime examples](https://github.com/mne-tools/mne-realtime/tree/master/examples) +* [NeuroPype](https://www.neuropype.io/) +* [Sig-Visualizer](https://github.com/labstreaminglayer/App-SigVisualizer) +* [Pupil-Labs has 2 different pylsl uses](https://github.com/labstreaminglayer/App-PupilLabs) +* [@agricolab's pyliesl](https://github.com/pyreiz/pyliesl) makes use of some less-common features of LSL and XDF. +* [muse-lsl](https://github.com/alexandrebarachant/muse-lsl) diff --git a/apps/pylsl/src/pylsl/examples/ReceiveAndPlot.py b/apps/pylsl/src/pylsl/examples/ReceiveAndPlot.py new file mode 100644 index 0000000..8339395 --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/ReceiveAndPlot.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python +""" +ReceiveAndPlot example for LSL + +This example shows data from all found outlets in realtime. +It illustrates the following use cases: +- efficiently pulling data, re-using buffers +- automatically discarding older samples +- online postprocessing +""" + +import math +from typing import List + +import numpy as np +import pyqtgraph as pg +from pyqtgraph.Qt import QtCore, QtGui + +import pylsl + +# Basic parameters for the plotting window +plot_duration = 5 # how many seconds of data to show +update_interval = 60 # ms between screen updates +pull_interval = 500 # ms between each pull operation + + +class Inlet: + """Base class to represent a plottable inlet""" + + def __init__(self, info: pylsl.StreamInfo): + # create an inlet and connect it to the outlet we found earlier. + # max_buflen is set so data older the plot_duration is discarded + # automatically and we only pull data new enough to show it + + # Also, perform online clock synchronization so all streams are in the + # same time domain as the local lsl_clock() + # (see https://labstreaminglayer.readthedocs.io/projects/liblsl/ref/enums.html#_CPPv414proc_clocksync) + # and dejitter timestamps + self.inlet = pylsl.StreamInlet( + info, + max_buflen=plot_duration, + processing_flags=pylsl.proc_clocksync | pylsl.proc_dejitter, + ) + # store the name and channel count + self.name = info.name() + self.channel_count = info.channel_count() + + def pull_and_plot(self, plot_time: float, plt: pg.PlotItem): + """Pull data from the inlet and add it to the plot. + :param plot_time: lowest timestamp that's still visible in the plot + :param plt: the plot the data should be shown on + """ + # We don't know what to do with a generic inlet, so we skip it. + pass + + +class DataInlet(Inlet): + """A DataInlet represents an inlet with continuous, multi-channel data that + should be plotted as multiple lines.""" + + dtypes = [[], np.float32, np.float64, None, np.int32, np.int16, np.int8, np.int64] + + def __init__(self, info: pylsl.StreamInfo, plt: pg.PlotItem): + super().__init__(info) + # calculate the size for our buffer, i.e. two times the displayed data + bufsize = ( + 2 * math.ceil(info.nominal_srate() * plot_duration), + info.channel_count(), + ) + self.buffer = np.empty(bufsize, dtype=self.dtypes[info.channel_format()]) + empty = np.array([]) + # create one curve object for each channel/line that will handle displaying the data + self.curves = [ + pg.PlotCurveItem(x=empty, y=empty, autoDownsample=True) + for _ in range(self.channel_count) + ] + for curve in self.curves: + plt.addItem(curve) + + def pull_and_plot(self, plot_time, plt): + # pull the data + _, ts = self.inlet.pull_chunk( + timeout=0.0, max_samples=self.buffer.shape[0], dest_obj=self.buffer + ) + # ts will be empty if no samples were pulled, a list of timestamps otherwise + if ts: + ts = np.asarray(ts) + y = self.buffer[0 : ts.size, :] + this_x = None + old_offset = 0 + new_offset = 0 + for ch_ix in range(self.channel_count): + # we don't pull an entire screen's worth of data, so we have to + # trim the old data and append the new data to it + old_x, old_y = self.curves[ch_ix].getData() + # the timestamps are identical for all channels, so we need to do + # this calculation only once + if ch_ix == 0: + # find the index of the first sample that's still visible, + # i.e. newer than the left border of the plot + old_offset = old_x.searchsorted(plot_time) + # same for the new data, in case we pulled more data than + # can be shown at once + new_offset = ts.searchsorted(plot_time) + # append new timestamps to the trimmed old timestamps + this_x = np.hstack((old_x[old_offset:], ts[new_offset:])) + # append new data to the trimmed old data + this_y = np.hstack((old_y[old_offset:], y[new_offset:, ch_ix] - ch_ix)) + # replace the old data + self.curves[ch_ix].setData(this_x, this_y) + + +class MarkerInlet(Inlet): + """A MarkerInlet shows events that happen sporadically as vertical lines""" + + def __init__(self, info: pylsl.StreamInfo): + super().__init__(info) + + def pull_and_plot(self, plot_time, plt): + # TODO: purge old markers + strings, timestamps = self.inlet.pull_chunk(0) + if timestamps: + for string, ts in zip(strings, timestamps): + plt.addItem( + pg.InfiniteLine(ts, angle=90, movable=False, label=string[0]) + ) + + +def main(): + # firstly resolve all streams that could be shown + inlets: List[Inlet] = [] + print("looking for streams") + streams = pylsl.resolve_streams() + + # Create the pyqtgraph window + pw = pg.plot(title="LSL Plot") + plt = pw.getPlotItem() + plt.enableAutoRange(x=False, y=True) + + # iterate over found streams, creating specialized inlet objects that will + # handle plotting the data + for info in streams: + if info.type() == "Markers": + if ( + info.nominal_srate() != pylsl.IRREGULAR_RATE + or info.channel_format() != pylsl.cf_string + ): + print("Invalid marker stream " + info.name()) + print("Adding marker inlet: " + info.name()) + inlets.append(MarkerInlet(info)) + elif ( + info.nominal_srate() != pylsl.IRREGULAR_RATE + and info.channel_format() != pylsl.cf_string + ): + print("Adding data inlet: " + info.name()) + inlets.append(DataInlet(info, plt)) + else: + print("Don't know what to do with stream " + info.name()) + + def scroll(): + """Move the view so the data appears to scroll""" + # We show data only up to a timepoint shortly before the current time + # so new data doesn't suddenly appear in the middle of the plot + fudge_factor = pull_interval * 0.002 + plot_time = pylsl.local_clock() + pw.setXRange(plot_time - plot_duration + fudge_factor, plot_time - fudge_factor) + + def update(): + # Read data from the inlet. Use a timeout of 0.0 so we don't block GUI interaction. + mintime = pylsl.local_clock() - plot_duration + # call pull_and_plot for each inlet. + # Special handling of inlet types (markers, continuous data) is done in + # the different inlet classes. + for inlet in inlets: + inlet.pull_and_plot(mintime, plt) + + # create a timer that will move the view every update_interval ms + update_timer = QtCore.QTimer() + update_timer.timeout.connect(scroll) + update_timer.start(update_interval) + + # create a timer that will pull and add new data occasionally + pull_timer = QtCore.QTimer() + pull_timer.timeout.connect(update) + pull_timer.start(pull_interval) + + import sys + + # Start Qt event loop unless running in interactive mode or using pyside. + if (sys.flags.interactive != 1) or not hasattr(QtCore, "PYQT_VERSION"): + QtGui.QGuiApplication.instance().exec_() + + +if __name__ == "__main__": + main() diff --git a/apps/pylsl/src/pylsl/examples/ReceiveData.py b/apps/pylsl/src/pylsl/examples/ReceiveData.py new file mode 100644 index 0000000..8615d21 --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/ReceiveData.py @@ -0,0 +1,22 @@ +"""Example program to show how to read a multi-channel time series from LSL.""" + +from pylsl import StreamInlet, resolve_byprop + + +def main(): + # first resolve an EEG stream on the lab network + print("looking for an EEG stream...") + streams = resolve_byprop("type", "EEG") + + # create a new inlet to read from the stream + inlet = StreamInlet(streams[0]) + + while True: + # get a new sample (you can also omit the timestamp part if you're not + # interested in it) + sample, timestamp = inlet.pull_sample() + print(timestamp, sample) + + +if __name__ == "__main__": + main() diff --git a/apps/pylsl/src/pylsl/examples/ReceiveDataInChunks.py b/apps/pylsl/src/pylsl/examples/ReceiveDataInChunks.py new file mode 100644 index 0000000..817846b --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/ReceiveDataInChunks.py @@ -0,0 +1,24 @@ +"""Example program to demonstrate how to read a multi-channel time-series +from LSL in a chunk-by-chunk manner (which is more efficient).""" + +from pylsl import StreamInlet, resolve_byprop + + +def main(): + # first resolve an EEG stream on the lab network + print("looking for an EEG stream...") + streams = resolve_byprop("type", "EEG") + + # create a new inlet to read from the stream + inlet = StreamInlet(streams[0]) + + while True: + # get a new sample (you can also omit the timestamp part if you're not + # interested in it) + chunk, timestamps = inlet.pull_chunk() + if timestamps: + print(timestamps, chunk) + + +if __name__ == "__main__": + main() diff --git a/apps/pylsl/src/pylsl/examples/ReceiveStringMarkers.py b/apps/pylsl/src/pylsl/examples/ReceiveStringMarkers.py new file mode 100644 index 0000000..78cc413 --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/ReceiveStringMarkers.py @@ -0,0 +1,22 @@ +"""Example program to demonstrate how to read string-valued markers from LSL.""" + +from pylsl import StreamInlet, resolve_byprop + + +def main(): + # first resolve a marker stream on the lab network + print("looking for a marker stream...") + streams = resolve_byprop("type", "Markers") + + # create a new inlet to read from the stream + inlet = StreamInlet(streams[0]) + + while True: + # get a new sample (you can also omit the timestamp part if you're not + # interested in it) + sample, timestamp = inlet.pull_sample() + print("got %s at time %s" % (sample[0], timestamp)) + + +if __name__ == "__main__": + main() diff --git a/apps/pylsl/src/pylsl/examples/SendData.py b/apps/pylsl/src/pylsl/examples/SendData.py new file mode 100644 index 0000000..f73241d --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/SendData.py @@ -0,0 +1,66 @@ +"""Example program to demonstrate how to send a multi-channel time series to +LSL.""" + +import getopt +import sys +import time +from random import random as rand + +from pylsl import StreamInfo, StreamOutlet, local_clock + + +def main(argv): + srate = 100 + name = "BioSemi" + type = "EEG" + n_channels = 8 + help_string = "SendData.py -s -n -t " + try: + opts, args = getopt.getopt( + argv, "hs:c:n:t:", longopts=["srate=", "channels=", "name=", "type"] + ) + except getopt.GetoptError: + print(help_string) + sys.exit(2) + for opt, arg in opts: + if opt == "-h": + print(help_string) + sys.exit() + elif opt in ("-s", "--srate"): + srate = float(arg) + elif opt in ("-c", "--channels"): + n_channels = int(arg) + elif opt in ("-n", "--name"): + name = arg + elif opt in ("-t", "--type"): + type = arg + + # first create a new stream info (here we set the name to BioSemi, + # the content-type to EEG, 8 channels, 100 Hz, and float-valued data) The + # last value would be the serial number of the device or some other more or + # less locally unique identifier for the stream as far as available (you + # could also omit it but interrupted connections wouldn't auto-recover) + info = StreamInfo(name, type, n_channels, srate, "float32", "myuid34234") + + # next make an outlet + outlet = StreamOutlet(info) + + print("now sending data...") + start_time = local_clock() + sent_samples = 0 + while True: + elapsed_time = local_clock() - start_time + required_samples = int(srate * elapsed_time) - sent_samples + for sample_ix in range(required_samples): + # make a new random n_channels sample; this is converted into a + # pylsl.vectorf (the data type that is expected by push_sample) + mysample = [rand() for _ in range(n_channels)] + # now send it + outlet.push_sample(mysample) + sent_samples += required_samples + # now send it and wait for a bit before trying again. + time.sleep(0.01) + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/apps/pylsl/src/pylsl/examples/SendDataAdvanced.py b/apps/pylsl/src/pylsl/examples/SendDataAdvanced.py new file mode 100644 index 0000000..f4f3dbc --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/SendDataAdvanced.py @@ -0,0 +1,113 @@ +"""Example program to demonstrate how to send a multi-channel time-series +with proper meta-data to LSL.""" + +import argparse +import time +from random import random as rand + +import pylsl + + +def main(name="LSLExampleAmp", stream_type="EEG", srate=100): + channel_names = ["Fp1", "Fp2", "C3", "C4", "Cz", "P3", "P4", "Pz", "O1", "O2"] + channel_locations = [ + [-0.0307, 0.0949, -0.0047], + [0.0307, 0.0949, -0.0047], + [-0.0742, 4.54343962e-18, 0.0668], + [0.0743, 4.54956286e-18, 0.0669], + [0, 6.123234e-18, 0.1], + [-0.0567, -0.0677, 0.0469], + [0.0566, -0.0677, 0.0469], + [8.74397815e-18, -0.0714, 0.0699], + [-0.0307, -0.0949, -0.0047], + [0.0307, -0.0949, -0.0047], + ] + n_channels = len(channel_names) + + # First create a new stream info. + # The first 4 arguments are stream name, stream type, number of channels, and + # sampling rate -- all parameterized by the keyword arguments or the channel list above. + # The 5th parameter is the data format. This should match the origin format (unless the + # data will be transformed prior to pushing, then it should match the transformed-to format). + # Possible values are "float32", "double64", "string", "int32", "int16", "int8", or "int64". + # Alternatively, one could use the constants in the pylsl namespace beginning with `cf_`. + # i.e., cf_float32, cf_double64, etc. + # For this example, we will always use float32 data so we provide that as the 5th parameter. + # The last value would be the serial number of the device or some other more or + # less locally unique identifier for the stream as far as available (you + # could also omit it but interrupted connections wouldn't auto-recover). + info = pylsl.StreamInfo( + name, stream_type, n_channels, srate, "float32", "myuid2424" + ) + + # append some meta-data + # https://github.com/sccn/xdf/wiki/EEG-Meta-Data + info.desc().append_child_value("manufacturer", "LSLExampleAmp") + chns = info.desc().append_child("channels") + for chan_ix, label in enumerate(channel_names): + ch = chns.append_child("channel") + ch.append_child_value("label", label) + ch.append_child_value("unit", "microvolts") + ch.append_child_value("type", "EEG") + ch.append_child_value("scaling_factor", "1") + loc = ch.append_child("location") + for ax_str, pos in zip(["X", "Y", "Z"], channel_locations[chan_ix]): + loc.append_child_value(ax_str, str(pos)) + cap = info.desc().append_child("cap") + cap.append_child_value("name", "ComfyCap") + cap.append_child_value("size", "54") + cap.append_child_value("labelscheme", "10-20") + + # next make an outlet; we set the transmission chunk size to 32 samples + # and the outgoing buffer size to 360 seconds (max.) + outlet = pylsl.StreamOutlet(info, 32, 360) + + if False: + # It's unnecessary to check the info when the stream was created in the same scope; just use info. + # Use this code only as a sanity check if you think something when wrong during stream creation. + check_info = outlet.get_info() + assert check_info.name() == name + assert check_info.type() == stream_type + assert check_info.channel_count() == len(channel_names) + assert check_info.channel_format() == pylsl.cf_float32 + assert check_info.nominal_srate() == srate + + print("now sending data...") + start_time = pylsl.local_clock() + sent_samples = 0 + while True: + elapsed_time = pylsl.local_clock() - start_time + required_samples = int(srate * elapsed_time) - sent_samples + if required_samples > 0: + # make a chunk==array of length required_samples, where each element in the array + # is a new random n_channels sample vector + mychunk = [ + [rand() for chan_ix in range(n_channels)] + for samp_ix in range(required_samples) + ] + # Get a time stamp in seconds. We pretend that our samples are actually + # 125ms old, e.g., as if coming from some external hardware with known latency. + stamp = pylsl.local_clock() - 0.125 + # now send it and wait for a bit + # Note that even though `rand()` returns a 64-bit value, the `push_chunk` method + # will convert it to c_float before passing the data to liblsl. + outlet.push_chunk(mychunk, stamp) + sent_samples += required_samples + time.sleep(0.02) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument( + "--name", default="LSLExampleAmp", help="Name of the created stream." + ) + parser.add_argument("--type", default="EEG", help="Type of the created stream.") + parser.add_argument( + "--srate", + default=100.0, + help="Sampling rate of the created stream.", + type=float, + ) + arg = parser.parse_args() + + main(name=arg.name, stream_type=arg.type, srate=arg.srate) diff --git a/apps/pylsl/src/pylsl/examples/SendStringMarkers.py b/apps/pylsl/src/pylsl/examples/SendStringMarkers.py new file mode 100644 index 0000000..21eeab8 --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/SendStringMarkers.py @@ -0,0 +1,32 @@ +"""Example program to demonstrate how to send string-valued markers into LSL.""" + +import random +import time + +from pylsl import StreamInfo, StreamOutlet + + +def main(): + # first create a new stream info (here we set the name to MyMarkerStream, + # the content-type to Markers, 1 channel, irregular sampling rate, + # and string-valued data) The last value would be the locally unique + # identifier for the stream as far as available, e.g. + # program-scriptname-subjectnumber (you could also omit it but interrupted + # connections wouldn't auto-recover). The important part is that the + # content-type is set to 'Markers', because then other programs will know how + # to interpret the content + info = StreamInfo("MyMarkerStream", "Markers", 1, 0, "string", "myuidw43536") + + # next make an outlet + outlet = StreamOutlet(info) + + print("now sending markers...") + markernames = ["Test", "Blah", "Marker", "XXX", "Testtest", "Test-1-2-3"] + while True: + # pick a sample to send an wait for a bit + outlet.push_sample([random.choice(markernames)]) + time.sleep(random.random() * 3) + + +if __name__ == "__main__": + main() diff --git a/apps/pylsl/src/pylsl/examples/__init__.py b/apps/pylsl/src/pylsl/examples/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/apps/pylsl/src/pylsl/examples/__init__.py @@ -0,0 +1 @@ + diff --git a/apps/pylsl/src/pylsl/info.py b/apps/pylsl/src/pylsl/info.py new file mode 100644 index 0000000..a84307a --- /dev/null +++ b/apps/pylsl/src/pylsl/info.py @@ -0,0 +1,619 @@ +import ctypes +import typing + +from .lib import lib, string2fmt, cf_float32, _stream_security_api_available +from .util import IRREGULAR_RATE + + +class StreamInfo: + """The StreamInfo object stores the declaration of a data stream. + + Represents the following information: + a) stream data format (#channels, channel format) + b) core information (stream name, content type, sampling rate) + c) optional meta-data about the stream content (channel labels, + measurement units, etc.) + + Whenever a program wants to provide a new stream on the lab network it will + typically first create a StreamInfo to describe its properties and then + construct a StreamOutlet with it to create the stream on the network. + Recipients who discover the outlet can query the StreamInfo; it is also + written to disk when recording the stream (playing a similar role as a file + header). + + """ + + def __init__( + self, + name: str = "untitled", + type: str = "", + channel_count: int = 1, + nominal_srate: float = IRREGULAR_RATE, + channel_format: int = cf_float32, + source_id: typing.Optional[str] = None, + handle=None, + ): + """Construct a new StreamInfo object. + + Core stream information is specified here. Any remaining meta-data can + be added later. + + Keyword arguments: + name -- Name of the stream. Describes the device (or product series) + that this stream makes available (for use by programs, + experimenters or data analysts). Cannot be empty. + type -- Content type of the stream. By convention LSL uses the content + types defined in the XDF file format specification where + applicable (https://github.com/sccn/xdf). The content type is the + preferred way to find streams (as opposed to searching by name). + channel_count -- Number of channels per sample. This stays constant for + the lifetime of the stream. (default 1) + nominal_srate -- The sampling rate (in Hz) as advertised by the data + source, regular (otherwise set to IRREGULAR_RATE). + (default IRREGULAR_RATE) + channel_format -- Format/type of each channel. If your channels have + different formats, consider supplying multiple + streams or use the largest type that can hold + them all (such as cf_double64). It is also allowed + to pass this as a string, without the cf_ prefix, + e.g., 'float32' (default cf_float32) + source_id -- Unique identifier of the device or source of the data, if + available (such as the serial number). This is critical + for system robustness since it allows recipients to + recover from failure even after the serving app, device or + computer crashes (just by finding a stream with the same + source id on the network again). If the provided value is None + then a source id will be generated automatically from a hash of + the other arguments. If recovery is not desired, for example + when a disconnection should raise an error, set the source_id + to "" (empty string) . (default None) + """ + if handle is not None: + self.obj = ctypes.c_void_p(handle) + else: + if isinstance(channel_format, str): + channel_format = string2fmt[channel_format] + if source_id is None: + source_id = str( + hash((name, type, channel_count, nominal_srate, channel_format)) + ) + print( + f"Generated source_id: '{source_id}' for StreamInfo with name '{name}', type '{type}', " + f"channel_count {channel_count}, nominal_srate {nominal_srate}, " + f"and channel_format {channel_format}." + ) + self.obj = lib.lsl_create_streaminfo( + ctypes.c_char_p(str.encode(name)), + ctypes.c_char_p(str.encode(type)), + channel_count, + ctypes.c_double(nominal_srate), + channel_format, + ctypes.c_char_p(str.encode(source_id)), + ) + self.obj = ctypes.c_void_p(self.obj) + if not self.obj: + raise RuntimeError("could not create stream description " "object.") + + def __del__(self): + """Destroy a previously created StreamInfo object.""" + # noinspection PyBroadException + try: + lib.lsl_destroy_streaminfo(self.obj) + except Exception as e: + print(f"StreamInfo deletion triggered error: {e}") + + def __str__(self): + return ( + f"" + ) + + def __repr__(self): + return self.__str__() + + # === Core Information (assigned at construction) === + + def name(self) -> str: + """Name of the stream. + + This is a human-readable name. For streams offered by device modules, + it refers to the type of device or product series that is generating + the data of the stream. If the source is an application, the name may + be a more generic or specific identifier. Multiple streams with the + same name can coexist, though potentially at the cost of ambiguity (for + the recording app or experimenter). + + """ + return lib.lsl_get_name(self.obj).decode("utf-8") + + def type(self) -> str: + """Content type of the stream. + + The content type is a short string such as "EEG", "Gaze" which + describes the content carried by the channel (if known). If a stream + contains mixed content this value need not be assigned but may instead + be stored in the description of channel types. To be useful to + applications and automated processing systems using the recommended + content types is preferred. + + """ + return lib.lsl_get_type(self.obj).decode("utf-8") + + def channel_count(self) -> int: + """Number of channels of the stream. + + A stream has at least one channel; the channel count stays constant for + all samples. + + """ + return lib.lsl_get_channel_count(self.obj) + + def nominal_srate(self) -> float: + """Sampling rate of the stream, according to the source (in Hz). + + If a stream is irregularly sampled, this should be set to + IRREGULAR_RATE. + + Note that no data will be lost even if this sampling rate is incorrect + or if a device has temporary hiccups, since all samples will be + transmitted anyway (except for those dropped by the device itself). + However, when the recording is imported into an application, a good + data importer may correct such errors more accurately if the advertised + sampling rate was close to the specs of the device. + + """ + return lib.lsl_get_nominal_srate(self.obj) + + def channel_format(self) -> int: + """Channel format of the stream. + + All channels in a stream have the same format. However, a device might + offer multiple time-synched streams each with its own format. + + """ + return lib.lsl_get_channel_format(self.obj) + + def source_id(self) -> str: + """Unique identifier of the stream's source, if available. + + The unique source (or device) identifier is an optional piece of + information that, if available, allows that endpoints (such as the + recording program) can re-acquire a stream automatically once it is + back online. + + """ + return lib.lsl_get_source_id(self.obj).decode("utf-8") + + # === Hosting Information (assigned when bound to an outlet/inlet) === + + def version(self): + """Protocol version used to deliver the stream.""" + return lib.lsl_get_version(self.obj) + + def created_at(self): + """Creation time stamp of the stream. + + This is the time stamp when the stream was first created + (as determined via local_clock() on the providing machine). + + """ + return lib.lsl_get_created_at(self.obj) + + def uid(self) -> str: + """Unique ID of the stream outlet instance (once assigned). + + This is a unique identifier of the stream outlet, and is guaranteed to + be different across multiple instantiations of the same outlet (e.g., + after a re-start). + + """ + return lib.lsl_get_uid(self.obj).decode("utf-8") + + def session_id(self) -> str: + """Session ID for the given stream. + + The session id is an optional human-assigned identifier of the + recording session. While it is rarely used, it can be used to prevent + concurrent recording activities on the same sub-network (e.g., in + multiple experiment areas) from seeing each other's streams + (can be assigned in a configuration file read by liblsl, see also + Network Connectivity in the LSL wiki). + + """ + return lib.lsl_get_session_id(self.obj).decode("utf-8") + + def hostname(self) -> str: + """Hostname of the providing machine.""" + return lib.lsl_get_hostname(self.obj).decode("utf-8") + + # === Security Information (assigned when bound to an outlet/inlet) === + + def security_enabled(self) -> bool: + """Check if the stream has security/encryption enabled. + + When security is enabled, all data transmitted on this stream is + encrypted using ChaCha20-Poly1305 authenticated encryption. + + Returns True if security is enabled, False otherwise. + Returns False if the security API is not available. + """ + if not _stream_security_api_available: + return False + result = lib.lsl_get_security_enabled(self.obj) + return result == 1 + + def security_fingerprint(self) -> str: + """Get the security fingerprint of the stream's public key. + + The fingerprint is a SHA256 hash of the device's Ed25519 public key, + formatted as "SHA256:xxxx...". This can be used to verify device identity. + + Returns the fingerprint string, or empty string if security is not + enabled or the security API is not available. + """ + if not _stream_security_api_available: + return "" + result = lib.lsl_get_security_fingerprint(self.obj) + if result: + return result.decode("utf-8") + return "" + + # === Data Description (can be modified) === + def desc(self) -> "XMLElement": + """Extended description of the stream. + + It is highly recommended that at least the channel labels are described + here. See code examples on the LSL wiki. Other information, such + as amplifier settings, measurement units if deviating from defaults, + setup information, subject information, etc., can be specified here, as + well. Meta-data recommendations follow the XDF file format project + (github.com/sccn/xdf/wiki/Meta-Data or web search for: XDF meta-data). + + Important: if you use a stream content type for which meta-data + recommendations exist, please try to lay out your meta-data in + agreement with these recommendations for compatibility with other + applications. + + """ + return XMLElement(lib.lsl_get_desc(self.obj)) + + def as_xml(self) -> str: + """Retrieve the entire stream_info in XML format. + + This yields an XML document (in string form) whose top-level element is + . The description element contains one element for each + field of the stream_info class, including: + a) the core elements , , , , + , + b) the misc elements , , , , + , , , , + , + c) the extended description element with user-defined + sub-elements. + + """ + return lib.lsl_get_xml(self.obj).decode("utf-8") + + def get_channel_labels(self) -> typing.Optional[list[typing.Optional[str]]]: + """Get the channel names in the description. + + Returns + ------- + labels : list of str or ``None`` | None + List of channel names, matching the number of total channels. + If ``None``, the channel names are not set. + + .. warning:: + + If a list of str and ``None`` are returned, some of the channel names + are missing. This is not expected and could occur if the XML tree in + the ``desc`` property is tempered with outside of the defined getter and + setter. + """ + return self._get_channel_info("label") + + def get_channel_types(self) -> typing.Optional[list[typing.Optional[str]]]: + """Get the channel types in the description. + + Returns + ------- + types : list of str or ``None`` | None + List of channel types, matching the number of total channels. + If ``None``, the channel types are not set. + + .. warning:: + + If a list of str and ``None`` are returned, some of the channel types + are missing. This is not expected and could occur if the XML tree in + the ``desc`` property is tempered with outside of the defined getter and + setter. + """ + return self._get_channel_info("type") + + def get_channel_units(self) -> typing.Optional[list[typing.Optional[str]]]: + """Get the channel units in the description. + + Returns + ------- + units : list of str or ``None`` | None + List of channel units, matching the number of total channels. + If ``None``, the channel units are not set. + + .. warning:: + + If a list of str and ``None`` are returned, some of the channel units + are missing. This is not expected and could occur if the XML tree in + the ``desc`` property is tempered with outside of the defined getter and + setter. + """ + return self._get_channel_info("unit") + + def _get_channel_info(self, name) -> typing.Optional[list[typing.Optional[str]]]: + """Get the 'channel/name' element in the XML tree.""" + if self.desc().child("channels").empty(): + return None + ch_infos = list() + channels = self.desc().child("channels") + ch = channels.child("channel") + while not ch.empty(): + ch_info = ch.child(name).first_child().value() + if len(ch_info) != 0: + ch_infos.append(ch_info) + else: + ch_infos.append(None) + ch = ch.next_sibling() + if all(ch_info is None for ch_info in ch_infos): + return None + if len(ch_infos) != self.channel_count(): + print( + f"The stream description contains {len(ch_infos)} elements for " + f"{self.channel_count()} channels.", + ) + return ch_infos + + def set_channel_labels(self, labels: list[str]): + """Set the channel names in the description. Existing labels are overwritten. + + Parameters + ---------- + labels : list of str + List of channel names, matching the number of total channels. + """ + self._set_channel_info(labels, "label") + + def set_channel_types(self, types: typing.Union[str, list[str]]): + """Set the channel types in the description. Existing types are overwritten. + + The types are given as human-readable strings, e.g. ``'eeg'``. + + Parameters + ---------- + types : list of str | str + List of channel types, matching the number of total channels. + If a single `str` is provided, the type is applied to all channels. + """ + types = [types] * self.channel_count() if isinstance(types, str) else types + self._set_channel_info(types, "type") + + def set_channel_units( + self, units: typing.Union[str, int, list[typing.Union[str, int]]] + ) -> None: + """Set the channel units in the description. Existing units are overwritten. + + The units are given as human-readable strings, e.g. ``'microvolts'``, or as + multiplication factor, e.g. ``-6`` for ``1e-6`` thus converting e.g. Volts to + microvolts. + + Parameters + ---------- + units : list of str | list of int | array of int | str | int + List of channel units, matching the number of total channels. + If a single `str` or `int` is provided, the unit is applied to all channels. + + Notes + ----- + Some channel types do not have a unit. The `str` ``none`` or the `int` 0 should + be used to denote this channel unit, corresponding to ``FIFF_UNITM_NONE`` in + MNE-Python. + """ + if isinstance(units, (int, str)): + units = [units] * self.channel_count() + else: # iterable + units = [ + str(int(unit)) if isinstance(unit, int) else unit for unit in units + ] + self._set_channel_info(units, "unit") + + def _set_channel_info(self, ch_infos, name: str) -> None: + """Set the 'channel/name' element in the XML tree.""" + if len(ch_infos) != self.channel_count(): + raise ValueError( + f"The number of provided channel {name} {len(ch_infos)} " + f"must match the number of channels {self.channel_count()}." + ) + + channels = StreamInfo._add_first_node(self.desc, "channels") + # fill the 'channel/name' element of the tree and overwrite existing values + ch = channels.child("channel") + for ch_info in ch_infos: + ch = channels.append_child("channel") if ch.empty() else ch + StreamInfo._set_description_node(ch, {name: ch_info}) + ch = ch.next_sibling() + StreamInfo._prune_description_node(ch, channels) + + # -- Helper methods to interact with the XMLElement tree --------------------------- + @staticmethod + def _add_first_node(desc, name: str) -> "XMLElement": + """Add the first node in the description and return it.""" + if desc().child(name).empty(): + node = desc().append_child(name) + else: + node = desc().child(name) + return node + + @staticmethod + def _prune_description_node(node, parent): + """Prune a node and remove outdated entries.""" + # this is useful in case the sinfo is tempered with and had more entries of type + # 'node' than it should. + while not node.empty(): + node_next = node.next_sibling() + parent.remove_child(node) + node = node_next + + @staticmethod + def _set_description_node(node, mapping): + """Set the key: value child(s) of a node.""" + for key, value in mapping.items(): + value = str(int(value)) if isinstance(value, int) else str(value) + if node.child(key).empty(): + node.append_child_value(key, value) + else: + node.child(key).first_child().set_value(value) + + +class XMLElement: + """A lightweight XML element tree modeling the .desc() field of StreamInfo. + + Has a name and can have multiple named children or have text content as + value; attributes are omitted. Insider note: The interface is modeled after + a subset of pugixml's node type and is compatible with it. See also + http://pugixml.googlecode.com/svn/tags/latest/docs/manual/access.html for + additional documentation. + + """ + + def __init__(self, handle): + """Construct new XML element from existing handle.""" + self.e = ctypes.c_void_p(handle) + + # === Tree Navigation === + + def first_child(self) -> "XMLElement": + """Get the first child of the element.""" + return XMLElement(lib.lsl_first_child(self.e)) + + def last_child(self) -> "XMLElement": + """Get the last child of the element.""" + return XMLElement(lib.lsl_last_child(self.e)) + + def child(self, name: str) -> "XMLElement": + """Get a child with a specified name.""" + return XMLElement(lib.lsl_child(self.e, str.encode(name))) + + def next_sibling(self, name: typing.Optional[str] = None) -> "XMLElement": + """Get the next sibling in the children list of the parent node. + + If a name is provided, the next sibling with the given name is returned. + + """ + if name is None: + return XMLElement(lib.lsl_next_sibling(self.e)) + else: + return XMLElement(lib.lsl_next_sibling_n(self.e, str.encode(name))) + + def previous_sibling(self, name: typing.Optional[str] = None) -> "XMLElement": + """Get the previous sibling in the children list of the parent node. + + If a name is provided, the previous sibling with the given name is + returned. + + """ + if name is None: + return XMLElement(lib.lsl_previous_sibling(self.e)) + else: + return XMLElement(lib.lsl_previous_sibling_n(self.e, str.encode(name))) + + def parent(self) -> "XMLElement": + """Get the parent node.""" + return XMLElement(lib.lsl_parent(self.e)) + + # === Content Queries === + + def empty(self) -> bool: + """Whether this node is empty.""" + return bool(lib.lsl_empty(self.e)) + + def is_text(self) -> bool: + """Whether this is a text body (instead of an XML element). + + True both for plain char data and CData. + + """ + return bool(lib.lsl_is_text(self.e)) + + def name(self) -> str: + """Name of the element.""" + return lib.lsl_name(self.e).decode("utf-8") + + def value(self) -> str: + """Value of the element.""" + return lib.lsl_value(self.e).decode("utf-8") + + def child_value(self, name: typing.Optional[str] = None) -> str: + """Get child value (value of the first child that is text). + + If a name is provided, then the value of the first child with the + given name is returned. + + """ + if name is None: + res = lib.lsl_child_value(self.e) + else: + res = lib.lsl_child_value_n(self.e, str.encode(name)) + return res.decode("utf-8") + + # === Modification === + + def append_child_value(self, name: str, value: str) -> "XMLElement": + """Append a child node with a given name, which has a (nameless) + plain-text child with the given text value.""" + return XMLElement( + lib.lsl_append_child_value(self.e, str.encode(name), str.encode(value)) + ) + + def prepend_child_value(self, name: str, value: str) -> "XMLElement": + """Prepend a child node with a given name, which has a (nameless) + plain-text child with the given text value.""" + return XMLElement( + lib.lsl_prepend_child_value(self.e, str.encode(name), str.encode(value)) + ) + + def set_child_value(self, name: str, value: str) -> "XMLElement": + """Set the text value of the (nameless) plain-text child of a named + child node.""" + return XMLElement( + lib.lsl_set_child_value(self.e, str.encode(name), str.encode(value)) + ) + + def set_name(self, name: str) -> bool: + """Set the element's name. Returns False if the node is empty.""" + return bool(lib.lsl_set_name(self.e, str.encode(name))) + + def set_value(self, value: str) -> bool: + """Set the element's value. Returns False if the node is empty.""" + return bool(lib.lsl_set_value(self.e, str.encode(value))) + + def append_child(self, name: str) -> "XMLElement": + """Append a child element with the specified name.""" + return XMLElement(lib.lsl_append_child(self.e, str.encode(name))) + + def prepend_child(self, name: str) -> "XMLElement": + """Prepend a child element with the specified name.""" + return XMLElement(lib.lsl_prepend_child(self.e, str.encode(name))) + + def append_copy(self, elem: "XMLElement") -> "XMLElement": + """Append a copy of the specified element as a child.""" + return XMLElement(lib.lsl_append_copy(self.e, elem.e)) + + def prepend_copy(self, elem: "XMLElement") -> "XMLElement": + """Prepend a copy of the specified element as a child.""" + return XMLElement(lib.lsl_prepend_copy(self.e, elem.e)) + + def remove_child(self, rhs: "XMLElement") -> None: + """Remove a given child element, specified by name or as element.""" + if type(rhs) is XMLElement: + lib.lsl_remove_child(self.e, rhs.e) + else: + lib.lsl_remove_child_n(self.e, rhs) diff --git a/apps/pylsl/src/pylsl/inlet.py b/apps/pylsl/src/pylsl/inlet.py new file mode 100644 index 0000000..472733b --- /dev/null +++ b/apps/pylsl/src/pylsl/inlet.py @@ -0,0 +1,309 @@ +import ctypes + +from .lib import lib, fmt2type, fmt2pull_sample, fmt2pull_chunk, cf_string +from .util import handle_error, FOREVER +from .info import StreamInfo + + +def free_char_p_array_memory(char_p_array, num_elements): + pointers = ctypes.cast(char_p_array, ctypes.POINTER(ctypes.c_void_p)) + for p in range(num_elements): + if pointers[p] is not None: # only free initialized pointers + lib.lsl_destroy_string(pointers[p]) + + +class StreamInlet: + """A stream inlet. + + Inlets are used to receive streaming data (and meta-data) from the lab + network. + + """ + + def __init__( + self, info, max_buflen=360, max_chunklen=0, recover=True, processing_flags=0 + ): + """Construct a new stream inlet from a resolved stream description. + + Keyword arguments: + description -- A resolved stream description object (as coming from one + of the resolver functions). Note: the stream_inlet may also be + constructed with a fully-specified stream_info, if the desired + channel format and count is already known up-front, but this is + strongly discouraged and should only ever be done if there is + no time to resolve the stream up-front (e.g., due to + limitations in the client program). + max_buflen -- Optionally the maximum amount of data to buffer (in + seconds if there is a nominal sampling rate, otherwise + x100 in samples). Recording applications want to use a + fairly large buffer size here, while real-time + applications would only buffer as much as they need to + perform their next calculation. (default 360) + max_chunklen -- Optionally the maximum size, in samples, at which + chunks are transmitted (the default corresponds to the + chunk sizes used by the sender). Recording programs + can use a generous size here (leaving it to the network + how to pack things), while real-time applications may + want a finer (perhaps 1-sample) granularity. If left + unspecified (=0), the sender determines the chunk + granularity. (default 0) + recover -- Try to silently recover lost streams that are recoverable + (=those that that have a source_id set). In all other cases + (recover is False or the stream is not recoverable) + functions may throw a lost_error if the stream's source is + lost (e.g., due to an app or computer crash). (default True) + processing_flags -- Post-processing options. Use one of the post-processing + flags `proc_none`, `proc_clocksync`, `proc_dejitter`, `proc_monotonize`, + or `proc_threadsafe`. Can also be a logical OR combination of multiple + flags. Use `proc_ALL` for all flags. (default proc_none). + """ + if type(info) is list: + raise TypeError( + "description needs to be of type StreamInfo, " "got a list." + ) + self.obj = lib.lsl_create_inlet(info.obj, max_buflen, max_chunklen, recover) + self.obj = ctypes.c_void_p(self.obj) + if not self.obj: + raise RuntimeError("could not create stream inlet.") + if processing_flags > 0: + handle_error(lib.lsl_set_postprocessing(self.obj, processing_flags)) + self.channel_format = info.channel_format() + self.channel_count = info.channel_count() + self.do_pull_sample = fmt2pull_sample[self.channel_format] + self.do_pull_chunk = fmt2pull_chunk[self.channel_format] + self.value_type = fmt2type[self.channel_format] + self.sample_type = self.value_type * self.channel_count + self.sample = self.sample_type() + self.buffers = {} + + def __del__(self): + """Destructor. The inlet will automatically disconnect if destroyed.""" + # noinspection PyBroadException + try: + lib.lsl_destroy_inlet(self.obj) + except Exception: + pass + + def info(self, timeout=FOREVER): + """Retrieve the complete information of the given stream. + + This includes the extended description. Can be invoked at any time of + the stream's lifetime. + + Keyword arguments: + timeout -- Timeout of the operation. (default FOREVER) + + Throws a TimeoutError (if the timeout expires), or LostError (if the + stream source has been lost). + + """ + errcode = ctypes.c_int() + result = lib.lsl_get_fullinfo( + self.obj, ctypes.c_double(timeout), ctypes.byref(errcode) + ) + handle_error(errcode) + return StreamInfo(handle=result) + + def open_stream(self, timeout=FOREVER): + """Subscribe to the data stream. + + All samples pushed in at the other end from this moment onwards will be + queued and eventually be delivered in response to pull_sample() or + pull_chunk() calls. Pulling a sample without some preceding open_stream + is permitted (the stream will then be opened implicitly). + + Keyword arguments: + timeout -- Optional timeout of the operation (default FOREVER). + + Throws a TimeoutError (if the timeout expires), or LostError (if the + stream source has been lost). + + """ + errcode = ctypes.c_int() + lib.lsl_open_stream(self.obj, ctypes.c_double(timeout), ctypes.byref(errcode)) + handle_error(errcode) + + def close_stream(self): + """Drop the current data stream. + + All samples that are still buffered or in flight will be dropped and + transmission and buffering of data for this inlet will be stopped. If + an application stops being interested in data from a source + (temporarily or not) but keeps the outlet alive, it should call + lsl_close_stream() to not waste unnecessary system and network + resources. + + """ + lib.lsl_close_stream(self.obj) + + def time_correction(self, timeout=FOREVER): + """Retrieve an estimated time correction offset for the given stream. + + The first call to this function takes several milliseconds until a + reliable first estimate is obtained. Subsequent calls are instantaneous + (and rely on periodic background updates). The precision of these + estimates should be below 1 ms (empirically within +/-0.2 ms). + + Keyword arguments: + timeout -- Timeout to acquire the first time-correction estimate + (default FOREVER). + + Returns the current time correction estimate. This is the number that + needs to be added to a time stamp that was remotely generated via + local_clock() to map it into the local clock domain of this + machine. + + Throws a TimeoutError (if the timeout expires), or LostError (if the + stream source has been lost). + + """ + errcode = ctypes.c_int() + result = lib.lsl_time_correction( + self.obj, ctypes.c_double(timeout), ctypes.byref(errcode) + ) + handle_error(errcode) + return result + + def pull_sample(self, timeout=FOREVER, sample=None): + """Pull a sample from the inlet and return it. + + Keyword arguments: + timeout -- The timeout for this operation, if any. (default FOREVER) + If this is passed as 0.0, then the function returns only a + sample if one is buffered for immediate pickup. + + Returns a tuple (sample,timestamp) where sample is a list of channel + values and timestamp is the capture time of the sample on the remote + machine, or (None,None) if no new sample was available. To remap this + time stamp to the local clock, add the value returned by + .time_correction() to it. + + Throws a LostError if the stream source has been lost. Note that, if + the timeout expires, no TimeoutError is thrown (because this case is + not considered an error). + + """ + + # support for the legacy API + if type(timeout) is list: + assign_to = timeout + timeout = sample if type(sample) is float else 0.0 + else: + assign_to = None + + errcode = ctypes.c_int() + timestamp = self.do_pull_sample( + self.obj, + ctypes.byref(self.sample), + self.channel_count, + ctypes.c_double(timeout), + ctypes.byref(errcode), + ) + handle_error(errcode) + if timestamp: + sample = [v for v in self.sample] + if self.channel_format == cf_string: + sample = [v.decode("utf-8") for v in sample] + if assign_to is not None: + assign_to[:] = sample + return sample, timestamp + else: + return None, None + + def pull_chunk(self, timeout=0.0, max_samples=1024, dest_obj=None): + """Pull a chunk of samples from the inlet. + + Keyword arguments: + timeout -- The timeout of the operation; if passed as 0.0, then only + samples available for immediate pickup will be returned. + (default 0.0) + max_samples -- Maximum number of samples to return. (default + 1024) + dest_obj -- A Python object that supports the buffer interface. + If this is provided then the dest_obj will be updated in place + and the samples list returned by this method will be empty. + It is up to the caller to trim the buffer to the appropriate + number of samples. + A numpy buffer must be order='C' + (default None) + + Returns a tuple (samples,timestamps) where samples is a list of samples + (each itself a list of values), and timestamps is a list of time-stamps. + + Throws a LostError if the stream source has been lost. + + """ + # look up a pre-allocated buffer of appropriate length + num_channels = self.channel_count + max_values = max_samples * num_channels + + if max_samples not in self.buffers: + # noinspection PyCallingNonCallable + self.buffers[max_samples] = ( + (self.value_type * max_values)(), + (ctypes.c_double * max_samples)(), + ) + if dest_obj is not None: + data_buff = (self.value_type * max_values).from_buffer(dest_obj) + else: + data_buff = self.buffers[max_samples][0] + ts_buff = self.buffers[max_samples][1] + + # read data into it + errcode = ctypes.c_int() + # noinspection PyCallingNonCallable + num_elements = self.do_pull_chunk( + self.obj, + ctypes.byref(data_buff), + ctypes.byref(ts_buff), + ctypes.c_size_t(max_values), + ctypes.c_size_t(max_samples), + ctypes.c_double(timeout), + ctypes.byref(errcode), + ) + handle_error(errcode) + # return results (note: could offer a more efficient format in the + # future, e.g., a numpy array) + num_samples = num_elements / num_channels + if dest_obj is None: + samples = [ + [data_buff[s * num_channels + c] for c in range(num_channels)] + for s in range(int(num_samples)) + ] + if self.channel_format == cf_string: + samples = [[v.decode("utf-8") for v in s] for s in samples] + free_char_p_array_memory(data_buff, max_values) + else: + samples = None + timestamps = [ts_buff[s] for s in range(int(num_samples))] + return samples, timestamps + + def samples_available(self): + """Query whether samples are currently available for immediate pickup. + + Note that it is not a good idea to use samples_available() to determine + whether a pull_*() call would block: to be sure, set the pull timeout + to 0.0 or an acceptably low value. If the underlying implementation + supports it, the value will be the number of samples available + (otherwise it will be 1 or 0). + + """ + return lib.lsl_samples_available(self.obj) + + def flush(self): + """ + Drop all queued not-yet pulled samples. + :return: The number of dropped samples. + """ + return lib.lsl_inlet_flush(self.obj) + + def was_clock_reset(self): + """Query whether the clock was potentially reset since the last call. + + This is rarely-used function is only needed for applications that + combine multiple time_correction values to estimate precise clock + drift if they should tolerate cases where the source machine was + hot-swapped or restarted. + + """ + return bool(lib.lsl_was_clock_reset(self.obj)) diff --git a/apps/pylsl/src/pylsl/outlet.py b/apps/pylsl/src/pylsl/outlet.py new file mode 100644 index 0000000..e269645 --- /dev/null +++ b/apps/pylsl/src/pylsl/outlet.py @@ -0,0 +1,228 @@ +import ctypes + +from .lib import ( + lib, + fmt2push_sample, + fmt2push_chunk, + fmt2push_chunk_n, + fmt2type, + cf_string, +) +from .util import handle_error +from .info import StreamInfo + + +class StreamOutlet: + """A stream outlet. + + Outlets are used to make streaming data (and the meta-data) available on + the lab network. + + """ + + def __init__(self, info: StreamInfo, chunk_size: int = 0, max_buffered: int = 360): + """Establish a new stream outlet. This makes the stream discoverable. + + Keyword arguments: + description -- The StreamInfo object to describe this stream. Stays + constant over the lifetime of the outlet. + chunk_size --- Optionally the desired chunk granularity (in samples) + for transmission. If unspecified, each push operation + yields one chunk. Inlets can override this setting. + (default 0) + max_buffered -- Optionally the maximum amount of data to buffer (in + seconds if there is a nominal sampling rate, otherwise + x100 in samples). The default is 6 minutes of data. + Note that, for high-bandwidth data, you will want to + use a lower value here to avoid running out of RAM. + (default 360) + + """ + + """ + # If the source_id matches the default then we can assume it was created automatically. + # It may be desirable to include the host name in the source_id hash to avoid collisions. + # However, there are likely implications to re-creating the info so this is commented out + # until a need arises. + expected_src_id = str(hash(( + info.name(), info.type(), info.channel_count(), info.nominal_srate(), info.channel_format() + ))) + if info.source_id() == expected_src_id: + old_desc = info.desc() # save the old metadata + import socket + new_source_id = str(hash(( + info.name(), + info.type(), + info.channel_count(), + info.nominal_srate(), + info.channel_format(), + socket.gethostname() + ))) + info = StreamInfo( + name=info.name(), + type=info.type(), + channel_count=info.channel_count(), + nominal_srate=info.nominal_srate(), + channel_format=info.channel_format(), + source_id=new_source_id, + ) + # Add the old metadata to the new info object + new_desc_parent = info.desc().parent() + new_desc_parent.remove_child(info.desc()) + new_desc_parent.append_copy(old_desc) + """ + self.obj = lib.lsl_create_outlet(info.obj, chunk_size, max_buffered) + self.obj = ctypes.c_void_p(self.obj) + if not self.obj: + raise RuntimeError("could not create stream outlet.") + self.channel_format = info.channel_format() + self.channel_count = info.channel_count() + self.do_push_sample = fmt2push_sample[self.channel_format] + self.do_push_chunk = fmt2push_chunk[self.channel_format] + self.do_push_chunk_n = fmt2push_chunk_n[self.channel_format] + self.value_type = fmt2type[self.channel_format] + self.sample_type = self.value_type * self.channel_count + + def __del__(self): + """Destroy an outlet. + + The outlet will no longer be discoverable after destruction and all + connected inlets will stop delivering data. + + """ + # noinspection PyBroadException + try: + lib.lsl_destroy_outlet(self.obj) + except Exception as e: + print(f"StreamOutlet deletion triggered error: {e}") + + def push_sample(self, x, timestamp: float = 0.0, pushthrough: bool = True): + """Push a sample into the outlet. + + Each entry in the list corresponds to one channel. + + Keyword arguments: + x -- A list of values to push (one per channel). + timestamp -- Optionally the capture time of the sample, in agreement + with local_clock(); if 0.0, the current + time is used. (default 0.0) + pushthrough -- Whether to push the sample through to the receivers + instead of buffering it with subsequent samples. + Note that the chunk_size, if specified at outlet + construction, takes precedence over the pushthrough flag. + (default True) + + """ + if len(x) == self.channel_count: + if self.channel_format == cf_string: + x = [v.encode("utf-8") for v in x] + handle_error( + self.do_push_sample( + self.obj, + self.sample_type(*x), + ctypes.c_double(timestamp), + ctypes.c_int(pushthrough), + ) + ) + else: + raise ValueError( + "length of the sample (" + str(len(x)) + ") must " + "correspond to the stream's channel count (" + + str(self.channel_count) + + ")." + ) + + def push_chunk(self, x, timestamp: float = 0.0, pushthrough: bool = True): + """Push a list of samples into the outlet. + + samples -- A list of samples, preferably as a 2-D numpy array. + `samples` can also be a list of lists, or a list of + multiplexed values. + timestamp -- Optional, float or 1-D list of floats. + If float and != 0.0: the capture time of the most recent sample, in + agreement with local_clock(); if default (0.0), the current + time is used. The time stamps of other samples are + automatically derived according to the sampling rate of + the stream. + If list of floats: the time stamps for each sample. + Must be the same length as `samples`. + pushthrough Whether to push the chunk through to the receivers instead + of buffering it with subsequent samples. Note that the + chunk_size, if specified at outlet construction, takes + precedence over the pushthrough flag. (default True) + + Note: performance is optimized for the following argument types: + - `samples`: 2-D numpy array + - `timestamp`: float + """ + # Convert timestamp to corresponding ctype + try: + timestamp_c = ctypes.c_double(timestamp) + # Select the corresponding push_chunk method + liblsl_push_chunk_func = self.do_push_chunk + except TypeError: + try: + timestamp_c = (ctypes.c_double * len(timestamp))(*timestamp) + liblsl_push_chunk_func = self.do_push_chunk_n + except TypeError: + raise TypeError("timestamp must be a float or an iterable of floats") + + try: + n_values = self.channel_count * len(x) + data_buff = (self.value_type * n_values).from_buffer(x) + handle_error( + liblsl_push_chunk_func( + self.obj, + data_buff, + ctypes.c_long(n_values), + timestamp_c, + ctypes.c_int(pushthrough), + ) + ) + except TypeError: + # don't send empty chunks + if len(x): + if type(x[0]) is list: + x = [v for sample in x for v in sample] + if self.channel_format == cf_string: + x = [v.encode("utf-8") for v in x] + if len(x) % self.channel_count == 0: + # x is a flattened list of multiplexed values + constructor = self.value_type * len(x) + # noinspection PyCallingNonCallable + handle_error( + liblsl_push_chunk_func( + self.obj, + constructor(*x), + ctypes.c_long(len(x)), + timestamp_c, + ctypes.c_int(pushthrough), + ) + ) + else: + raise ValueError( + "Each sample must have the same number of channels (" + + str(self.channel_count) + + ")." + ) + + def have_consumers(self) -> bool: + """Check whether consumers are currently registered. + + While it does not hurt, there is technically no reason to push samples + if there is no consumer. + + """ + return bool(lib.lsl_have_consumers(self.obj)) + + def wait_for_consumers(self, timeout: float) -> bool: + """Wait until some consumer shows up (without wasting resources). + + Returns True if the wait was successful, False if the timeout expired. + + """ + return bool(lib.lsl_wait_for_consumers(self.obj, ctypes.c_double(timeout))) + + def get_info(self) -> StreamInfo: + outlet_info = lib.lsl_get_info(self.obj) + return StreamInfo(handle=outlet_info) diff --git a/apps/pylsl/src/pylsl/resolve.py b/apps/pylsl/src/pylsl/resolve.py new file mode 100644 index 0000000..f96fe7d --- /dev/null +++ b/apps/pylsl/src/pylsl/resolve.py @@ -0,0 +1,188 @@ +import ctypes + +from .lib import lib +from .info import StreamInfo +from .util import deprecated, FOREVER + + +def resolve_streams(wait_time=1.0): + """Resolve all streams on the network. + + This function returns all currently available streams from any outlet on + the network. The network is usually the subnet specified at the local + router, but may also include a group of machines visible to each other via + multicast packets (given that the network supports it), or list of + hostnames. These details may optionally be customized by the experimenter + in a configuration file (see Network Connectivity in the LSL wiki). + + Keyword arguments: + wait_time -- The waiting time for the operation, in seconds, to search for + streams. Warning: If this is too short (<0.5s) only a subset + (or none) of the outlets that are present on the network may + be returned. (default 1.0) + + Returns a list of StreamInfo objects (with empty desc field), any of which + can subsequently be used to open an inlet. The full description can be + retrieved from the inlet. + + """ + # noinspection PyCallingNonCallable + buffer = (ctypes.c_void_p * 1024)() + num_found = lib.lsl_resolve_all( + ctypes.byref(buffer), 1024, ctypes.c_double(wait_time) + ) + return [StreamInfo(handle=buffer[k]) for k in range(num_found)] + + +def resolve_byprop(prop, value, minimum=1, timeout=FOREVER): + """Resolve all streams with a specific value for a given property. + + If the goal is to resolve a specific stream, this method is preferred over + resolving all streams and then selecting the desired one. + + Keyword arguments: + prop -- The StreamInfo property that should have a specific value (e.g., + "name", "type", "source_id", or "desc/manufacturer"). + value -- The string value that the property should have (e.g., "EEG" as + the type property). + minimum -- Return at least this many streams. (default 1) + timeout -- Optionally a timeout of the operation, in seconds. If the + timeout expires, less than the desired number of streams + (possibly none) will be returned. (default FOREVER) + + Returns a list of matching StreamInfo objects (with empty desc field), any + of which can subsequently be used to open an inlet. + + Example: results = resolve_Stream_byprop("type","EEG") + + """ + # noinspection PyCallingNonCallable + buffer = (ctypes.c_void_p * 1024)() + num_found = lib.lsl_resolve_byprop( + ctypes.byref(buffer), + 1024, + ctypes.c_char_p(str.encode(prop)), + ctypes.c_char_p(str.encode(value)), + minimum, + ctypes.c_double(timeout), + ) + return [StreamInfo(handle=buffer[k]) for k in range(num_found)] + + +def resolve_bypred(predicate, minimum=1, timeout=FOREVER): + """Resolve all streams that match a given predicate. + + Advanced query that allows to impose more conditions on the retrieved + streams; the given string is an XPath 1.0 predicate for the + node (omitting the surrounding []'s), see also + http://en.wikipedia.org/w/index.php?title=XPath_1.0&oldid=474981951. + + Keyword arguments: + predicate -- The predicate string, e.g. "name='BioSemi'" or + "type='EEG' and starts-with(name,'BioSemi') and + count(description/desc/channels/channel)=32" + minimum -- Return at least this many streams. (default 1) + timeout -- Optionally a timeout of the operation, in seconds. If the + timeout expires, less than the desired number of streams + (possibly none) will be returned. (default FOREVER) + + Returns a list of matching StreamInfo objects (with empty desc field), any + of which can subsequently be used to open an inlet. + + """ + # noinspection PyCallingNonCallable + buffer = (ctypes.c_void_p * 1024)() + num_found = lib.lsl_resolve_bypred( + ctypes.byref(buffer), + 1024, + ctypes.c_char_p(str.encode(predicate)), + minimum, + ctypes.c_double(timeout), + ) + return [StreamInfo(handle=buffer[k]) for k in range(num_found)] + + +class ContinuousResolver: + """A convenience class resolving streams continuously in the background. + + This object can be queried at any time for the set of streams that are + currently visible on the network. + + """ + + def __init__(self, prop=None, value=None, pred=None, forget_after=5.0): + """Construct a new continuous_resolver. + + Keyword arguments: + forget_after -- When a stream is no longer visible on the network + (e.g., because it was shut down), this is the time in + seconds after which it is no longer reported by the + resolver. + + """ + if pred is not None: + if prop is not None or value is not None: + raise ValueError( + "you can only either pass the prop/value " + "argument or the pred argument, but not " + "both." + ) + self.obj = lib.lsl_create_continuous_resolver_bypred( + str.encode(pred), ctypes.c_double(forget_after) + ) + elif prop is not None and value is not None: + self.obj = lib.lsl_create_continuous_resolver_byprop( + str.encode(prop), str.encode(value), ctypes.c_double(forget_after) + ) + elif prop is not None or value is not None: + raise ValueError( + "if prop is specified, then value must be " + "specified, too, and vice versa." + ) + else: + self.obj = lib.lsl_create_continuous_resolver(ctypes.c_double(forget_after)) + self.obj = ctypes.c_void_p(self.obj) + if not self.obj: + raise RuntimeError("could not create continuous resolver.") + + def __del__(self): + """Destructor for the continuous resolver.""" + # noinspection PyBroadException + try: + lib.lsl_destroy_continuous_resolver(self.obj) + except Exception: + pass + + def results(self): + """Obtain the set of currently present streams on the network. + + Returns a list of matching StreamInfo objects (with empty desc + field), any of which can subsequently be used to open an inlet. + + """ + # noinspection PyCallingNonCallable + buffer = (ctypes.c_void_p * 1024)() + num_found = lib.lsl_resolver_results(self.obj, ctypes.byref(buffer), 1024) + return [StreamInfo(handle=buffer[k]) for k in range(num_found)] + + +@deprecated("Use `resolve_streams` instead.") +def resolve_stream(*args): + """Resolve a stream. + + This function is deprecated. Use `resolve_streams` instead. + """ + if len(args) == 0: + return resolve_streams() + elif type(args[0]) in [int, float]: + return resolve_streams(args[0]) + elif type(args[0]) is str: + if len(args) == 1: + return resolve_bypred(args[0]) + elif type(args[1]) in [int, float]: + return resolve_bypred(args[0], args[1]) + else: + if len(args) == 2: + return resolve_byprop(args[0], args[1]) + else: + return resolve_byprop(args[0], args[1], args[2]) diff --git a/apps/pylsl/src/pylsl/util.py b/apps/pylsl/src/pylsl/util.py new file mode 100644 index 0000000..69c4f46 --- /dev/null +++ b/apps/pylsl/src/pylsl/util.py @@ -0,0 +1,213 @@ +import ctypes +import functools +import warnings + +from .lib import lib, _security_api_available + +# Constant to indicate that a stream has variable sampling rate. +IRREGULAR_RATE = 0.0 + +# Constant to indicate that a sample has the next successive time stamp +# according to the stream's defined sampling rate. Optional optimization to +# transmit less data per sample. +DEDUCED_TIMESTAMP = -1.0 + +# A very large time value (ca. 1 year); can be used in timeouts. +FOREVER = 32000000.0 + +# Value formats supported by LSL. LSL data streams are sequences of samples, +# each of which is a same-size vector of values with one of the below types. + +# Post processing flags +proc_none = 0 # No automatic post-processing; return the ground-truth time stamps for manual post-processing. +proc_clocksync = 1 # Perform automatic clock synchronization; equivalent to manually adding the time_correction(). +proc_dejitter = 2 # Remove jitter from time stamps using a smoothing algorithm to the received time stamps. +proc_monotonize = 4 # Force the time-stamps to be monotonically ascending. Only makes sense if timestamps are dejittered. +proc_threadsafe = 8 # Post-processing is thread-safe (same inlet can be read from by multiple threads). +proc_ALL = ( + proc_none | proc_clocksync | proc_dejitter | proc_monotonize | proc_threadsafe +) + + +def protocol_version(): + """Protocol version. + + The major version is protocol_version() / 100; + The minor version is protocol_version() % 100; + + Clients with different minor versions are protocol-compatible with each + other while clients with different major versions will refuse to work + together. + + """ + return lib.lsl_protocol_version() + + +def library_version(): + """Version of the underlying liblsl library. + + The major version is library_version() / 100; + The minor version is library_version() % 100; + + """ + return lib.lsl_library_version() + + +def library_info(): + """Get a string containing library information. The format of the string shouldn't be used + for anything important except giving a a debugging person a good idea which exact library + version is used.""" + return lib.lsl_library_info().decode("utf-8") + + +def is_secure_build(): + """Check if this is a secure build of liblsl. + + Returns True if the library was built with security features (LSL_SECURITY=ON), + False otherwise. This can be used to verify you're using the correct library. + + Example: + if not pylsl.is_secure_build(): + print("WARNING: Not using secure LSL library!") + """ + if not _security_api_available: + return False + return lib.lsl_is_secure_build() != 0 + + +def base_version(): + """Get the base liblsl version string. + + Returns the upstream liblsl version (e.g., "1.16.1"). + Returns "unknown" if security API is not available. + """ + if not _security_api_available: + return "unknown" + return lib.lsl_base_version().decode("utf-8") + + +def security_version(): + """Get the security layer version string. + + Returns the security layer version (e.g., "1.0.0" or "1.0.0-alpha"). + Returns "0.0.0" if this is not a secure build or security API is not available. + """ + if not _security_api_available: + return "0.0.0" + return lib.lsl_security_version().decode("utf-8") + + +def full_version(): + """Get the full version string. + + Returns the combined version (e.g., "1.16.1-secure.1.0.0-alpha"). + For non-secure builds, returns just the base version. + Returns "unknown" if security API is not available. + """ + if not _security_api_available: + return "unknown" + return lib.lsl_full_version().decode("utf-8") + + +def check_security(warn=True): + """Check if using secure LSL library and optionally warn if not. + + Args: + warn: If True, emit a warning when not using secure library. + + Returns: + True if using secure library, False otherwise. + """ + secure = is_secure_build() + if not secure and warn: + warnings.warn( + "Not using secure LSL library. " + "Set PYLSL_LIB to point to liblsl-secure or install liblsl-secure. " + "See https://github.com/sccn/secureLSL for details.", + UserWarning, + stacklevel=2 + ) + return secure + + +def local_clock(): + """Obtain a local system time stamp in seconds. + + The resolution is better than a millisecond. This reading can be used to + assign time stamps to samples as they are being acquired. + + If the "age" of a sample is known at a particular time (e.g., from USB + transmission delays), it can be used as an offset to lsl_local_clock() to + obtain a better estimate of when a sample was actually captured. See + StreamOutlet.push_sample() for a use case. + + """ + return lib.lsl_local_clock() + + +class TimeoutError(RuntimeError): + # note: although this overrides the name of a built-in exception, + # this API is retained here for compatibility with the Python 2.x + # version of pylsl + pass + + +class LostError(RuntimeError): + pass + + +class InvalidArgumentError(RuntimeError): + pass + + +class InternalError(RuntimeError): + pass + + +def handle_error(errcode): + """Error handler function. Translates an error code into an exception.""" + if type(errcode) is ctypes.c_int: + errcode = errcode.value + if errcode == 0: + pass # no error + elif errcode == -1: + raise TimeoutError("the operation failed due to a timeout.") + elif errcode == -2: + raise LostError("the stream has been lost.") + elif errcode == -3: + raise InvalidArgumentError("an argument was incorrectly specified.") + elif errcode == -4: + raise InternalError("an internal error has occurred.") + elif errcode < 0: + raise RuntimeError("an unknown error has occurred.") + + +def deprecated(reason: str = None): + """Mark functions as deprecated. + + It will result in a warning being emitted when the function is used. + + Example: + @deprecated("use new_function instead") + def old_function(): + pass + """ + def decorator(func): + message = f"Function '{func.__name__}' is deprecated." + if reason: + message += f" {reason}" + + @functools.wraps(func) + def wrapper(*args, **kwargs): + warnings.simplefilter("always", DeprecationWarning) # ensure it shows up + warnings.warn( + message, + category=DeprecationWarning, + stacklevel=2 + ) + warnings.simplefilter("default", DeprecationWarning) + return func(*args, **kwargs) + + return wrapper + + return decorator diff --git a/apps/pylsl/test/test_format.py b/apps/pylsl/test/test_format.py new file mode 100644 index 0000000..4cb96f3 --- /dev/null +++ b/apps/pylsl/test/test_format.py @@ -0,0 +1,32 @@ +import ctypes + +import pytest + +import pylsl + + +@pytest.mark.parametrize( + "channel_format", [pylsl.cf_int32, pylsl.cf_float32, pylsl.cf_double64] +) +def test_format(channel_format: int): + expected_type = { + pylsl.cf_int32: ctypes.c_int, + pylsl.cf_float32: ctypes.c_float, + pylsl.cf_double64: ctypes.c_double, + }[channel_format] + + feature_info = pylsl.StreamInfo( + name="test", + type="EEG", + channel_count=1, + nominal_srate=1, + channel_format=channel_format, + source_id="testid", + ) + outlet = pylsl.StreamOutlet(feature_info) + assert outlet.value_type == expected_type + + streams = pylsl.resolve_byprop("name", "test", timeout=1) + inlet = pylsl.StreamInlet(streams[0]) + + assert inlet.value_type == expected_type diff --git a/apps/pylsl/test/test_info.py b/apps/pylsl/test/test_info.py new file mode 100644 index 0000000..c4271b4 --- /dev/null +++ b/apps/pylsl/test/test_info.py @@ -0,0 +1,44 @@ +import pylsl + + +def test_info_src_id(): + name = "TestName" + strm_type = "TestType" + chans = 32 + srate = 1000.0 + fmt = pylsl.cf_float32 + + info = pylsl.StreamInfo( + name=name, + type=strm_type, + channel_count=chans, + nominal_srate=srate, + channel_format=fmt, + source_id=None, + ) + expected_src_id = str(hash((name, strm_type, chans, srate, fmt))) + assert info.source_id() == expected_src_id + + # Augment info with desc + info.desc().append_child_value("manufacturer", "pytest") + chns = info.desc().append_child("channels") + for chan_ix in range(1, chans + 1): + ch = chns.append_child("channel") + ch.append_child_value("label", f"Ch{chan_ix}") + + outlet = pylsl.StreamOutlet(info) + outlet_info = outlet.get_info() + + """ + # See comment block in StreamOutlet.__init__ to see why this is commented out. + import socket + outlet_expected_source_id = str(hash((name, strm_type, chans, srate, fmt, socket.gethostname()))) + """ + outlet_expected_source_id = expected_src_id + + assert outlet_info.source_id() == outlet_expected_source_id + out_desc = outlet_info.desc() + assert out_desc.child_value("manufacturer") == "pytest" + assert outlet_info.get_channel_labels() == [ + f"Ch{chan_ix}" for chan_ix in range(1, chans + 1) + ] diff --git a/docs/integration/pylsl.md b/docs/integration/pylsl.md new file mode 100644 index 0000000..e2cf411 --- /dev/null +++ b/docs/integration/pylsl.md @@ -0,0 +1,92 @@ +# pylsl Integration + +Using the security-enabled Python binding. + +--- + +## Overview + +`pylsl` is the Python binding for Lab Streaming Layer. The security-enabled +version ships as a component of this repository at `apps/pylsl/` and adds: + +- automatic preference for `liblsl-secure` when it is present +- library-level version queries (base, security, and full version strings) +- per-stream security status, so an application can tell whether a discovered + stream is encrypted + +It remains under its upstream MIT license, including the security additions. + +--- + +## Install + +```bash +pip install -e apps/pylsl +``` + +The binding locates `liblsl-secure` automatically when it is installed. To point +at a specific build, set `PYLSL_LIB`: + +```bash +export PYLSL_LIB=/path/to/secureLSL/liblsl/build/liblsl-secure.so +``` + +--- + +## Checking the build + +```python +import pylsl + +pylsl.is_secure_build() # True when linked against liblsl-secure +pylsl.base_version() # underlying liblsl version +pylsl.security_version() # security layer version +pylsl.full_version() # combined version string +``` + +`check_security()` combines the test with a warning, which is convenient at +application start-up: + +```python +pylsl.check_security() # warns if not linked against a secure build +``` + +--- + +## Checking stream security + +`StreamInfo` exposes two methods: + +```python +from pylsl import resolve_streams + +for info in resolve_streams(): + if info.security_enabled(): + print(f"{info.name()}: encrypted, key {info.security_fingerprint()[:16]}") + else: + print(f"{info.name()}: plaintext") +``` + +Both are methods rather than properties. `security_fingerprint()` returns the +fingerprint of the public key the stream was authorized under, which lets an +application confirm a stream belongs to the expected trust group. + +--- + +## Behavior on a stock liblsl + +The security symbols are resolved defensively at import time. Linked against a +stock `liblsl` that does not export them, the binding imports and behaves exactly +as upstream, and the security properties report that the information is +unavailable rather than raising. + +This means the same application code runs against both libraries, which is what +makes the drop-in library replacement work without application changes. + +--- + +## See also + +- [LabRecorder Integration](labrecorder.md) +- [SigVisualizer Integration](sigvisualizer.md) +- [Custom Applications](custom-apps.md) diff --git a/mkdocs.yml b/mkdocs.yml index 06eaae0..ee02e56 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -131,6 +131,7 @@ nav: - ESP32: - Overview: esp32/overview.md - Integration: + - pylsl (Python): integration/pylsl.md - LabRecorder: integration/labrecorder.md - SigVisualizer: integration/sigvisualizer.md - Custom Applications: integration/custom-apps.md