From 2445d706df70768ed3fcd9d4ea11b4040375a2e3 Mon Sep 17 00:00:00 2001 From: George Tsiamasiotis Date: Tue, 17 Sep 2024 13:59:49 +0300 Subject: [PATCH] Fixed typing errors --- frigate/__main__.py | 2 +- frigate/log.py | 18 +++++++++++++----- 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/frigate/__main__.py b/frigate/__main__.py index 82be44f06..9c6607cad 100644 --- a/frigate/__main__.py +++ b/frigate/__main__.py @@ -7,7 +7,7 @@ from flask import cli from frigate.app import FrigateApp -def main(): +def main() -> None: faulthandler.enable() logging.basicConfig(level=logging.INFO, handlers=[]) diff --git a/frigate/log.py b/frigate/log.py index 9936a9104..30286eeb5 100644 --- a/frigate/log.py +++ b/frigate/log.py @@ -6,7 +6,10 @@ import threading from collections import deque from contextlib import AbstractContextManager, ContextDecorator from logging.handlers import QueueHandler, QueueListener -from typing import Deque +from types import TracebackType +from typing import Deque, Optional + +from typing_extensions import Self from frigate.util.builtin import clean_camera_user_pass @@ -30,7 +33,7 @@ class log_thread(AbstractContextManager, ContextDecorator): self._handler = handler - log_queue = mp.Queue() + log_queue: mp.Queue = mp.Queue() self._queue_handler = QueueHandler(log_queue) self._log_listener = QueueListener( @@ -41,10 +44,10 @@ class log_thread(AbstractContextManager, ContextDecorator): def handler(self) -> logging.Handler: return self._handler - def _stop_thread(self): + def _stop_thread(self) -> None: self._log_listener.stop() - def __enter__(self): + def __enter__(self) -> Self: logging.getLogger().addHandler(self._queue_handler) atexit.register(self._stop_thread) @@ -52,7 +55,12 @@ class log_thread(AbstractContextManager, ContextDecorator): return self - def __exit__(self, *exc): + def __exit__( + self, + exc_type: Optional[type[BaseException]], + exc_info: Optional[BaseException], + exc_tb: Optional[TracebackType], + ) -> None: logging.getLogger().removeHandler(self._queue_handler) atexit.unregister(self._stop_thread)