mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 20:57:21 +00:00
Improve type hints in sighthound (#145432)
* Improve type hints in sighthound * More
This commit is contained in:
parent
5ddadcbd65
commit
687bedd251
@ -5,6 +5,7 @@ from __future__ import annotations
|
|||||||
import io
|
import io
|
||||||
import logging
|
import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
from PIL import Image, ImageDraw, UnidentifiedImageError
|
from PIL import Image, ImageDraw, UnidentifiedImageError
|
||||||
import simplehound.core as hound
|
import simplehound.core as hound
|
||||||
@ -59,8 +60,8 @@ def setup_platform(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Set up the platform."""
|
"""Set up the platform."""
|
||||||
# Validate credentials by processing image.
|
# Validate credentials by processing image.
|
||||||
api_key = config[CONF_API_KEY]
|
api_key: str = config[CONF_API_KEY]
|
||||||
account_type = config[CONF_ACCOUNT_TYPE]
|
account_type: str = config[CONF_ACCOUNT_TYPE]
|
||||||
api = hound.cloud(api_key, account_type)
|
api = hound.cloud(api_key, account_type)
|
||||||
try:
|
try:
|
||||||
api.detect(b"Test")
|
api.detect(b"Test")
|
||||||
@ -72,7 +73,8 @@ def setup_platform(
|
|||||||
save_file_folder = Path(save_file_folder)
|
save_file_folder = Path(save_file_folder)
|
||||||
|
|
||||||
entities = []
|
entities = []
|
||||||
for camera in config[CONF_SOURCE]:
|
source: list[dict[str, str]] = config[CONF_SOURCE]
|
||||||
|
for camera in source:
|
||||||
sighthound = SighthoundEntity(
|
sighthound = SighthoundEntity(
|
||||||
api,
|
api,
|
||||||
camera[CONF_ENTITY_ID],
|
camera[CONF_ENTITY_ID],
|
||||||
@ -91,29 +93,34 @@ class SighthoundEntity(ImageProcessingEntity):
|
|||||||
_attr_unit_of_measurement = ATTR_PEOPLE
|
_attr_unit_of_measurement = ATTR_PEOPLE
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self, api, camera_entity, name, save_file_folder, save_timestamped_file
|
self,
|
||||||
):
|
api: hound.cloud,
|
||||||
|
camera_entity: str,
|
||||||
|
name: str | None,
|
||||||
|
save_file_folder: Path | None,
|
||||||
|
save_timestamped_file: bool,
|
||||||
|
) -> None:
|
||||||
"""Init."""
|
"""Init."""
|
||||||
self._api = api
|
self._api = api
|
||||||
self._camera = camera_entity
|
self._attr_camera_entity = camera_entity
|
||||||
if name:
|
if name:
|
||||||
self._name = name
|
self._attr_name = name
|
||||||
else:
|
else:
|
||||||
camera_name = split_entity_id(camera_entity)[1]
|
camera_name = split_entity_id(camera_entity)[1]
|
||||||
self._name = f"sighthound_{camera_name}"
|
self._attr_name = f"sighthound_{camera_name}"
|
||||||
self._state = None
|
self._attr_state = None
|
||||||
self._last_detection = None
|
self._last_detection: str | None = None
|
||||||
self._image_width = None
|
self._image_width: int | None = None
|
||||||
self._image_height = None
|
self._image_height: int | None = None
|
||||||
self._save_file_folder = save_file_folder
|
self._save_file_folder = save_file_folder
|
||||||
self._save_timestamped_file = save_timestamped_file
|
self._save_timestamped_file = save_timestamped_file
|
||||||
|
|
||||||
def process_image(self, image):
|
def process_image(self, image: bytes) -> None:
|
||||||
"""Process an image."""
|
"""Process an image."""
|
||||||
detections = self._api.detect(image)
|
detections = self._api.detect(image)
|
||||||
people = hound.get_people(detections)
|
people = hound.get_people(detections)
|
||||||
self._state = len(people)
|
self._attr_state = len(people)
|
||||||
if self._state > 0:
|
if self._attr_state > 0:
|
||||||
self._last_detection = dt_util.now().strftime(DATETIME_FORMAT)
|
self._last_detection = dt_util.now().strftime(DATETIME_FORMAT)
|
||||||
|
|
||||||
metadata = hound.get_metadata(detections)
|
metadata = hound.get_metadata(detections)
|
||||||
@ -121,10 +128,10 @@ class SighthoundEntity(ImageProcessingEntity):
|
|||||||
self._image_height = metadata["image_height"]
|
self._image_height = metadata["image_height"]
|
||||||
for person in people:
|
for person in people:
|
||||||
self.fire_person_detected_event(person)
|
self.fire_person_detected_event(person)
|
||||||
if self._save_file_folder and self._state > 0:
|
if self._save_file_folder and self._attr_state > 0:
|
||||||
self.save_image(image, people, self._save_file_folder)
|
self.save_image(image, people, self._save_file_folder)
|
||||||
|
|
||||||
def fire_person_detected_event(self, person):
|
def fire_person_detected_event(self, person: dict[str, Any]) -> None:
|
||||||
"""Send event with detected total_persons."""
|
"""Send event with detected total_persons."""
|
||||||
self.hass.bus.fire(
|
self.hass.bus.fire(
|
||||||
EVENT_PERSON_DETECTED,
|
EVENT_PERSON_DETECTED,
|
||||||
@ -136,7 +143,9 @@ class SighthoundEntity(ImageProcessingEntity):
|
|||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
def save_image(self, image, people, directory):
|
def save_image(
|
||||||
|
self, image: bytes, people: list[dict[str, Any]], directory: Path
|
||||||
|
) -> None:
|
||||||
"""Save a timestamped image with bounding boxes around targets."""
|
"""Save a timestamped image with bounding boxes around targets."""
|
||||||
try:
|
try:
|
||||||
img = Image.open(io.BytesIO(bytearray(image))).convert("RGB")
|
img = Image.open(io.BytesIO(bytearray(image))).convert("RGB")
|
||||||
@ -145,37 +154,26 @@ class SighthoundEntity(ImageProcessingEntity):
|
|||||||
return
|
return
|
||||||
draw = ImageDraw.Draw(img)
|
draw = ImageDraw.Draw(img)
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
assert self._image_width is not None
|
||||||
|
assert self._image_height is not None
|
||||||
|
|
||||||
for person in people:
|
for person in people:
|
||||||
box = hound.bbox_to_tf_style(
|
box = hound.bbox_to_tf_style(
|
||||||
person["boundingBox"], self._image_width, self._image_height
|
person["boundingBox"], self._image_width, self._image_height
|
||||||
)
|
)
|
||||||
draw_box(draw, box, self._image_width, self._image_height)
|
draw_box(draw, box, self._image_width, self._image_height)
|
||||||
|
|
||||||
latest_save_path = directory / f"{self._name}_latest.jpg"
|
latest_save_path = directory / f"{self.name}_latest.jpg"
|
||||||
img.save(latest_save_path)
|
img.save(latest_save_path)
|
||||||
|
|
||||||
if self._save_timestamped_file:
|
if self._save_timestamped_file:
|
||||||
timestamp_save_path = directory / f"{self._name}_{self._last_detection}.jpg"
|
timestamp_save_path = directory / f"{self.name}_{self._last_detection}.jpg"
|
||||||
img.save(timestamp_save_path)
|
img.save(timestamp_save_path)
|
||||||
_LOGGER.debug("Sighthound saved file %s", timestamp_save_path)
|
_LOGGER.debug("Sighthound saved file %s", timestamp_save_path)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def camera_entity(self):
|
def extra_state_attributes(self) -> dict[str, str]:
|
||||||
"""Return camera entity id from process pictures."""
|
|
||||||
return self._camera
|
|
||||||
|
|
||||||
@property
|
|
||||||
def name(self):
|
|
||||||
"""Return the name of the sensor."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
|
||||||
def state(self):
|
|
||||||
"""Return the state of the entity."""
|
|
||||||
return self._state
|
|
||||||
|
|
||||||
@property
|
|
||||||
def extra_state_attributes(self):
|
|
||||||
"""Return the attributes."""
|
"""Return the attributes."""
|
||||||
if not self._last_detection:
|
if not self._last_detection:
|
||||||
return {}
|
return {}
|
||||||
|
Loading…
x
Reference in New Issue
Block a user