administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2025-12-07 05:54:10 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

ruff format

Browse Source
This commit is contained in:
Abinila Siva 2025-11-24 11:51:17 -05:00
parent 4570808030
commit 14f7d27274
2 changed files with 8 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
frigate/detectors/plugins/memryx.py
View File

@ -54,7 +54,7 @@ class MemryXDetector(DetectionApi):
"MemryX SDK is not installed. Install it and set up MIX environment." "MemryX SDK is not installed. Install it and set up MIX environment."
) )
return return
# Get stop_event from detector_config # Get stop_event from detector_config
self.stop_event = getattr(detector_config, "_stop_event", stop_event) self.stop_event = getattr(detector_config, "_stop_event", stop_event)
@ -372,10 +372,7 @@ class MemryXDetector(DetectionApi):
return None return None
try: try:
# Wait for a frame from the queue with timeout to check stop_event periodically # Wait for a frame from the queue with timeout to check stop_event periodically
frame = self.capture_queue.get( frame = self.capture_queue.get(block=True, timeout=0.5)
block=True,
timeout=0.5
)
return frame return frame
@ -388,18 +385,15 @@ class MemryXDetector(DetectionApi):
def receive_output(self): def receive_output(self):
"""Retrieve processed results from MemryX output queue + a copy of the original frame""" """Retrieve processed results from MemryX output queue + a copy of the original frame"""
try: try:
# Get connection ID with timeout # Get connection ID with timeout
connection_id = ( connection_id = self.capture_id_queue.get(
self.capture_id_queue.get( block=True, timeout=1.0
block=True,
timeout=1.0
)
) # Get the corresponding connection ID ) # Get the corresponding connection ID
detections = self.output_queue.get() # Get detections from MemryX detections = self.output_queue.get() # Get detections from MemryX
return connection_id, detections return connection_id, detections
except Exception as e: except Exception as e:
# On timeout or stop event, return None # On timeout or stop event, return None
if self.stop_event.is_set(): if self.stop_event.is_set():
@ -409,7 +403,6 @@ class MemryXDetector(DetectionApi):
logger.warning(f"[receive_output] Error receiving output: {e}") logger.warning(f"[receive_output] Error receiving output: {e}")
return None, None return None, None
def post_process_yolonas(self, output): def post_process_yolonas(self, output):
predictions = output[0] predictions = output[0]
@ -857,7 +850,7 @@ class MemryXDetector(DetectionApi):
raise Exception( raise Exception(
f"{self.memx_model_type} is currently not supported for memryx. See the docs for more info on supported models." f"{self.memx_model_type} is currently not supported for memryx. See the docs for more info on supported models."
) )
def shutdown(self): def shutdown(self):
"""Gracefully shutdown the MemryX accelerator""" """Gracefully shutdown the MemryX accelerator"""
try: try:

2
frigate/object_detection/base.py
View File

@ -336,7 +336,7 @@ class ObjectDetectProcess:
# if the process has already exited on its own, just return # if the process has already exited on its own, just return
if self.detect_process and self.detect_process.exitcode: if self.detect_process and self.detect_process.exitcode:
return return
logging.info("Waiting for detection process to exit gracefully...") logging.info("Waiting for detection process to exit gracefully...")
self.detect_process.join(timeout=30) self.detect_process.join(timeout=30)
if self.detect_process.exitcode is None: if self.detect_process.exitcode is None: