Skip to content

base_camera

BaseCamera(width=640, height=360, callback_thread_closed=None) ¤

Bases: object

Source code in motiontracker\camera\base_camera.py 
75
76
77
78
79
80
81
def __init__(self, width=640, height=360, callback_thread_closed=None):
    """Start the background camera thread if it isn't running yet."""
    BaseCamera.width = width
    BaseCamera.height = height
    BaseCamera.callback_thread_closed = callback_thread_closed
    dummy_image = Image.open(DUMMY_IMAGE)
    BaseCamera.dummy_image = np.asarray(dummy_image.resize((width, height)))

frames() ¤

"Generator that returns frames from the camera.

Source code in motiontracker\camera\base_camera.py 
119
120
121
def frames():
    """"Generator that returns frames from the camera."""
    raise RuntimeError('Must be implemented by subclasses.')

get_frame(timeout=None) ¤

Return the current camera frame.

Source code in motiontracker\camera\base_camera.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get_frame(self, timeout=None):
    """Return the current camera frame."""
    BaseCamera.last_access = time.time()

    # if the camera is off just return the dummy image
    if BaseCamera.thread is None or BaseCamera.stop_flag:
        log.debug("Sending Dummy Image")
        return self.dummy_image

    # wait for a signal from the camera thread
    log.debug("Inside Base Camera Get Frame, Waiting for Event")
    success = BaseCamera.event.wait(timeout=timeout)
    if not success:
        log.debug("Frame took to long, skipping frame and returning None")
        raise RuntimeError("No frame camera frame received")
    log.debug("Inside Base Camera Get Frame, Waiting to Clear Event")
    BaseCamera.event.clear()
    return BaseCamera.frame

FrameEvent() ¤

Bases: object

An Event-like class that signals all active clients when a new frame is available.

Source code in motiontracker\camera\base_camera.py 
26
27
def __init__(self):
    self.events = {}

clear() ¤

Invoked from each client's thread after a frame was processed.

Source code in motiontracker\camera\base_camera.py 
59
60
61
def clear(self):
    """Invoked from each client's thread after a frame was processed."""
    self.events[get_ident()][0].clear()

set() ¤

Invoked by the camera thread when a new frame is available.

Source code in motiontracker\camera\base_camera.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def set(self):
    """Invoked by the camera thread when 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]

wait(timeout=None) ¤

Invoked from each client's thread to wait for the next frame.

Source code in motiontracker\camera\base_camera.py 
29
30
31
32
33
34
35
36
37
def wait(self, timeout=None):
    """Invoked from each client's thread to wait for the next frame."""
    ident = 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] = [threading.Event(), time.time()]
    return self.events[ident][0].wait(timeout=timeout)