Use LabThings ClientEvent
This commit is contained in:
parent
9e4934c054
commit
43c8fea37f
1 changed files with 6 additions and 49 deletions
|
|
@ -3,16 +3,18 @@ import time
|
|||
import os
|
||||
import shutil
|
||||
import threading
|
||||
import gevent.event
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
import threading
|
||||
|
||||
from abc import ABCMeta, abstractmethod
|
||||
from collections import OrderedDict
|
||||
|
||||
from .capture import CaptureObject, build_captures_from_exif
|
||||
from openflexure_microscope.utilities import entry_by_uuid
|
||||
from labthings.core.lock import StrictLock
|
||||
from labthings.core.event import ClientEvent
|
||||
|
||||
from openflexure_microscope.paths import data_file_path
|
||||
|
||||
|
|
@ -45,51 +47,6 @@ def generate_numbered_basename(obj_list: list) -> str:
|
|||
return basename
|
||||
|
||||
|
||||
class CameraEvent(object):
|
||||
"""
|
||||
A frame-signaller object used by any instances or subclasses of BaseCamera.
|
||||
|
||||
An event-like class that signals all active clients when a new frame is available.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.events = {}
|
||||
|
||||
def wait(self, timeout: int = 5):
|
||||
"""Wait for the next frame (invoked from each client's thread)."""
|
||||
ident = threading.get_ident()
|
||||
if ident not in self.events:
|
||||
# this is a new client
|
||||
# add an entry for it in the self.events dict
|
||||
# each entry has two elements, a threading.Event() and a timestamp
|
||||
self.events[ident] = [gevent.event.Event(), time.time()]
|
||||
return self.events[ident][0].wait(timeout)
|
||||
|
||||
def set(self):
|
||||
"""Signal that a new frame is available."""
|
||||
now = time.time()
|
||||
remove = None
|
||||
for ident, event in self.events.items():
|
||||
if not event[0].isSet():
|
||||
# if this client's event is not set, then set it
|
||||
# also update the last set timestamp to now
|
||||
event[0].set()
|
||||
event[1] = now
|
||||
else:
|
||||
# if the client's event is already set, it means the client
|
||||
# did not process a previous frame
|
||||
# if the event stays set for more than 5 seconds, then assume
|
||||
# the client is gone and remove it
|
||||
if now - event[1] > 5:
|
||||
remove = ident
|
||||
if remove:
|
||||
del self.events[remove]
|
||||
|
||||
def clear(self):
|
||||
"""Clear frame event, once processed."""
|
||||
self.events[threading.get_ident()][0].clear()
|
||||
|
||||
|
||||
class BaseCamera(metaclass=ABCMeta):
|
||||
"""
|
||||
Base implementation of StreamingCamera.
|
||||
|
|
@ -99,11 +56,11 @@ class BaseCamera(metaclass=ABCMeta):
|
|||
self.thread = None
|
||||
self.camera = None
|
||||
|
||||
self.lock = StrictLock(timeout=1)
|
||||
self.lock = StrictLock(timeout=1, name="Camera")
|
||||
|
||||
self.frame = None
|
||||
self.last_access = 0
|
||||
self.event = CameraEvent()
|
||||
self.event = ClientEvent()
|
||||
self.stop = False # Used to indicate that the stream loop should break
|
||||
|
||||
self.stream_timeout = 20
|
||||
|
|
@ -307,6 +264,7 @@ class BaseCamera(metaclass=ABCMeta):
|
|||
self.stop = False
|
||||
|
||||
if not self.stream_active:
|
||||
# Spawn a greenlet to handle stream
|
||||
# start background frame thread
|
||||
self.thread = threading.Thread(target=self._thread)
|
||||
self.thread.daemon = True
|
||||
|
|
@ -368,7 +326,6 @@ class BaseCamera(metaclass=ABCMeta):
|
|||
for frame in self.frames_iterator:
|
||||
self.frame = frame
|
||||
self.event.set() # send signal to clients
|
||||
time.sleep(0)
|
||||
|
||||
# Handle timeout
|
||||
if (
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue