12753874046

Committed 13 Jan 2025 06:40PM UTC coverage: 95.076% (+14.5%) from 80.557%

Build # 12753874046

Build Type

Pull #61

github

Specific Base 3a8d3f

Committed by

kevdevg

Commit Message

fix: vision pillow read bytes

Pull Request Pull Request #61: feat(Hugging face): Vision methods - image classification / image segmentation / object detection

Run Details

179 of 189 new or added lines in 5 files covered. (94.71%)

34 existing lines in 9 files now uncovered.

2008 of 2112 relevant lines covered (95.08%)

3.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.69

/scope3ai/worker.py

import logging
import queue
import threading
from time import sleep
from typing import Callable, Optional

from time import monotonic

logger = logging.getLogger("scope3ai.worker")


class BackgroundWorker:
    STOP_WORKER = object()

    def __init__(self, size: int) -> None:
        self._size = size
        self._queue = queue.Queue(maxsize=size)
        self._lock = threading.Lock()
        self._thread: Optional[threading.Thread] = None
        self._pause_event = threading.Event()
        self._pause_event.set()

    @property
    def is_alive(self) -> bool:
        return self._thread and self._thread.is_alive()

    def _ensure_thread(self) -> None:
        if not self.is_alive:
            self.start()

    def submit(self, callback: Callable[[], None]) -> bool:
        self._ensure_thread()
        try:
            self._queue.put_nowait(callback)
            return True
        except queue.Full:
            return False

    def start(self) -> None:
        with self._lock:
            if self.is_alive:
                return
            logger.debug("Starting background worker")
            self._thread = threading.Thread(
                target=self._run,
                name="scope3ai.BackgroundWorker",
                daemon=True,
            )
            try:
                self._thread.start()
            except RuntimeError:
                self._thread = None

    def kill(self) -> None:
        logger.debug("Got kill signal")
        with self._lock:
            if not self._thread:
                return
            try:
                self._queue.put_nowait(self.STOP_WORKER)
            except queue.Full:
                logger.debug("Failed to kill worker")
            except queue.ShutDown:
                logger.debug("Worker already shutdown")
            self._thread = None
            self._queue = queue.Queue(maxsize=self._size)

    def flush(self, timeout: float = 5) -> None:
        logger.debug("Got flush signal")
        with self._lock:
            if not self.is_alive:
                return
            self._wait_flush(timeout)
        logger.debug("Worker flushed")

    def _wait_flush(self, timeout: float) -> None:
        initial_timeout = min(0.1, timeout)
        if not self._timed_queue_join(initial_timeout):
            pending = self._queue.qsize() + 1
            logger.debug(f"{pending} event(s) pending on flush")

            if not self._timed_queue_join(timeout - initial_timeout):
                pending = self._queue.qsize() + 1
                logger.error(f"flush timed out, dropped {pending} events")

    def _timed_queue_join(self, timeout: float) -> bool:
        deadline = monotonic() + timeout
        queue = self._queue

        queue.all_tasks_done.acquire()

        try:
            while queue.unfinished_tasks:
                delay = deadline - monotonic()
                if delay <= 0:
                    return False
                queue.all_tasks_done.wait(timeout=delay)

            return True
        finally:
            queue.all_tasks_done.release()

    def _run(self) -> None:
        q = self._queue
        while True:
            callback = q.get()
            try:
                if callback is self.STOP_WORKER:
                    break
                self._pause_event.wait()
                try:
                    callback()
                except Exception:
                    logger.error("Failed processing job", exc_info=True)
            finally:
                q.task_done()
            sleep(0)

    def pause(self) -> None:
        self._pause_event.clear()

    def resume(self) -> None:
        self._pause_event.set()

1	import logging	4✔
2	import queue	4✔
3	import threading	4✔
4	from time import sleep	4✔
5	from typing import Callable, Optional	4✔
6
7	from time import monotonic	4✔
8
9	logger = logging.getLogger("scope3ai.worker")	4✔
10
11
12	class BackgroundWorker:	4✔
13	STOP_WORKER = object()	4✔
14
15	def __init__(self, size: int) -> None:	4✔
16	self._size = size	4✔
17	self._queue = queue.Queue(maxsize=size)	4✔
18	self._lock = threading.Lock()	4✔
19	self._thread: Optional[threading.Thread] = None	4✔
20	self._pause_event = threading.Event()	4✔
21	self._pause_event.set()	4✔
22
23	@property	4✔
24	def is_alive(self) -> bool:	4✔
25	return self._thread and self._thread.is_alive()	4✔
26
27	def _ensure_thread(self) -> None:	4✔
28	if not self.is_alive:	4✔
29	self.start()	4✔
30
31	def submit(self, callback: Callable[[], None]) -> bool:	4✔
32	self._ensure_thread()	4✔
33	try:	4✔
34	self._queue.put_nowait(callback)	4✔
35	return True	4✔
36	except queue.Full:	4✔
37	return False	4✔
38
39	def start(self) -> None:	4✔
40	with self._lock:	4✔
41	if self.is_alive:	4✔
UNCOV 42	return	×
43	logger.debug("Starting background worker")	4✔
44	self._thread = threading.Thread(	4✔
45	target=self._run,
46	name="scope3ai.BackgroundWorker",
47	daemon=True,
48	)
49	try:	4✔
50	self._thread.start()	4✔
51	except RuntimeError:	×
UNCOV 52	self._thread = None	×
53
54	def kill(self) -> None:	4✔
55	logger.debug("Got kill signal")	4✔
56	with self._lock:	4✔
57	if not self._thread:	4✔
58	return	4✔
59	try:	4✔
60	self._queue.put_nowait(self.STOP_WORKER)	4✔
61	except queue.Full:	×
62	logger.debug("Failed to kill worker")	×
63	except queue.ShutDown:	×
UNCOV 64	logger.debug("Worker already shutdown")	×
65	self._thread = None	4✔
66	self._queue = queue.Queue(maxsize=self._size)	4✔
67
68	def flush(self, timeout: float = 5) -> None:	4✔
69	logger.debug("Got flush signal")	4✔
70	with self._lock:	4✔
71	if not self.is_alive:	4✔
72	return	×
73	self._wait_flush(timeout)	4✔
74	logger.debug("Worker flushed")	4✔
75
76	def _wait_flush(self, timeout: float) -> None:	4✔
77	initial_timeout = min(0.1, timeout)	4✔
78	if not self._timed_queue_join(initial_timeout):	4✔
79	pending = self._queue.qsize() + 1	4✔
80	logger.debug(f"{pending} event(s) pending on flush")	4✔
81
82	if not self._timed_queue_join(timeout - initial_timeout):	4✔
UNCOV 83	pending = self._queue.qsize() + 1	×
UNCOV 84	logger.error(f"flush timed out, dropped {pending} events")	×
85
86	def _timed_queue_join(self, timeout: float) -> bool:	4✔
87	deadline = monotonic() + timeout	4✔
88	queue = self._queue	4✔
89
90	queue.all_tasks_done.acquire()	4✔
91
92	try:	4✔
93	while queue.unfinished_tasks:	4✔
94	delay = deadline - monotonic()	4✔
95	if delay <= 0:	4✔
96	return False	4✔
97	queue.all_tasks_done.wait(timeout=delay)	4✔
98
99	return True	4✔
100	finally:
101	queue.all_tasks_done.release()	4✔
102
103	def _run(self) -> None:	4✔
104	q = self._queue	4✔
105	while True:	3✔
106	callback = q.get()	4✔
107	try:	4✔
108	if callback is self.STOP_WORKER:	4✔
109	break	4✔
110	self._pause_event.wait()	4✔
111	try:	4✔
112	callback()	4✔
113	except Exception:	4✔
114	logger.error("Failed processing job", exc_info=True)	4✔
115	finally:
116	q.task_done()	4✔
117	sleep(0)	4✔
118
119	def pause(self) -> None:	4✔
120	self._pause_event.clear()	4✔
121
122	def resume(self) -> None:	4✔
123	self._pause_event.set()	4✔

scope3data / scope3ai-py / 12753874046

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous