administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-02-05 10:45:21 +03:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

treewide: fix issues detected by ruff

Browse Source
This commit is contained in:
Martin Weinelt 2023-05-23 02:38:21 +02:00
parent b339f53bb4
commit 2807a229a2
No known key found for this signature in database
GPG Key ID: 87C1E9888F856759
39 changed files with 124 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
benchmark.py
View File

@ -53,7 +53,7 @@ def start(id, num_detections, detection_queue, event):
frame_times = []
for x in range(0, num_detections):
start_frame = datetime.datetime.now().timestamp()
detections = object_detector.detect(my_frame)
object_detector.detect(my_frame)
frame_times.append(datetime.datetime.now().timestamp() - start_frame)
duration = datetime.datetime.now().timestamp() - start

10
frigate/__main__.py
View File

@ -1,13 +1,15 @@
import faulthandler
from flask import cli
faulthandler.enable()
import threading
threading.current_thread().name = "frigate"
from flask import cli
from frigate.app import FrigateApp
faulthandler.enable()
threading.current_thread().name = "frigate"
cli.show_server_banner = lambda *x: None
if __name__ == "__main__":

8
frigate/app.py
View File

@ -133,10 +133,10 @@ class FrigateApp:
for log, level in self.config.logger.logs.items():
logging.getLogger(log).setLevel(level.value.upper())
if not "werkzeug" in self.config.logger.logs:
if "werkzeug" not in self.config.logger.logs:
logging.getLogger("werkzeug").setLevel("ERROR")
if not "ws4py" in self.config.logger.logs:
if "ws4py" not in self.config.logger.logs:
logging.getLogger("ws4py").setLevel("ERROR")
def init_queues(self) -> None:
@ -294,7 +294,7 @@ class FrigateApp:
def start_video_output_processor(self) -> None:
output_processor = mp.Process(
target=output_frames,
name=f"output_processor",
name="output_processor",
args=(
self.config,
self.video_output_queue,
@ -467,7 +467,7 @@ class FrigateApp:
self.stop()
def stop(self) -> None:
logger.info(f"Stopping...")
logger.info("Stopping...")
self.stop_event.set()
for detector in self.detectors.values():

8
frigate/comms/dispatcher.py
View File

@ -72,7 +72,7 @@ class Dispatcher:
camera_name = topic.split("/")[-3]
command = topic.split("/")[-2]
self._camera_settings_handlers[command](camera_name, payload)
except IndexError as e:
except IndexError:
logger.error(f"Received invalid set command: {topic}")
return
elif topic.endswith("ptz"):
@ -80,7 +80,7 @@ class Dispatcher:
# example /cam_name/ptz payload=MOVE_UP|MOVE_DOWN|STOP...
camera_name = topic.split("/")[-2]
self._on_ptz_command(camera_name, payload)
except IndexError as e:
except IndexError:
logger.error(f"Received invalid ptz command: {topic}")
return
elif topic == "restart":
@ -128,7 +128,7 @@ class Dispatcher:
elif payload == "OFF":
if self.camera_metrics[camera_name]["detection_enabled"].value:
logger.error(
f"Turning off motion is not allowed when detection is enabled."
"Turning off motion is not allowed when detection is enabled."
)
return
@ -196,7 +196,7 @@ class Dispatcher:
if payload == "ON":
if not self.config.cameras[camera_name].record.enabled_in_config:
logger.error(
f"Recordings must be enabled in the config to be turned on via MQTT."
"Recordings must be enabled in the config to be turned on via MQTT."
)
return

10
frigate/comms/mqtt.py
View File

@ -177,10 +177,10 @@ class MqttClient(Communicator): # type: ignore[misc]
f"{self.mqtt_config.topic_prefix}/restart", self.on_mqtt_command
)
if not self.mqtt_config.tls_ca_certs is None:
if self.mqtt_config.tls_ca_certs is not None:
if (
not self.mqtt_config.tls_client_cert is None
and not self.mqtt_config.tls_client_key is None
self.mqtt_config.tls_client_cert is not None
and self.mqtt_config.tls_client_key is not None
):
self.client.tls_set(
self.mqtt_config.tls_ca_certs,
@ -189,9 +189,9 @@ class MqttClient(Communicator): # type: ignore[misc]
)
else:
self.client.tls_set(self.mqtt_config.tls_ca_certs)
if not self.mqtt_config.tls_insecure is None:
if self.mqtt_config.tls_insecure is not None:
self.client.tls_insecure_set(self.mqtt_config.tls_insecure)
if not self.mqtt_config.user is None:
if self.mqtt_config.user is not None:
self.client.username_pw_set(
self.mqtt_config.user, password=self.mqtt_config.password
)

4
frigate/comms/ws.py
View File

@ -45,7 +45,7 @@ class WebSocketClient(Communicator): # type: ignore[misc]
"topic": json_message.get("topic"),
"payload": json_message.get("payload"),
}
except Exception as e:
except Exception:
logger.warning(
f"Unable to parse websocket message as valid json: {message.data.decode('utf-8')}"
)
@ -82,7 +82,7 @@ class WebSocketClient(Communicator): # type: ignore[misc]
"payload": payload,
}
)
except Exception as e:
except Exception:
# if the payload can't be decoded don't relay to clients
logger.debug(f"payload for {topic} wasn't text. Skipping...")
return

13
frigate/config.py
View File

@ -8,7 +8,6 @@ from typing import Dict, List, Optional, Tuple, Union
import matplotlib.pyplot as plt
import numpy as np
import yaml
from pydantic import BaseModel, Extra, Field, validator, parse_obj_as
from pydantic.fields import PrivateAttr
@ -26,7 +25,6 @@ from frigate.util import (
get_ffmpeg_arg_list,
escape_special_characters,
load_config_with_no_duplicates,
load_labels,
)
from frigate.ffmpeg_presets import (
parse_preset_hardware_acceleration_decode,
@ -36,12 +34,9 @@ from frigate.ffmpeg_presets import (
parse_preset_output_rtmp,
)
from frigate.detectors import (
PixelFormatEnum,
InputTensorEnum,
ModelConfig,
DetectorConfig,
)
from frigate.version import VERSION
logger = logging.getLogger(__name__)
@ -487,7 +482,7 @@ class CameraFfmpegConfig(FfmpegConfig):
if len(roles) > len(roles_set):
raise ValueError("Each input role may only be used once.")
if not "detect" in roles:
if "detect" not in roles:
raise ValueError("The detect role is required.")
return v
@ -776,12 +771,12 @@ def verify_config_roles(camera_config: CameraConfig) -> None:
set([r for i in camera_config.ffmpeg.inputs for r in i.roles])
)
if camera_config.record.enabled and not "record" in assigned_roles:
if camera_config.record.enabled and "record" not in assigned_roles:
raise ValueError(
f"Camera {camera_config.name} has record enabled, but record is not assigned to an input."
)
if camera_config.rtmp.enabled and not "rtmp" in assigned_roles:
if camera_config.rtmp.enabled and "rtmp" not in assigned_roles:
raise ValueError(
f"Camera {camera_config.name} has rtmp enabled, but rtmp is not assigned to an input."
)
@ -1062,7 +1057,7 @@ class FrigateConfig(FrigateBaseModel):
config.model.dict(exclude_unset=True),
)
if not "path" in merged_model:
if "path" not in merged_model:
if detector_config.type == "cpu":
merged_model["path"] = "/cpu_model.tflite"
elif detector_config.type == "edgetpu":

6
frigate/detectors/__init__.py
View File

@ -1,11 +1,5 @@
import logging
from .detection_api import DetectionApi
from .detector_config import (
PixelFormatEnum,
InputTensorEnum,
ModelConfig,
)
from .detector_types import DetectorTypeEnum, api_types, DetectorConfig

4
frigate/detectors/detector_config.py
View File

@ -3,12 +3,12 @@ import json
import logging
from enum import Enum
import os
from typing import Dict, List, Optional, Tuple, Union, Literal
from typing import Dict, Optional, Tuple
import requests
import matplotlib.pyplot as plt
from pydantic import BaseModel, Extra, Field, validator
from pydantic import BaseModel, Extra, Field
from pydantic.fields import PrivateAttr
from frigate.plus import PlusApi

3
frigate/detectors/plugins/cpu_tfl.py
View File

@ -3,8 +3,9 @@ import numpy as np
from frigate.detectors.detection_api import DetectionApi
from frigate.detectors.detector_config import BaseDetectorConfig
from typing_extensions import Literal
from pydantic import Extra, Field
from pydantic import Field
try:
from tflite_runtime.interpreter import Interpreter

6
frigate/detectors/plugins/deepstack.py
View File

@ -6,7 +6,7 @@ import io
from frigate.detectors.detection_api import DetectionApi
from frigate.detectors.detector_config import BaseDetectorConfig
from typing_extensions import Literal
from pydantic import Extra, Field
from pydantic import Field
from PIL import Image
@ -64,11 +64,11 @@ class DeepStack(DetectionApi):
for i, detection in enumerate(response_json.get("predictions")):
logger.debug(f"Response: {detection}")
if detection["confidence"] < 0.4:
logger.debug(f"Break due to confidence < 0.4")
logger.debug("Break due to confidence < 0.4")
break
label = self.get_label_index(detection["label"])
if label < 0:
logger.debug(f"Break due to unknown label")
logger.debug("Break due to unknown label")
break
detections[i] = [
label,

2
frigate/detectors/plugins/edgetpu_tfl.py
View File

@ -4,7 +4,7 @@ import numpy as np
from frigate.detectors.detection_api import DetectionApi
from frigate.detectors.detector_config import BaseDetectorConfig
from typing_extensions import Literal
from pydantic import Extra, Field
from pydantic import Field
try:
from tflite_runtime.interpreter import Interpreter, load_delegate

2
frigate/detectors/plugins/openvino.py
View File

@ -5,7 +5,7 @@ import openvino.runtime as ov
from frigate.detectors.detection_api import DetectionApi
from frigate.detectors.detector_config import BaseDetectorConfig, ModelTypeEnum
from typing_extensions import Literal
from pydantic import Extra, Field
from pydantic import Field
logger = logging.getLogger(__name__)

4
frigate/detectors/plugins/tensorrt.py
View File

@ -8,7 +8,7 @@ try:
from cuda import cuda
TRT_SUPPORT = True
except ModuleNotFoundError as e:
except ModuleNotFoundError:
TRT_SUPPORT = False
from frigate.detectors.detection_api import DetectionApi
@ -172,7 +172,7 @@ class TensorRtDetector(DetectionApi):
if not self.context.execute_async_v2(
bindings=self.bindings, stream_handle=self.stream
):
logger.warn(f"Execute returned false")
logger.warn("Execute returned false")
# Transfer predictions back from the GPU.
[

35
frigate/events/cleanup.py
View File

@ -7,7 +7,6 @@ import threading
from pathlib import Path
from peewee import fn
from frigate.config import FrigateConfig
from frigate.const import CLIPS_DIR
@ -45,9 +44,9 @@ class EventCleanup(threading.Thread):
)
# loop over object types in db
for l in distinct_labels:
for event in distinct_labels:
# get expiration time for this label
expire_days = retain_config.objects.get(l.label, retain_config.default)
expire_days = retain_config.objects.get(event.label, retain_config.default)
expire_after = (
datetime.datetime.now() - datetime.timedelta(days=expire_days)
).timestamp()
@ -55,8 +54,8 @@ class EventCleanup(threading.Thread):
expired_events = Event.select().where(
Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
Event.label == event.label,
Event.retain_indefinitely is False,
)
# delete the media from disk
for event in expired_events:
@ -75,8 +74,8 @@ class EventCleanup(threading.Thread):
update_query = Event.update(update_params).where(
Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
Event.label == event.label,
Event.retain_indefinitely is False,
)
update_query.execute()
@ -92,9 +91,9 @@ class EventCleanup(threading.Thread):
)
# loop over object types in db
for l in distinct_labels:
for event in distinct_labels:
# get expiration time for this label
expire_days = retain_config.objects.get(l.label, retain_config.default)
expire_days = retain_config.objects.get(event.label, retain_config.default)
expire_after = (
datetime.datetime.now() - datetime.timedelta(days=expire_days)
).timestamp()
@ -102,8 +101,8 @@ class EventCleanup(threading.Thread):
expired_events = Event.select().where(
Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
Event.label == event.label,
Event.retain_indefinitely is False,
)
# delete the grabbed clips from disk
for event in expired_events:
@ -121,8 +120,8 @@ class EventCleanup(threading.Thread):
update_query = Event.update(update_params).where(
Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
Event.label == event.label,
Event.retain_indefinitely is False,
)
update_query.execute()
@ -131,9 +130,9 @@ class EventCleanup(threading.Thread):
select id,
label,
camera,
has_snapshot,
has_clip,
row_number() over (
has_snapshot,
has_clip,
row_number() over (
partition by label, camera, round(start_time/5,0)*5
order by end_time-start_time desc
) as copy_number
@ -169,8 +168,8 @@ class EventCleanup(threading.Thread):
# drop events from db where has_clip and has_snapshot are false
delete_query = Event.delete().where(
Event.has_clip == False, Event.has_snapshot == False
Event.has_clip is False, Event.has_snapshot is False
)
delete_query.execute()
logger.info(f"Exiting event cleanup...")
logger.info("Exiting event cleanup...")

1
frigate/events/external.py
View File

@ -3,7 +3,6 @@
import base64
import cv2
import datetime
import glob
import logging
import os
import random

7
frigate/events/maintainer.py
View File

@ -5,7 +5,6 @@ import threading
from enum import Enum
from peewee import fn
from frigate.config import EventsConfig, FrigateConfig
from frigate.models import Event
@ -65,7 +64,7 @@ class EventProcessor(threading.Thread):
def run(self) -> None:
# set an end_time on events without an end_time on startup
Event.update(end_time=Event.start_time + 30).where(
Event.end_time == None
Event.end_time is None
).execute()
while not self.stop_event.is_set():
@ -99,9 +98,9 @@ class EventProcessor(threading.Thread):
# set an end_time on events without an end_time before exiting
Event.update(end_time=datetime.datetime.now().timestamp()).where(
Event.end_time == None
Event.end_time is None
).execute()
logger.info(f"Exiting event processor...")
logger.info("Exiting event processor...")
def handle_object_detection(
self,

43
frigate/http.py
View File

@ -105,10 +105,10 @@ def events_summary():
clauses = []
if not has_clip is None:
if has_clip is not None:
clauses.append((Event.has_clip == has_clip))
if not has_snapshot is None:
if has_snapshot is not None:
clauses.append((Event.has_snapshot == has_snapshot))
if len(clauses) == 0:
@ -253,7 +253,7 @@ def send_to_plus(id):
event.plus_id = plus_id
event.save()
if not include_annotation is None:
if include_annotation is not None:
box = event.data["box"]
try:
@ -296,12 +296,12 @@ def false_positive(id):
# events from before the conversion to relative dimensions cant include annotations
if event.data.get("box") is None:
message = f"Events prior to 0.13 cannot be submitted as false positives"
message = "Events prior to 0.13 cannot be submitted as false positives"
logger.error(message)
return make_response(jsonify({"success": False, "message": message}), 400)
if event.false_positive:
message = f"False positive already submitted to Frigate+"
message = "False positive already submitted to Frigate+"
logger.error(message)
return make_response(jsonify({"success": False, "message": message}), 400)
@ -437,7 +437,7 @@ def get_sub_labels():
parts = label.split(",")
for part in parts:
if not (part.strip()) in sub_labels:
if part.strip() not in sub_labels:
sub_labels.append(part.strip())
sub_labels.sort()
@ -476,7 +476,7 @@ def event_thumbnail(id, max_cache_age=2592000):
event_complete = False
try:
event = Event.get(Event.id == id)
if not event.end_time is None:
if event.end_time is not None:
event_complete = True
thumbnail_bytes = base64.b64decode(event.thumbnail)
except DoesNotExist:
@ -486,7 +486,7 @@ def event_thumbnail(id, max_cache_age=2592000):
for camera_state in camera_states:
if id in camera_state.tracked_objects:
tracked_obj = camera_state.tracked_objects.get(id)
if not tracked_obj is None:
if tracked_obj is not None:
thumbnail_bytes = tracked_obj.get_thumbnail()
except:
return "Event not found", 404
@ -593,7 +593,7 @@ def event_snapshot(id):
event_complete = False
jpg_bytes = None
try:
event = Event.get(Event.id == id, Event.end_time != None)
event = Event.get(Event.id == id, Event.end_time is not None)
event_complete = True
if not event.has_snapshot:
return "Snapshot not available", 404
@ -609,7 +609,7 @@ def event_snapshot(id):
for camera_state in camera_states:
if id in camera_state.tracked_objects:
tracked_obj = camera_state.tracked_objects.get(id)
if not tracked_obj is None:
if tracked_obj is not None:
jpg_bytes = tracked_obj.get_jpg_bytes(
timestamp=request.args.get("timestamp", type=int),
bounding_box=request.args.get("bbox", type=int),
@ -645,7 +645,7 @@ def label_snapshot(camera_name, label):
event_query = (
Event.select()
.where(Event.camera == camera_name)
.where(Event.has_snapshot == True)
.where(Event.has_snapshot is True)
.order_by(Event.start_time.desc())
)
else:
@ -653,7 +653,7 @@ def label_snapshot(camera_name, label):
Event.select()
.where(Event.camera == camera_name)
.where(Event.label == label)
.where(Event.has_snapshot == True)
.where(Event.has_snapshot is True)
.order_by(Event.start_time.desc())
)
@ -820,13 +820,13 @@ def events():
if before:
clauses.append((Event.start_time < before))
if not has_clip is None:
if has_clip is not None:
clauses.append((Event.has_clip == has_clip))
if not has_snapshot is None:
if has_snapshot is not None:
clauses.append((Event.has_snapshot == has_snapshot))
if not in_progress is None:
if in_progress is not None:
clauses.append((Event.end_time.is_null(in_progress)))
if not include_thumbnails:
@ -899,7 +899,7 @@ def end_event(event_id):
{"success": False, "message": f"{event_id} must be set and valid."}, 404
)
return jsonify({"success": True, "message": f"Event successfully ended."}, 200)
return jsonify({"success": True, "message": "Event successfully ended."}, 200)
@bp.route("/config")
@ -959,9 +959,8 @@ def config_save():
# Validate the config schema
try:
new_yaml = FrigateConfig.parse_raw(new_config)
check_runtime = new_yaml.runtime_config
except Exception as e:
FrigateConfig.parse_raw(new_config)
except Exception:
return make_response(
jsonify(
{
@ -985,12 +984,12 @@ def config_save():
with open(config_file, "w") as f:
f.write(new_config)
f.close()
except Exception as e:
except Exception:
return make_response(
jsonify(
{
"success": False,
"message": f"Could not write config file, be sure that Frigate has write permission on the config file.",
"message": "Could not write config file, be sure that Frigate has write permission on the config file.",
}
),
400,
@ -1531,7 +1530,7 @@ def ffprobe():
if not path_param:
return jsonify(
{"success": False, "message": f"Path needs to be provided."}, "404"
{"success": False, "message": "Path needs to be provided."}, "404"
)
if path_param.startswith("camera"):

5
frigate/log.py
View File

@ -10,8 +10,7 @@ from logging import handlers
from typing import Optional
from types import FrameType
from setproctitle import setproctitle
from typing import Deque, Optional
from types import FrameType
from typing import Deque
from collections import deque
from frigate.util import clean_camera_user_pass
@ -44,7 +43,7 @@ def root_configurer(queue: Queue) -> None:
def log_process(log_queue: Queue) -> None:
threading.current_thread().name = f"logger"
threading.current_thread().name = "logger"
setproctitle("frigate.logger")
listener_configurer()

1
frigate/models.py
View File

@ -1,4 +1,3 @@
from numpy import unique
from peewee import (
Model,
CharField,

2
frigate/object_detection.py
View File

@ -161,7 +161,7 @@ class ObjectDetectProcess:
def start_or_restart(self):
self.detection_start.value = 0.0
if (not self.detect_process is None) and self.detect_process.is_alive():
if (self.detect_process is not None) and self.detect_process.is_alive():
self.stop()
self.detect_process = mp.Process(
target=run_detector,

10
frigate/object_processing.py
View File

@ -141,7 +141,7 @@ class TrackedObject:
# check each zone
for name, zone in self.camera_config.zones.items():
# if the zone is not for this object type, skip
if len(zone.objects) > 0 and not obj_data["label"] in zone.objects:
if len(zone.objects) > 0 and obj_data["label"] not in zone.objects:
continue
contour = zone.contour
# check if the object is in the zone
@ -177,9 +177,9 @@ class TrackedObject:
return (thumb_update, significant_change)
def to_dict(self, include_thumbnail: bool = False):
snapshot_time = (
(
self.thumbnail_data["frame_time"]
if not self.thumbnail_data is None
if self.thumbnail_data is not None
else 0.0
)
event = {
@ -526,7 +526,7 @@ class CameraState:
for id in removed_ids:
# publish events to mqtt
removed_obj = tracked_objects[id]
if not "end_time" in removed_obj.obj_data:
if "end_time" not in removed_obj.obj_data:
removed_obj.obj_data["end_time"] = frame_time
for c in self.callbacks["end"]:
c(self.name, removed_obj, frame_time)
@ -1028,4 +1028,4 @@ class TrackedObjectProcessor(threading.Thread):
event_id, camera = self.event_processed_queue.get()
self.camera_states[camera].finished(event_id)
logger.info(f"Exiting object processor...")
logger.info("Exiting object processor...")

9
frigate/objects.py
View File

@ -1,14 +1,7 @@
import copy
import datetime
import itertools
import multiprocessing as mp
import random
import string
import threading
import time
from collections import defaultdict
import cv2
import numpy as np
from scipy.spatial import distance as dist
@ -160,7 +153,7 @@ class ObjectTracker:
# update any tracked objects with labels that are not
# seen in the current objects and deregister if needed
for obj in list(self.tracked_objects.values()):
if not obj["label"] in new_object_groups:
if obj["label"] not in new_object_groups:
if self.disappeared[obj["id"]] >= self.max_disappeared:
self.deregister(obj["id"])
else:

13
frigate/output.py
View File

@ -4,7 +4,6 @@ import logging
import math
import multiprocessing as mp
import os
import operator
import queue
import signal
import subprocess as sp
@ -185,7 +184,7 @@ class BirdsEyeFrameManager:
if len(logo_files) > 0:
birdseye_logo = cv2.imread(logo_files[0], cv2.IMREAD_UNCHANGED)
if not birdseye_logo is None:
if birdseye_logo is not None:
transparent_layer = birdseye_logo[:, :, 3]
y_offset = height // 2 - transparent_layer.shape[0] // 2
x_offset = width // 2 - transparent_layer.shape[1] // 2
@ -229,7 +228,7 @@ class BirdsEyeFrameManager:
self.last_output_time = 0.0
def clear_frame(self):
logger.debug(f"Clearing the birdseye frame")
logger.debug("Clearing the birdseye frame")
self.frame[:] = self.blank_frame
def copy_to_position(self, position, camera=None, frame_time=None):
@ -301,7 +300,7 @@ class BirdsEyeFrameManager:
# reset the layout if it needs to be different
if layout_dim != self.layout_dim or reset_layout:
if reset_layout:
logger.debug(f"Added new cameras, resetting layout...")
logger.debug("Added new cameras, resetting layout...")
logger.debug(f"Changing layout size from {self.layout_dim} to {layout_dim}")
self.layout_dim = layout_dim
@ -385,7 +384,7 @@ class BirdsEyeFrameManager:
]
# if not an empty spot and the camera has a newer frame, copy it
elif (
not camera is None
camera is not None
and self.cameras[camera]["current_frame"]
!= self.cameras[camera]["layout_frame"]
):
@ -423,8 +422,8 @@ class BirdsEyeFrameManager:
def output_frames(config: FrigateConfig, video_output_queue):
threading.current_thread().name = f"output"
setproctitle(f"frigate.output")
threading.current_thread().name = "output"
setproctitle("frigate.output")
stop_event = mp.Event()

2
frigate/plus.py
View File

@ -3,7 +3,7 @@ import json
import logging
import os
import re
from typing import Any, Dict, List
from typing import Any, List
import requests
from frigate.const import PLUS_ENV_VAR, PLUS_API_HOST
from requests.models import Response

2
frigate/ptz.py
View File

@ -145,7 +145,7 @@ class OnvifController:
onvif.get_service("ptz").ContinuousMove(move_request)
def _move_to_preset(self, camera_name: str, preset: str) -> None:
if not preset in self.cams[camera_name]["presets"]:
if preset not in self.cams[camera_name]["presets"]:
logger.error(f"{preset} is not a valid preset for {camera_name}")
return

4
frigate/record/cleanup.py
View File

@ -225,7 +225,7 @@ class RecordingCleanup(threading.Thread):
recordings_to_delete = []
for recording in recordings.objects().iterator():
if not recording.path in files_on_disk:
if recording.path not in files_on_disk:
recordings_to_delete.append(recording.id)
logger.debug(
@ -247,7 +247,7 @@ class RecordingCleanup(threading.Thread):
# Expire tmp clips every minute, recordings and clean directories every hour.
for counter in itertools.cycle(range(self.config.record.expire_interval)):
if self.stop_event.wait(60):
logger.info(f"Exiting recording cleanup...")
logger.info("Exiting recording cleanup...")
break
self.clean_tmp_clips()

6
frigate/record/maintainer.py
View File

@ -115,7 +115,7 @@ class RecordingMaintainer(threading.Thread):
Event.select()
.where(
Event.camera == camera,
(Event.end_time == None)
(Event.end_time is None)
| (Event.end_time >= recordings[0]["start_time"].timestamp()),
Event.has_clip,
)
@ -127,7 +127,7 @@ class RecordingMaintainer(threading.Thread):
# Just delete files if recordings are turned off
if (
not camera in self.config.cameras
camera not in self.config.cameras
or not self.process_info[camera]["record_enabled"].value
):
Path(cache_path).unlink(missing_ok=True)
@ -394,4 +394,4 @@ class RecordingMaintainer(threading.Thread):
duration = datetime.datetime.now().timestamp() - run_start
wait_time = max(0, 5 - duration)
logger.info(f"Exiting recording maintenance...")
logger.info("Exiting recording maintenance...")

2
frigate/stats.py
View File

@ -308,4 +308,4 @@ class StatsEmitter(threading.Thread):
)
self.dispatcher.publish("stats", json.dumps(stats), retain=False)
logger.debug("Finished stats collection")
logger.info(f"Exiting stats emitter...")
logger.info("Exiting stats emitter...")

4
frigate/storage.py
View File

@ -107,7 +107,7 @@ class StorageMaintainer(threading.Thread):
retained_events: Event = (
Event.select()
.where(
Event.retain_indefinitely == True,
Event.retain_indefinitely is True,
Event.has_clip,
)
.order_by(Event.start_time.asc())
@ -188,4 +188,4 @@ class StorageMaintainer(threading.Thread):
if self.check_storage_needs_cleanup():
self.reduce_storage_consumption()
logger.info(f"Exiting storage maintainer...")
logger.info("Exiting storage maintainer...")

6
frigate/test/test_config.py
View File

@ -675,7 +675,7 @@ class TestConfig(unittest.TestCase):
runtime_config = frigate_config.runtime_config()
ffmpeg_cmds = runtime_config.cameras["back"].ffmpeg_cmds
assert len(ffmpeg_cmds) == 1
assert not "clips" in ffmpeg_cmds[0]["roles"]
assert "clips" not in ffmpeg_cmds[0]["roles"]
def test_max_disappeared_default(self):
config = {
@ -986,7 +986,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
runtime_config = frigate_config.runtime_config()
frigate_config.runtime_config()
def test_global_detect(self):
config = {
@ -1145,7 +1145,7 @@ class TestConfig(unittest.TestCase):
assert config == frigate_config.dict(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert runtime_config.cameras["back"].snapshots.bounding_box == False
assert runtime_config.cameras["back"].snapshots.bounding_box is False
assert runtime_config.cameras["back"].snapshots.height == 150
assert runtime_config.cameras["back"].snapshots.enabled

2
frigate/test/test_gpu_stats.py
View File

@ -1,7 +1,7 @@
import unittest
from unittest.mock import MagicMock, patch
from frigate.util import get_amd_gpu_stats, get_intel_gpu_stats, get_nvidia_gpu_stats
from frigate.util import get_amd_gpu_stats, get_intel_gpu_stats
class TestGpuStats(unittest.TestCase):

12
frigate/test/test_http.py
View File

@ -128,22 +128,22 @@ class TestHttp(unittest.TestCase):
with app.test_client() as client:
_insert_mock_event(id)
events = client.get(f"/events").json
events = client.get("/events").json
assert events
assert len(events) == 1
assert events[0]["id"] == id
_insert_mock_event(id2)
events = client.get(f"/events").json
events = client.get("/events").json
assert events
assert len(events) == 2
events = client.get(
f"/events",
"/events",
query_string={"limit": 1},
).json
assert events
assert len(events) == 1
events = client.get(
f"/events",
"/events",
query_string={"has_clip": 0},
).json
assert not events
@ -230,12 +230,12 @@ class TestHttp(unittest.TestCase):
event = client.get(f"/events/{id}").json
assert event
assert event["id"] == id
assert event["retain_indefinitely"] == True
assert event["retain_indefinitely"] is True
client.delete(f"/events/{id}/retain")
event = client.get(f"/events/{id}").json
assert event
assert event["id"] == id
assert event["retain_indefinitely"] == False
assert event["retain_indefinitely"] is False
def test_set_delete_sub_label(self):
app = create_app(

1
frigate/test/test_reduce_boxes.py
View File

@ -1,4 +1,3 @@
import numpy as np
from unittest import TestCase, main
from frigate.video import box_overlaps, reduce_boxes

5
frigate/test/test_storage.py
View File

@ -1,18 +1,15 @@
import datetime
import json
import logging
import os
import unittest
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
from peewee import DoesNotExist
from peewee_migrate import Router
from playhouse.sqlite_ext import SqliteExtDatabase
from playhouse.sqliteq import SqliteQueueDatabase
from playhouse.shortcuts import model_to_dict
from frigate.config import FrigateConfig
from frigate.http import create_app
from frigate.models import Event, Recordings
from frigate.storage import StorageMaintainer

2
frigate/test/test_yuv_region_2_rgb.py
View File

@ -33,7 +33,7 @@ class TestYuvRegion2RGB(TestCase):
# cv2.imwrite(f"bgr_frame.jpg", self.bgr_frame)
yuv_frame = cv2.cvtColor(bgr_frame, cv2.COLOR_BGR2YUV_I420)
cropped = yuv_region_2_rgb(yuv_frame, (0, 852, 648, 1500))
yuv_region_2_rgb(yuv_frame, (0, 852, 648, 1500))
# cv2.imwrite(f"cropped.jpg", cv2.cvtColor(cropped, cv2.COLOR_RGB2BGR))

3
frigate/util.py
View File

@ -20,7 +20,6 @@ import py3nvml.py3nvml as nvml
import cv2
import numpy as np
import os
import psutil
import pytz
@ -457,7 +456,7 @@ def copy_yuv_to_position(
# clear v2
destination_frame[v2[1] : v2[3], v2[0] : v2[2]] = 128
if not source_frame is None:
if source_frame is not None:
# calculate the resized frame, maintaining the aspect ratio
source_aspect_ratio = source_frame.shape[1] / (source_frame.shape[0] // 3 * 2)
dest_aspect_ratio = destination_shape[1] / destination_shape[0]

14
frigate/video.py
View File

@ -45,7 +45,7 @@ def filtered(obj, objects_to_track, object_filters):
object_area = obj[3]
object_ratio = obj[4]
if not object_name in objects_to_track:
if object_name not in objects_to_track:
return True
if object_name in object_filters:
@ -73,7 +73,7 @@ def filtered(obj, objects_to_track, object_filters):
if obj_settings.max_ratio < object_ratio:
return True
if not obj_settings.mask is None:
if obj_settings.mask is not None:
# compute the coordinates of the object and make sure
# the location isn't outside the bounds of the image (can happen from rounding)
object_xmin = object_box[0]
@ -169,20 +169,20 @@ def capture_frames(
skipped_eps.start()
while True:
fps.value = frame_rate.eps()
skipped_fps = skipped_eps.eps()
skipped_eps.eps()
current_frame.value = datetime.datetime.now().timestamp()
frame_name = f"{camera_name}{current_frame.value}"
frame_buffer = frame_manager.create(frame_name, frame_size)
try:
frame_buffer[:] = ffmpeg_process.stdout.read(frame_size)
except Exception as e:
except Exception:
# shutdown has been initiated
if stop_event.is_set():
break
logger.error(f"{camera_name}: Unable to read frames from ffmpeg process.")
if ffmpeg_process.poll() != None:
if ffmpeg_process.poll() is not None:
logger.error(
f"{camera_name}: ffmpeg process is not running. exiting capture thread..."
)
@ -604,7 +604,7 @@ def process_frames(
while not stop_event.is_set():
if exit_on_empty and frame_queue.empty():
logger.info(f"Exiting track_objects...")
logger.info("Exiting track_objects...")
break
try:
@ -655,7 +655,7 @@ def process_frames(
tracked_object_boxes = [
obj["box"]
for obj in object_tracker.tracked_objects.values()
if not obj["id"] in stationary_object_ids
if obj["id"] not in stationary_object_ids
]
# combine motion boxes with known locations of existing objects

4
frigate/watchdog.py
View File

@ -2,8 +2,6 @@ import datetime
import logging
import threading
import time
import os
import signal
from frigate.object_detection import ObjectDetectProcess
from frigate.util import restart_frigate
@ -39,4 +37,4 @@ class FrigateWatchdog(threading.Thread):
logger.info("Detection appears to have stopped. Exiting Frigate...")
restart_frigate()
logger.info(f"Exiting watchdog...")
logger.info("Exiting watchdog...")