Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def refreshStreams(self):
""" Refresh the stream list. """
# get all streams on the lab network
self.ResolvedStreams = list(pylsl.resolve_streams(1.0))
# sort them by UID to get a reproducible order
self.ResolvedStreams.sort(key=lambda x: x.uid())
# get their names
self.ResolvedNames = [];
for k in range(len(self.ResolvedStreams)):
self.ResolvedNames.append(self.ResolvedStreams[k].name() + ' (' + self.ResolvedStreams[k].hostname() +')')
# find those that are required but missing
missing = [n for n in self.RequiredStreams if (n not in self.ResolvedNames)]
self.MissingStreams = missing
allnames = []
allnames += self.ResolvedNames
allnames += missing
# update the listbox contents
good_brush = QBrush()
good_brush.setColor(QColor(0,128,0))
bad_brush = QBrush()
def init_markers(lsl,logfile,datariver,serialport,uid):
""" Initialize the marker protocols to use. """
if lsl:
try:
global lsl_backend
import pylsl.pylsl as pylsl
info = pylsl.stream_info("SNAP-Markers","Markers",1,0,pylsl.cf_string,"SNAPmarkers-" + uid)
lsl_backend = pylsl.stream_outlet(info)
lsl_backend.pylsl = pylsl
print "The lab streaming layer is ready for sending markers."
except:
print "Error initializing the lab streaming layer backend. You will not be able to send and record event markers via LSL."
if logfile:
try:
# find a new slot for the logfiles
for k in xrange(10000):
fname = 'logs/markerlog-' + str(k) + '.log'
if not os.path.exists(fname):
global marker_log
marker_log = open(fname,'w')
break
print "A marker logfile has been prepared for logging."
except:
def init_markers(lsl,logfile,datariver,serialport,uid):
""" Initialize the marker protocols to use. """
if lsl:
try:
global lsl_backend
import pylsl.pylsl as pylsl
info = pylsl.stream_info("SNAP-Markers","Markers",1,0,pylsl.cf_string,"SNAPmarkers-" + uid)
lsl_backend = pylsl.stream_outlet(info)
lsl_backend.pylsl = pylsl
print "The lab streaming layer is ready for sending markers."
except:
print "Error initializing the lab streaming layer backend. You will not be able to send and record event markers via LSL."
if logfile:
try:
# find a new slot for the logfiles
for k in xrange(10000):
fname = 'logs/markerlog-' + str(k) + '.log'
if not os.path.exists(fname):
global marker_log
marker_log = open(fname,'w')
break
print "A marker logfile has been prepared for logging."
def init_markers(lsl,logfile,datariver,serialport,uid):
""" Initialize the marker protocols to use. """
if lsl:
try:
global lsl_backend
import pylsl.pylsl as pylsl
info = pylsl.stream_info("SNAP-Markers","Markers",1,0,pylsl.cf_string,"SNAPmarkers-" + uid)
lsl_backend = pylsl.stream_outlet(info)
lsl_backend.pylsl = pylsl
print "The lab streaming layer is ready for sending markers."
except:
print "Error initializing the lab streaming layer backend. You will not be able to send and record event markers via LSL."
if logfile:
try:
# find a new slot for the logfiles
for k in xrange(10000):
fname = 'logs/markerlog-' + str(k) + '.log'
if not os.path.exists(fname):
global marker_log
marker_log = open(fname,'w')
break
print "A marker logfile has been prepared for logging."
except:
print "Error initializing the marker logging. Your event markers will not be logged into a file."
raise ValueError("Please specify a stream name or a property and value.")
if offset_correction:
self.logger.warning(
'The "offset_correction" parameter is deprecated. Use "sync" instead.'
)
sync = "local"
self._prop = prop
self._value = value
self._inlet = None
self._labels = None
self._unit = unit
self._sync = sync
self._channels = channels
self._timeout = timeout
self._max_samples = max_samples
self._offset = time() - pylsl.local_clock()