from abc import ABC, abstractmethod
import asyncio
import io
import logging
import multiprocessing
from threading import Timer
import time
from PIL import Image, ImageDraw
import numpy as np
from fury.decorators import warn_on_args_to_kwargs
from fury.stream.constants import PY_VERSION_8
if PY_VERSION_8:
from multiprocessing import resource_tracker, shared_memory
else:
shared_memory = None # type: ignore
_FLOAT_ShM_TYPE = "d"
_INT_ShM_TYPE = "i"
_UINT_ShM_TYPE = "I"
_BYTE_ShM_TYPE = "B"
_FLOAT_SIZE = np.dtype(_FLOAT_ShM_TYPE).itemsize
_INT_SIZE = np.dtype(_INT_ShM_TYPE).itemsize
_UINT_SIZE = np.dtype(_UINT_ShM_TYPE).itemsize
_BYTE_SIZE = np.dtype(_BYTE_ShM_TYPE).itemsize
[docs]
def remove_shm_from_resource_tracker():
"""Monkey-patch multiprocessing.resource_tracker so SharedMemory won't
be tracked
Notes
-----
More details at: https://bugs.python.org/issue38119
"""
def fix_register(name, rtype):
if rtype == "shared_memory":
return
try:
return resource_tracker._resource_tracker.register(self, name, rtype)
except NameError:
return None
resource_tracker.register = fix_register
def fix_unregister(name, rtype):
if rtype == "shared_memory":
return
try:
return resource_tracker._resource_tracker.unregister(self, name, rtype)
except NameError:
return None
resource_tracker.unregister = fix_unregister
if "shared_memory" in resource_tracker._CLEANUP_FUNCS:
del resource_tracker._CLEANUP_FUNCS["shared_memory"]
[docs]
class GenericMultiDimensionalBuffer(ABC):
"""This implements a abstract (generic) multidimensional buffer."""
@warn_on_args_to_kwargs()
def __init__(self, *, max_size=None, dimension=8):
"""Initialize the multidimensional buffer.
Parameters
----------
max_size : int, optional
If buffer_name or buffer was not passed then max_size
it's mandatory
dimension : int, default 8
"""
self.max_size = max_size
self.dimension = dimension
self.buffer_name = None
self._buffer = None
self._buffer_repr = None
self._created = False
@property
def buffer(self):
return self._buffer
@buffer.setter
def buffer(self, data):
if isinstance(data, (np.ndarray, np.generic)):
if data.dtype == _FLOAT_ShM_TYPE:
self._buffer_repr[:] = data
[docs]
def get_start_end(self, idx):
dim = self.dimension
start = idx * dim
end = dim * (idx + 1)
return start, end
def __getitem__(self, idx):
start, end = self.get_start_end(idx)
logging.info(f"dequeue start {int(time.time()*1000)}")
ts = time.time() * 1000
items = self._buffer_repr[start:end]
te = time.time() * 1000
logging.info(f"dequeue frombuffer cost {te-ts:.2f}")
return items
def __setitem__(self, idx, data):
start, end = self.get_start_end(idx)
if isinstance(data, (np.ndarray, np.generic)):
if data.dtype == _FLOAT_ShM_TYPE:
# if end - start == self.dimension and start >= 0 and end >= 0:
self._buffer_repr[start:end] = data
[docs]
@abstractmethod
def load_mem_resource(self): ... # pragma: no cover
[docs]
@abstractmethod
def create_mem_resource(self): ... # pragma: no cover
[docs]
@abstractmethod
def cleanup(self): ... # pragma: no cover
[docs]
class RawArrayMultiDimensionalBuffer(GenericMultiDimensionalBuffer):
"""This implements a multidimensional buffer with RawArray."""
@warn_on_args_to_kwargs()
def __init__(self, max_size, *, dimension=4, buffer=None):
"""Stream system uses that to implement the CircularQueue
with shared memory resources.
Parameters
----------
max_size : int, optional
If buffer_name or buffer was not passed then max_size
it's mandatory
dimension : int, default 8
buffer : buffer, optional
If buffer is not passed to __init__
then the multidimensional buffer obj will create a new
RawArray object to store the data
If buffer is passed than this Obj will read a
a already created RawArray
"""
super().__init__(max_size=max_size, dimension=dimension)
if buffer is None:
self.create_mem_resource()
else:
self._buffer = buffer
self.load_mem_resource()
[docs]
def create_mem_resource(self):
buffer_arr = np.zeros(
self.dimension * (self.max_size + 1), dtype=_FLOAT_ShM_TYPE
)
buffer = multiprocessing.RawArray(
_FLOAT_ShM_TYPE, np.ctypeslib.as_ctypes(buffer_arr)
)
self._buffer = buffer
self._buffer_repr = np.ctypeslib.as_array(self._buffer)
[docs]
def load_mem_resource(self):
self.max_size = int(len(self._buffer) // self.dimension)
self.max_size -= 1
self._buffer_repr = np.ctypeslib.as_array(self._buffer)
[docs]
def cleanup(self):
pass
[docs]
class SharedMemMultiDimensionalBuffer(GenericMultiDimensionalBuffer):
"""This implements a generic multidimensional buffer
with SharedMemory.
"""
@warn_on_args_to_kwargs()
def __init__(self, max_size, *, dimension=4, buffer_name=None):
"""Stream system uses that to implement the
CircularQueue with shared memory resources.
Parameters
----------
max_size : int, optional
If buffer_name or buffer was not passed then max_size
it's mandatory
dimension : int, default 8
buffer_name : str, optional
if buffer_name is passed than this Obj will read a
a already created SharedMemory
"""
super().__init__(max_size=max_size, dimension=dimension)
if buffer_name is None:
self.create_mem_resource()
self._created = True
else:
self.buffer_name = buffer_name
self.load_mem_resource()
self._created = False
self._create_repr()
[docs]
def create_mem_resource(self):
self._num_el = self.dimension * (self.max_size + 1)
buffer_arr = np.zeros(self._num_el + 2, dtype=_FLOAT_ShM_TYPE)
self._buffer = shared_memory.SharedMemory(create=True, size=buffer_arr.nbytes)
sizes = np.ndarray(
2, dtype=_FLOAT_ShM_TYPE, buffer=self._buffer.buf[0 : _FLOAT_SIZE * 2]
)
sizes[0] = self.max_size
sizes[1] = self.dimension
self.buffer_name = self._buffer.name
logging.info(
[
"create repr multidimensional buffer ",
]
)
[docs]
def load_mem_resource(self):
self._buffer = shared_memory.SharedMemory(self.buffer_name)
sizes = np.ndarray(2, dtype="d", buffer=self._buffer.buf[0 : _FLOAT_SIZE * 2])
self.max_size = int(sizes[0])
self.dimension = int(sizes[1])
num_el = int((sizes[0] + 1) * sizes[1])
self._num_el = num_el
logging.info(
[
"load repr multidimensional buffer",
]
)
def _create_repr(self):
start = _FLOAT_SIZE * 2
end = (self._num_el + 2) * _FLOAT_SIZE
self._buffer_repr = np.ndarray(
self._num_el, dtype=_FLOAT_ShM_TYPE, buffer=self._buffer.buf[start:end]
)
logging.info(
[
"create repr multidimensional buffer",
self._buffer_repr.shape,
"max size",
self.max_size,
"dimension",
self.dimension,
]
)
[docs]
def cleanup(self):
self._buffer.close()
if self._created:
# this it's due the python core issues
# https://bugs.python.org/issue38119
# https://bugs.python.org/issue39959
# https://github.com/luizalabs/shared-memory-dict/issues/13
try:
self._buffer.unlink()
except FileNotFoundError:
print(
f"Shared Memory {self.buffer_name}(queue_event_buffer)\
File not found"
)
[docs]
class GenericCircularQueue(ABC):
"""This implements a generic circular queue which works with
shared memory resources.
"""
@warn_on_args_to_kwargs()
def __init__(
self,
*,
max_size=None,
dimension=8,
use_shared_mem=False,
buffer=None,
buffer_name=None,
):
"""Initialize the circular queue.
Parameters
----------
max_size : int, optional
If buffer_name or buffer was not passed then max_size
it's mandatory. This will be used to construct the
multidimensional buffer
dimension : int, default 8
This will be used to construct the multidimensional buffer
use_shared_mem : bool, default False
If the multidimensional memory resource should create or read
using SharedMemory or RawArrays
buffer : RawArray, optional
buffer_name: str, optional
"""
self._created = False
self.head_tail_buffer_name = None
self.head_tail_buffer_repr = None
self.head_tail_buffer = None
self._use_shared_mem = use_shared_mem
if use_shared_mem:
self.buffer = SharedMemMultiDimensionalBuffer(
max_size=max_size, dimension=dimension, buffer_name=buffer_name
)
else:
self.buffer = RawArrayMultiDimensionalBuffer(
max_size=max_size, dimension=dimension, buffer=buffer
)
@property
def head(self):
if self._use_shared_mem:
return self.head_tail_buffer_repr[0]
else:
return np.frombuffer(self.head_tail_buffer.get_obj(), _INT_ShM_TYPE)[0]
@head.setter
def head(self, value):
self.head_tail_buffer_repr[0] = value
@property
def tail(self):
if self._use_shared_mem:
return self.head_tail_buffer_repr[1]
else:
return np.frombuffer(self.head_tail_buffer.get_obj(), _INT_ShM_TYPE)[1]
@tail.setter
def tail(self, value):
self.head_tail_buffer_repr[1] = value
[docs]
def set_head_tail(self, head, tail, lock=1):
self.head_tail_buffer_repr[0:3] = np.array([head, tail, lock]).astype(
_INT_ShM_TYPE
)
def _enqueue(self, data):
ok = False
if (self.tail + 1) % self.buffer.max_size == self.head:
ok = False
else:
if self.head == -1:
self.set_head_tail(0, 0, 1)
else:
self.tail = (self.tail + 1) % self.buffer.max_size
self.buffer[self.tail] = data
ok = True
return ok
def _dequeue(self):
if self.head == -1:
interactions = None
else:
if self.head != self.tail:
interactions = self.buffer[self.head]
self.head = (self.head + 1) % self.buffer.max_size
else:
interactions = self.buffer[self.head]
self.set_head_tail(-1, -1, 1)
return interactions
[docs]
@abstractmethod
def enqueue(self, data):
pass # pragma: no cover
[docs]
@abstractmethod
def dequeue(self):
pass # pragma: no cover
[docs]
@abstractmethod
def load_mem_resource(self):
pass # pragma: no cover
[docs]
@abstractmethod
def create_mem_resource(self):
pass # pragma: no cover
[docs]
@abstractmethod
def cleanup(self):
pass # pragma: no cover
[docs]
class ArrayCircularQueue(GenericCircularQueue):
"""This implements a MultiDimensional Queue which works with
Arrays and RawArrays.
"""
@warn_on_args_to_kwargs()
def __init__(self, *, max_size=10, dimension=6, head_tail_buffer=None, buffer=None):
"""Stream system uses that to implement user interactions
Parameters
----------
max_size : int, optional
If buffer_name or buffer was not passed then max_size
it's mandatory. This will be used to construct the
multidimensional buffer
dimension : int, default 8
This will be used to construct the multidimensional buffer
head_tail_buffer : buffer, optional
If buffer is not passed to __init__
then this obj will create a new
RawArray to store head and tail position.
buffer : buffer, optional
If buffer is not passed to __init__
then the multidimensional buffer obj will create a new
RawArray to store the data
"""
super().__init__(
max_size=max_size,
dimension=dimension,
use_shared_mem=False,
buffer=buffer,
)
if head_tail_buffer is None:
self.create_mem_resource()
self._created = True
else:
self.head_tail_buffer = head_tail_buffer
self._created = False
self.head_tail_buffer_name = None
self.head_tail_buffer_repr = self.head_tail_buffer
if self._created:
self.set_head_tail(-1, -1, 0)
[docs]
def load_mem_resource(self):
pass # pragma: no cover
[docs]
def create_mem_resource(self):
# head_tail_arr[0] int; head position
# head_tail_arr[1] int; tail position
head_tail_arr = np.array([-1, -1, 0], dtype=_INT_ShM_TYPE)
self.head_tail_buffer = multiprocessing.Array(
_INT_ShM_TYPE,
head_tail_arr,
)
[docs]
def enqueue(self, data):
ok = False
with self.head_tail_buffer.get_lock():
ok = self._enqueue(data)
return ok
[docs]
def dequeue(self):
with self.head_tail_buffer.get_lock():
interactions = self._dequeue()
return interactions
[docs]
def cleanup(self):
pass
[docs]
class SharedMemCircularQueue(GenericCircularQueue):
"""This implements a MultiDimensional Queue which works with
SharedMemory.
"""
@warn_on_args_to_kwargs()
def __init__(
self, *, max_size=10, dimension=6, head_tail_buffer_name=None, buffer_name=None
):
"""Stream system uses that to implement user interactions
Parameters
----------
max_size : int, optional
If buffer_name or buffer was not passed then max_size
it's mandatory. This will be used to construct the
multidimensional buffer
dimension : int, default 8
This will be used to construct the multidimensional buffer
head_tail_buffer_name : str, optional
if buffer_name is passed than this Obj will read a
a already created SharedMemory with the head and tail
information
buffer_name : str, optional
if buffer_name is passed than this Obj will read a
a already created SharedMemory to create the MultiDimensionalBuffer
"""
super().__init__(
max_size=max_size,
dimension=dimension,
use_shared_mem=True,
buffer_name=buffer_name,
)
if head_tail_buffer_name is None:
self.create_mem_resource()
self._created = True
else:
self.head_tail_buffer_name = head_tail_buffer_name
self.load_mem_resource()
self._created = False
self.head_tail_buffer_repr = np.ndarray(
3, dtype=_INT_ShM_TYPE, buffer=self.head_tail_buffer.buf[0 : 3 * _INT_SIZE]
)
logging.info(
[
"create shared mem",
"size repr",
self.head_tail_buffer_repr.shape,
"size buffer",
self.head_tail_buffer.size / _INT_SIZE,
]
)
if self._created:
self.set_head_tail(-1, -1, 0)
[docs]
def load_mem_resource(self):
self.head_tail_buffer = shared_memory.SharedMemory(self.head_tail_buffer_name)
[docs]
def create_mem_resource(self):
# head_tail_arr[0] int; head position
# head_tail_arr[1] int; tail position
head_tail_arr = np.array([-1, -1, 0], dtype=_INT_ShM_TYPE)
self.head_tail_buffer = shared_memory.SharedMemory(
create=True, size=head_tail_arr.nbytes
)
self.head_tail_buffer_name = self.head_tail_buffer.name
[docs]
def is_unlocked(self):
return self.head_tail_buffer_repr[2] == 0
[docs]
def lock(self):
self.head_tail_buffer_repr[2] = 1
[docs]
def unlock(self):
self.head_tail_buffer_repr[2] = 0
[docs]
def enqueue(self, data):
ok = False
if self.is_unlocked():
self.lock()
ok = self._enqueue(data)
self.unlock()
return ok
[docs]
def dequeue(self):
interactions = None
if self.is_unlocked():
self.lock()
interactions = self._dequeue()
self.unlock()
return interactions
[docs]
def cleanup(self):
self.buffer.cleanup()
self.head_tail_buffer.close()
if self._created:
# this it's due the python core issues
# https://bugs.python.org/issue38119
# https://bugs.python.org/issue39959
# https://github.com/luizalabs/shared-memory-dict/issues/13
try:
self.head_tail_buffer.unlink()
except FileNotFoundError:
print(
f"Shared Memory {self.head_tail_buffer_name}(head_tail)\
File not found"
)
[docs]
class GenericImageBufferManager(ABC):
"""This implements a abstract (generic) ImageBufferManager with
the n-buffer technique.
"""
@warn_on_args_to_kwargs()
def __init__(self, *, max_window_size=None, num_buffers=2, use_shared_mem=False):
"""Initialize the ImageBufferManager.
Parameters
----------
max_window_size : tuple of ints, optional
This allows resize events inside of the FURY window instance.
Should be greater than the window size.
num_buffers : int, optional
Number of buffers to be used in the n-buffering
technique.
use_shared_mem: bool, default False
"""
self.max_window_size = np.array(max_window_size)
self.num_buffers = num_buffers
self.info_buffer_size = num_buffers * 2 + 2
self._use_shared_mem = use_shared_mem
self.max_size = None # int
self.num_components = 3
self.image_reprs = []
self.image_buffers = []
self.image_buffer_names = []
self.info_buffer_name = None
self.info_buffer = None
self.info_buffer_repr = None
self._created = False
size = (self.max_window_size[0], self.max_window_size[1])
img = Image.new("RGB", size, color=(0, 0, 0))
d = ImageDraw.Draw(img)
pos_text = (12, size[1] // 2)
d.text(
pos_text, "Image size have exceed the Buffer Max Size", fill=(255, 255, 0)
)
img = np.flipud(img)
self.img_exceed = np.asarray(img).flatten()
@property
def next_buffer_index(self):
index = int((self.info_buffer_repr[1] + 1) % self.num_buffers)
return index
@property
def buffer_index(self):
index = self.info_buffer_repr[1]
return index
[docs]
def write_into(self, w, h, np_arr):
buffer_size = buffer_size = int(h * w * 3)
next_buffer_index = self.next_buffer_index
if buffer_size == self.max_size:
self.image_reprs[next_buffer_index][:] = np_arr
elif buffer_size < self.max_size:
self.image_reprs[next_buffer_index][0:buffer_size] = np_arr
else:
self.image_reprs[next_buffer_index][0 : self.max_size] = self.img_exceed
w = self.max_window_size[0]
h = self.max_window_size[1]
self.info_buffer_repr[2 + next_buffer_index * 2] = w
self.info_buffer_repr[2 + next_buffer_index * 2 + 1] = h
self.info_buffer_repr[1] = next_buffer_index
[docs]
def get_current_frame(self):
"""Get the current frame from the buffer."""
if not self._use_shared_mem:
image_info = np.frombuffer(self.info_buffer, _UINT_ShM_TYPE)
else:
image_info = self.info_buffer_repr
buffer_index = int(image_info[1])
self.width = int(image_info[2 + buffer_index * 2])
self.height = int(image_info[2 + buffer_index * 2 + 1])
image = self.image_reprs[buffer_index]
self.image_buffer_repr = image
return self.width, self.height, image
[docs]
def get_jpeg(self):
"""Returns a jpeg image from the buffer.
Returns
-------
bytes: jpeg image.
"""
width, height, image = self.get_current_frame()
if self._use_shared_mem:
image = np.frombuffer(image, _BYTE_ShM_TYPE)
image = image[0 : width * height * 3].reshape((height, width, 3))
image = np.flipud(image)
image_encoded = Image.fromarray(image, mode="RGB")
bytes_img_data = io.BytesIO()
image_encoded.save(bytes_img_data, format="jpeg")
bytes_img = bytes_img_data.getvalue()
return bytes_img
[docs]
@warn_on_args_to_kwargs()
async def async_get_jpeg(self, *, ms=33):
jpeg = self.get_jpeg()
await asyncio.sleep(ms / 1000)
return jpeg
[docs]
@abstractmethod
def load_mem_resource(self):
pass # pragma: no cover
[docs]
@abstractmethod
def create_mem_resource(self):
pass # pragma: no cover
[docs]
@abstractmethod
def cleanup(self):
pass # pragma: no cover
[docs]
class RawArrayImageBufferManager(GenericImageBufferManager):
"""This implements an ImageBufferManager using RawArrays."""
@warn_on_args_to_kwargs()
def __init__(
self,
*,
max_window_size=(100, 100),
num_buffers=2,
image_buffers=None,
info_buffer=None,
):
"""Initialize the ImageBufferManager.
Parameters
----------
max_window_size : tuple of ints, optional
This allows resize events inside of the FURY window instance.
Should be greater than the window size.
num_buffers : int, optional
Number of buffers to be used in the n-buffering
technique.
info_buffer : buffer, optional
A buffer with the information about the current
frame to be streamed and the respective sizes
image_buffers : list of buffers, optional
A list of buffers with each one containing a frame.
"""
super().__init__(
max_window_size=max_window_size,
num_buffers=num_buffers,
use_shared_mem=False,
)
if image_buffers is None or info_buffer is None:
self.create_mem_resource()
else:
self.image_buffers = image_buffers
self.info_buffer = info_buffer
self.load_mem_resource()
[docs]
def create_mem_resource(self):
self.max_size = self.max_window_size[0] * self.max_window_size[1]
self.max_size *= self.num_components
for _ in range(self.num_buffers):
buffer = multiprocessing.RawArray(
_BYTE_ShM_TYPE,
np.ctypeslib.as_ctypes(
np.random.randint(0, 255, size=self.max_size, dtype=_BYTE_ShM_TYPE)
),
)
self.image_buffers.append(buffer)
self.image_reprs.append(np.ctypeslib.as_array(buffer))
# info_list stores the information about the n frame buffers
# as well the respectives sizes.
# 0 number of components
# 1 id buffer
# 2, 3, width first buffer, height first buffer
# 4, 5, width second buffer , height second buffer
info_list = [3, 0]
for _ in range(self.num_buffers):
info_list += [self.max_window_size[0]]
info_list += [self.max_window_size[1]]
info_list = np.array(info_list, dtype=_UINT_ShM_TYPE)
self.info_buffer = multiprocessing.RawArray(
_UINT_ShM_TYPE, np.ctypeslib.as_ctypes(np.array(info_list))
)
self.info_buffer_repr = np.ctypeslib.as_array(self.info_buffer)
[docs]
def load_mem_resource(self):
self.info_buffer = np.frombuffer(self.info_buffer, _UINT_ShM_TYPE)
self.info_buffer_repr = np.ctypeslib.as_array(self.info_buffer)
for img_buffer in self.image_buffers:
self.image_reprs.append(np.ctypeslib.as_array(img_buffer))
[docs]
def cleanup(self):
pass
[docs]
class SharedMemImageBufferManager(GenericImageBufferManager):
"""This implements an ImageBufferManager using the
SharedMemory approach.
"""
@warn_on_args_to_kwargs()
def __init__(
self,
*,
max_window_size=(100, 100),
num_buffers=2,
image_buffer_names=None,
info_buffer_name=None,
):
"""Initialize the ImageBufferManager.
Parameters
----------
max_window_size : tuple of ints, optional
This allows resize events inside of the FURY window instance.
Should be greater than the window size.
num_buffers : int, optional
Number of buffers to be used in the n-buffering
technique.
info_buffer_name : str
The name of a buffer with the information about the current
frame to be streamed and the respective sizes
image_buffer_names : list of str, optional
a list of buffer names. Each buffer contains a frame
Notes
-----
Python >=3.8 is a requirement to use this object.
"""
super().__init__(
max_window_size=max_window_size,
num_buffers=num_buffers,
use_shared_mem=True,
)
if image_buffer_names is None or info_buffer_name is None:
self.create_mem_resource()
self._created = True
else:
self.image_buffer_names = image_buffer_names
self.info_buffer_name = info_buffer_name
self._created = False
self.load_mem_resource()
[docs]
def create_mem_resource(self):
self.max_size = self.max_window_size[0] * self.max_window_size[1]
self.max_size *= self.num_components
self.max_size = int(self.max_size)
for _ in range(self.num_buffers):
buffer = shared_memory.SharedMemory(create=True, size=self.max_size)
self.image_buffers.append(buffer)
self.image_reprs.append(
np.ndarray(self.max_size, dtype=_BYTE_ShM_TYPE, buffer=buffer.buf)
)
self.image_buffer_names.append(buffer.name)
info_list = [2 + self.num_buffers * 2, 1, 3, 0]
for _ in range(self.num_buffers):
info_list += [self.max_window_size[0]]
info_list += [self.max_window_size[1]]
info_list = np.array(info_list, dtype=_UINT_ShM_TYPE)
self.info_buffer = shared_memory.SharedMemory(
create=True, size=info_list.nbytes
)
sizes = np.ndarray(
2, dtype=_UINT_ShM_TYPE, buffer=self.info_buffer.buf[0 : _UINT_SIZE * 2]
)
sizes[0] = info_list[0]
sizes[1] = 1
self.info_buffer_repr = np.ndarray(
sizes[0],
dtype=_UINT_ShM_TYPE,
buffer=self.info_buffer.buf[2 * _UINT_SIZE :],
)
logging.info(
[
"info buffer create",
"buffer size",
sizes[0],
"repr size",
self.info_buffer_repr.shape,
]
)
self.info_buffer_name = self.info_buffer.name
[docs]
def load_mem_resource(self):
self.info_buffer = shared_memory.SharedMemory(self.info_buffer_name)
sizes = np.ndarray(
2, dtype=_UINT_ShM_TYPE, buffer=self.info_buffer.buf[0 : _UINT_SIZE * 2]
)
self.info_buffer_repr = np.ndarray(
sizes[0],
dtype=_UINT_ShM_TYPE,
buffer=self.info_buffer.buf[2 * _UINT_SIZE :],
)
logging.info(
[
"info buffer load",
"buffer size",
sizes[0],
"repr size",
self.info_buffer_repr.shape,
]
)
for buffer_name in self.image_buffer_names:
buffer = shared_memory.SharedMemory(buffer_name)
self.image_buffers.append(buffer)
self.image_reprs.append(
np.ndarray(
buffer.size // _BYTE_SIZE, dtype=_BYTE_ShM_TYPE, buffer=buffer.buf
)
)
[docs]
def cleanup(self):
"""Release the resources used by the Shared Memory Manager"""
self.info_buffer.close()
# this it's due the python core issues
# https://bugs.python.org/issue38119
# https://bugs.python.org/issue39959
# https://github.com/luizalabs/shared-memory-dict/issues/13
if self._created:
try:
self.info_buffer.unlink()
except FileNotFoundError:
print(
f"Shared Memory {self.info_buffer_name}\
(info_buffer) File not found"
)
for buffer, name in zip(self.image_buffers, self.image_buffer_names):
buffer.close()
if self._created:
try:
buffer.unlink()
except FileNotFoundError:
print(f"Shared Memory {name}(buffer image) File not found")
[docs]
class IntervalTimerThreading:
"""Implements a object with the same behavior of setInterval from Js"""
def __init__(self, seconds, callback, *args, **kwargs):
"""
Parameters
----------
seconds : float
A positive float number. Represents the total amount of
seconds between each call
callback : function
The function to be called
*args : args
args to be passed to callback
**kwargs : kwargs
kwargs to be passed to callback
Examples
--------
.. code-block:: python
def callback(arr):
arr += [len(arr)]
arr = []
interval_timer = tools.IntervalTimer(1, callback, arr)
interval_timer.start()
time.sleep(5)
interval_timer.stop()
# len(arr) == 5
References
-----------
[1] https://stackoverflow.com/questions/3393612/run-certain-code-every-n-seconds
""" # noqa
self._timer = None
self.seconds = seconds
self.callback = callback
self.args = args
self.kwargs = kwargs
self.is_running = False
self.start()
def _run(self):
self.is_running = False
self.start()
self.callback(*self.args, **self.kwargs)
[docs]
def start(self):
"""Start the timer"""
if self.is_running:
return
self._timer = Timer(self.seconds, self._run)
self._timer.daemon = True
self._timer.start()
self.is_running = True
[docs]
def stop(self):
"""Stop the timer"""
if self._timer is None:
return
self._timer.cancel()
if self._timer.is_alive():
self._timer.join()
self.is_running = False
self._timer = None
[docs]
class IntervalTimer:
"""A object that creates a timer that calls a function periodically."""
def __init__(self, seconds, callback, *args, **kwargs):
"""Parameters
----------
seconds : float
A positive float number. Represents the total amount of
seconds between each call
callback : function
The function to be called
*args : args
args to be passed to callback
**kwargs : kwargs
kwargs to be passed to callback
"""
self._seconds = seconds
self._callback = callback
self.args = args
self.kwargs = kwargs
self._is_running = False
self.start()
async def _run(self):
self._is_running = True
while True:
await asyncio.sleep(self._seconds)
if self._is_running:
self._callback(*self.args, **self.kwargs)
[docs]
def start(self):
"""Start the timer"""
if self._is_running:
return
self._loop = asyncio.get_event_loop()
self._task = self._loop.create_task(self._run())
[docs]
def stop(self):
"""Stop the timer"""
self._task.cancel()
self._is_running = False
