jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-17 10:17:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Retry failed Amcrest commands that change settings (#36917)

Browse Source
This commit is contained in:
Phil Bruckner 2020-06-22 20:36:41 -05:00 committed by GitHub
parent 8b21b415c4
commit d68148417f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 102 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

197
homeassistant/components/amcrest/camera.py
View File

@ -130,6 +130,10 @@ class CannotSnapshot(Exception):
"""Conditions are not valid for taking a snapshot.""" """Conditions are not valid for taking a snapshot."""
class AmcrestCommandFailed(Exception):
"""Amcrest camera command did not work."""
class AmcrestCam(Camera): class AmcrestCam(Camera):
"""An implementation of an Amcrest IP camera.""" """An implementation of an Amcrest IP camera."""
@ -367,12 +371,12 @@ class AmcrestCam(Camera):
self._model = resp.split("=")[-1] self._model = resp.split("=")[-1]
else: else:
self._model = "unknown" self._model = "unknown"
self.is_streaming = self._api.video_enabled self.is_streaming = self._get_video()
self._is_recording = self._api.record_mode == "Manual" self._is_recording = self._get_recording()
self._motion_detection_enabled = self._api.is_motion_detector_on() self._motion_detection_enabled = self._get_motion_detection()
self._audio_enabled = self._api.audio_enabled self._audio_enabled = self._get_audio()
self._motion_recording_enabled = self._api.is_record_on_motion_detection() self._motion_recording_enabled = self._get_motion_recording()
self._color_bw = _CBW[self._api.day_night_color] self._color_bw = self._get_color_mode()
self._rtsp_url = self._api.rtsp_url(typeno=self._resolution) self._rtsp_url = self._api.rtsp_url(typeno=self._resolution)
except AmcrestError as error: except AmcrestError as error:
log_update_error(_LOGGER, "get", self.name, "camera attributes", error) log_update_error(_LOGGER, "get", self.name, "camera attributes", error)
@ -384,11 +388,11 @@ class AmcrestCam(Camera):
def turn_off(self): def turn_off(self):
"""Turn off camera.""" """Turn off camera."""
self._enable_video_stream(False) self._enable_video(False)
def turn_on(self): def turn_on(self):
"""Turn on camera.""" """Turn on camera."""
self._enable_video_stream(True) self._enable_video(True)
def enable_motion_detection(self): def enable_motion_detection(self):
"""Enable motion detection in the camera.""" """Enable motion detection in the camera."""
@ -465,28 +469,53 @@ class AmcrestCam(Camera):
# Methods to send commands to Amcrest camera and handle errors # Methods to send commands to Amcrest camera and handle errors
def _enable_video_stream(self, enable): def _change_setting(self, value, attr, description, action="set"):
func = description.replace(" ", "_")
description = f"camera {description} to {value}"
tries = 3
while True:
try:
getattr(self, f"_set_{func}")(value)
new_value = getattr(self, f"_get_{func}")()
if new_value != value:
raise AmcrestCommandFailed
except (AmcrestError, AmcrestCommandFailed) as error:
if tries == 1:
log_update_error(_LOGGER, action, self.name, description, error)
return
log_update_error(
_LOGGER, action, self.name, description, error, logging.DEBUG
)
else:
if attr:
setattr(self, attr, new_value)
self.schedule_update_ha_state()
return
tries -= 1
def _get_video(self):
return self._api.video_enabled
def _set_video(self, enable):
self._api.video_enabled = enable
def _enable_video(self, enable):
"""Enable or disable camera video stream.""" """Enable or disable camera video stream."""
# Given the way the camera's state is determined by # Given the way the camera's state is determined by
# is_streaming and is_recording, we can't leave # is_streaming and is_recording, we can't leave
# recording on if video stream is being turned off. # recording on if video stream is being turned off.
if self.is_recording and not enable: if self.is_recording and not enable:
self._enable_recording(False) self._enable_recording(False)
try: self._change_setting(enable, "is_streaming", "video")
self._api.video_enabled = enable
except AmcrestError as error:
log_update_error(
_LOGGER,
"enable" if enable else "disable",
self.name,
"camera video stream",
error,
)
else:
self.is_streaming = enable
self.schedule_update_ha_state()
if self._control_light: if self._control_light:
self._enable_light(self._audio_enabled or self.is_streaming) self._change_light()
def _get_recording(self):
return self._api.record_mode == "Manual"
def _set_recording(self, enable):
rec_mode = {"Automatic": 0, "Manual": 1}
self._api.record_mode = rec_mode["Manual" if enable else "Automatic"]
def _enable_recording(self, enable): def _enable_recording(self, enable):
"""Turn recording on or off.""" """Turn recording on or off."""
@ -494,86 +523,56 @@ class AmcrestCam(Camera):
# is_streaming and is_recording, we can't leave # is_streaming and is_recording, we can't leave
# video stream off if recording is being turned on. # video stream off if recording is being turned on.
if not self.is_streaming and enable: if not self.is_streaming and enable:
self._enable_video_stream(True) self._enable_video(True)
rec_mode = {"Automatic": 0, "Manual": 1} self._change_setting(enable, "_is_recording", "recording")
try:
self._api.record_mode = rec_mode["Manual" if enable else "Automatic"] def _get_motion_detection(self):
except AmcrestError as error: return self._api.is_motion_detector_on()
log_update_error(
_LOGGER, def _set_motion_detection(self, enable):
"enable" if enable else "disable", self._api.motion_detection = str(enable).lower()
self.name,
"camera recording",
error,
)
else:
self._is_recording = enable
self.schedule_update_ha_state()
def _enable_motion_detection(self, enable): def _enable_motion_detection(self, enable):
"""Enable or disable motion detection.""" """Enable or disable motion detection."""
try: self._change_setting(enable, "_motion_detection_enabled", "motion detection")
self._api.motion_detection = str(enable).lower()
except AmcrestError as error: def _get_audio(self):
log_update_error( return self._api.audio_enabled
_LOGGER,
"enable" if enable else "disable", def _set_audio(self, enable):
self.name, self._api.audio_enabled = enable
"camera motion detection",
error,
)
else:
self._motion_detection_enabled = enable
self.schedule_update_ha_state()
def _enable_audio(self, enable): def _enable_audio(self, enable):
"""Enable or disable audio stream.""" """Enable or disable audio stream."""
try: self._change_setting(enable, "_audio_enabled", "audio")
self._api.audio_enabled = enable
except AmcrestError as error:
log_update_error(
_LOGGER,
"enable" if enable else "disable",
self.name,
"camera audio stream",
error,
)
else:
self._audio_enabled = enable
self.schedule_update_ha_state()
if self._control_light: if self._control_light:
self._enable_light(self._audio_enabled or self.is_streaming) self._change_light()
def _enable_light(self, enable): def _get_indicator_light(self):
return "true" in self._api.command(
"configManager.cgi?action=getConfig&name=LightGlobal"
).content.decode("utf-8")
def _set_indicator_light(self, enable):
self._api.command(
f"configManager.cgi?action=setConfig&LightGlobal[0].Enable={str(enable).lower()}"
)
def _change_light(self):
"""Enable or disable indicator light.""" """Enable or disable indicator light."""
try: self._change_setting(
self._api.command( self._audio_enabled or self.is_streaming, None, "indicator light"
f"configManager.cgi?action=setConfig&LightGlobal[0].Enable={str(enable).lower()}" )
)
except AmcrestError as error: def _get_motion_recording(self):
log_update_error( return self._api.is_record_on_motion_detection()
_LOGGER,
"enable" if enable else "disable", def _set_motion_recording(self, enable):
self.name, self._api.motion_recording = str(enable).lower()
"indicator light",
error,
)
def _enable_motion_recording(self, enable): def _enable_motion_recording(self, enable):
"""Enable or disable motion recording.""" """Enable or disable motion recording."""
try: self._change_setting(enable, "_motion_recording_enabled", "motion recording")
self._api.motion_recording = str(enable).lower()
except AmcrestError as error:
log_update_error(
_LOGGER,
"enable" if enable else "disable",
self.name,
"camera motion recording",
error,
)
else:
self._motion_recording_enabled = enable
self.schedule_update_ha_state()
def _goto_preset(self, preset): def _goto_preset(self, preset):
"""Move camera position and zoom to preset.""" """Move camera position and zoom to preset."""
@ -584,17 +583,15 @@ class AmcrestCam(Camera):
_LOGGER, "move", self.name, f"camera to preset {preset}", error _LOGGER, "move", self.name, f"camera to preset {preset}", error
) )
def _get_color_mode(self):
return _CBW[self._api.day_night_color]
def _set_color_mode(self, cbw):
self._api.day_night_color = _CBW.index(cbw)
def _set_color_bw(self, cbw): def _set_color_bw(self, cbw):
"""Set camera color mode.""" """Set camera color mode."""
try: self._change_setting(cbw, "_color_bw", "color mode")
self._api.day_night_color = _CBW.index(cbw)
except AmcrestError as error:
log_update_error(
_LOGGER, "set", self.name, f"camera color mode to {cbw}", error
)
else:
self._color_bw = cbw
self.schedule_update_ha_state()
def _start_tour(self, start): def _start_tour(self, start):
"""Start camera tour.""" """Start camera tour."""

7
homeassistant/components/amcrest/helpers.py
View File

@ -1,4 +1,6 @@
"""Helpers for amcrest component.""" """Helpers for amcrest component."""
import logging
from .const import DOMAIN from .const import DOMAIN
@ -7,9 +9,10 @@ def service_signal(service, *args):
return "_".join([DOMAIN, service, *args]) return "_".join([DOMAIN, service, *args])
def log_update_error(logger, action, name, entity_type, error): def log_update_error(logger, action, name, entity_type, error, level=logging.ERROR):
"""Log an update error.""" """Log an update error."""
logger.error( logger.log(
level,
"Could not %s %s %s due to error: %s", "Could not %s %s %s due to error: %s",
action, action,
name, name,