Camera Status and Motion record status (#44936)

This commit is contained in:
Ryan Fleming 2021-01-27 04:50:44 -05:00 committed by GitHub
parent 67b309394f
commit 459236fcdd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 55 additions and 2 deletions

View file

@ -1,4 +1,5 @@
"""Support for Ubiquiti's UVC cameras.""" """Support for Ubiquiti's UVC cameras."""
from datetime import datetime
import logging import logging
import re import re
@ -109,10 +110,28 @@ class UnifiVideoCamera(Camera):
return 0 return 0
@property
def state_attributes(self):
"""Return the camera state attributes."""
attr = super().state_attributes
if self.motion_detection_enabled:
attr["last_recording_start_time"] = timestamp_ms_to_date(
self._caminfo["lastRecordingStartTime"]
)
return attr
@property @property
def is_recording(self): def is_recording(self):
"""Return true if the camera is recording.""" """Return true if the camera is recording."""
return self._caminfo["recordingSettings"]["fullTimeRecordEnabled"] recording_state = "DISABLED"
if "recordingIndicator" in self._caminfo.keys():
recording_state = self._caminfo["recordingIndicator"]
return (
self._caminfo["recordingSettings"]["fullTimeRecordEnabled"]
or recording_state == "MOTION_INPROGRESS"
or recording_state == "MOTION_FINISHED"
)
@property @property
def motion_detection_enabled(self): def motion_detection_enabled(self):
@ -235,3 +254,9 @@ class UnifiVideoCamera(Camera):
def update(self): def update(self):
"""Update the info.""" """Update the info."""
self._caminfo = self._nvr.get_camera(self._uuid) self._caminfo = self._nvr.get_camera(self._uuid)
def timestamp_ms_to_date(epoch_ms) -> datetime or None:
"""Convert millisecond timestamp to datetime."""
if epoch_ms:
return datetime.fromtimestamp(epoch_ms / 1000)

View file

@ -1,4 +1,5 @@
"""The tests for UVC camera module.""" """The tests for UVC camera module."""
from datetime import datetime
import socket import socket
import unittest import unittest
from unittest import mock from unittest import mock
@ -198,10 +199,14 @@ class TestUVC(unittest.TestCase):
self.nvr.get_camera.return_value = { self.nvr.get_camera.return_value = {
"model": "UVC Fake", "model": "UVC Fake",
"uuid": "06e3ff29-8048-31c2-8574-0852d1bd0e03", "uuid": "06e3ff29-8048-31c2-8574-0852d1bd0e03",
"recordingSettings": {"fullTimeRecordEnabled": True}, "recordingSettings": {
"fullTimeRecordEnabled": True,
"motionRecordEnabled": False,
},
"host": "host-a", "host": "host-a",
"internalHost": "host-b", "internalHost": "host-b",
"username": "admin", "username": "admin",
"lastRecordingStartTime": 1610070992367,
"channels": [ "channels": [
{ {
"id": "0", "id": "0",
@ -241,6 +246,29 @@ class TestUVC(unittest.TestCase):
assert SUPPORT_STREAM == self.uvc.supported_features assert SUPPORT_STREAM == self.uvc.supported_features
assert "uuid" == self.uvc.unique_id assert "uuid" == self.uvc.unique_id
def test_motion_recording_mode_properties(self):
"""Test the properties."""
self.nvr.get_camera.return_value["recordingSettings"][
"fullTimeRecordEnabled"
] = False
self.nvr.get_camera.return_value["recordingSettings"][
"motionRecordEnabled"
] = True
assert not self.uvc.is_recording
assert (
datetime(2021, 1, 8, 1, 56, 32, 367000)
== self.uvc.state_attributes["last_recording_start_time"]
)
self.nvr.get_camera.return_value["recordingIndicator"] = "DISABLED"
assert not self.uvc.is_recording
self.nvr.get_camera.return_value["recordingIndicator"] = "MOTION_INPROGRESS"
assert self.uvc.is_recording
self.nvr.get_camera.return_value["recordingIndicator"] = "MOTION_FINISHED"
assert self.uvc.is_recording
def test_stream(self): def test_stream(self):
"""Test the RTSP stream URI.""" """Test the RTSP stream URI."""
stream_source = yield from self.uvc.stream_source() stream_source = yield from self.uvc.stream_source()