Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 42 additions & 4 deletions blinkpy/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -568,14 +568,14 @@ async def request_get_config(blink, network, camera_id, product_type="owl"):
:param blink: Blink instance.
:param network: Sync module network id.
:param camera_id: ID of camera
:param product_type: Camera product type "owl" or "catalina"
:param product_type: Camera product type "owl", "catalina", or "sedona"
"""
if product_type == "owl":
url = (
f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
f"/networks/{network}/owls/{camera_id}/config"
)
elif product_type == "catalina":
elif product_type in ["catalina", "sedona"]:
url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/config"
Comment on lines +578 to 579

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in commit 70046de — replaced int() ID comparison in snoozed with str() to prevent TypeError from aborting the search loop on missing/non-numeric IDs, and extended request_update_config() to accept sedona alongside catalina to match request_get_config().

else:
_LOGGER.info(
Expand All @@ -596,15 +596,15 @@ async def request_update_config(
:param blink: Blink instance.
:param network: Sync module network id.
:param camera_id: ID of camera
:param product_type: Camera product type "owl" or "catalina"
:param product_type: Camera product type "owl", "catalina", or "sedona"
:param data: string w/JSON dict of parameters/values to update
"""
if product_type == "owl":
url = (
f"{blink.urls.base_url}/api/v1/accounts/"
f"{blink.account_id}/networks/{network}/owls/{camera_id}/config"
)
elif product_type == "catalina":
elif product_type in ["catalina", "sedona"]:
url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/update"
else:
_LOGGER.info(
Expand Down Expand Up @@ -1063,3 +1063,41 @@ async def oauth_refresh_token(auth, refresh_token, hardware_id):

_LOGGER.error("OAuth token refresh failed with status %s", response.status)
return None


async def request_camera_snooze(
blink, network, camera_id, product_type="owl", data=None
):
"""
Update camera snooze configuration.

:param blink: Blink instance.
:param network: Sync module network id.
:param camera_id: ID of camera
:param product_type: Camera product type "owl", "catalina",
"doorbell", "hawk", "lotus", or "sedona"
:param data: string w/JSON dict of parameters/values to update
"""
product_lookup = {
"catalina": "cameras",
"sedona": "cameras",
"owl": "owls",
"hawk": "owls",
"doorbell": "doorbells",
"lotus": "doorbells",
}

if product_type not in product_lookup:
_LOGGER.info(
"Camera %s with product type %s snooze update not implemented.",
camera_id,
product_type,
)
return None

url_root = (
f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
f"/networks/{network}"
)
url = f"{url_root}/{product_lookup[product_type]}/{camera_id}/snooze"
return await http_post(blink, url, json=True, data=data)
55 changes: 55 additions & 0 deletions blinkpy/camera.py
Original file line number Diff line number Diff line change
Expand Up @@ -177,6 +177,61 @@ async def async_set_night_vision(self, value):
return await res.json()
return None

@property
async def snoozed(self):
"""Return snooze status as boolean."""
response_data = None
try:
if self.product_type in ["catalina", "sedona"]:
res = await api.request_get_config(
self.sync.blink,
self.network_id,
self.camera_id,
product_type=self.product_type,
)
response_data = res
snooze_value = res["camera"][0].get("snooze_till")
return bool(snooze_value)
else:
response_data = self.sync.blink.homescreen
if self.product_type in ["doorbell", "lotus"]:
collection_key = "doorbells"
else:
collection_key = "owls"
for device in self.sync.blink.homescreen.get(collection_key, []):
if str(device.get("id")) == str(self.camera_id):
snooze_value = device.get("snooze")
return bool(snooze_value)
return False
except TypeError:
return False
except (IndexError, KeyError, ValueError) as e:
_LOGGER.warning(
"Exception %s: Encountered a likely malformed response "
"from the snooze API endpoint. Response: %s",
e,
response_data,
)
return False

async def async_snooze(self, snooze_time=3600):
"""
Set camera snooze status.

:param snooze_time: Time in seconds to snooze camera. Default is 3600 (1 hour).
"""
data = dumps({"snooze_time": snooze_time})
res = await api.request_camera_snooze(
self.sync.blink,
self.network_id,
self.camera_id,
product_type=self.product_type,
data=data,
)
if res and self.product_type not in ["catalina", "sedona"]:
await self.sync.blink.get_homescreen()
return res

@property
def floodlight_enabled(self):
"""Return last-known floodlight state if tracked, else None."""
Expand Down
40 changes: 40 additions & 0 deletions tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,46 @@ async def test_request_update_config(self, mock_resp):
)
)

async def test_request_camera_snooze(self, mock_resp):
"""Test camera snooze request."""
mock_resp.return_value = {"message": "Camera snoozed"}
# Test catalina camera
response = await api.request_camera_snooze(
self.blink, "network", "camera_id", "catalina", '{"snooze_time": 300}'
)
self.assertEqual(response, {"message": "Camera snoozed"})
# Test sedona camera
response = await api.request_camera_snooze(
self.blink, "network", "camera_id", "sedona", '{"snooze_time": 300}'
)
self.assertEqual(response, {"message": "Camera snoozed"})
# Test owl camera
response = await api.request_camera_snooze(
self.blink, "network", "camera_id", "owl", '{"snooze_time": 300}'
)
self.assertEqual(response, {"message": "Camera snoozed"})
# Test hawk camera
response = await api.request_camera_snooze(
self.blink, "network", "camera_id", "hawk", '{"snooze_time": 300}'
)
self.assertEqual(response, {"message": "Camera snoozed"})
# Test doorbell camera
response = await api.request_camera_snooze(
self.blink, "network", "camera_id", "doorbell", '{"snooze_time": 300}'
)
self.assertEqual(response, {"message": "Camera snoozed"})
# Test lotus camera
response = await api.request_camera_snooze(
self.blink, "network", "camera_id", "lotus", '{"snooze_time": 300}'
)
self.assertEqual(response, {"message": "Camera snoozed"})
# Test unsupported camera type
self.assertIsNone(
await api.request_camera_snooze(
self.blink, "network", "camera_id", "unsupported_type"
)
)

async def test_wait_for_command(self, mock_resp):
"""Test Motion detect enable."""
mock_resp.side_effect = (COMMAND_NOT_COMPLETE, COMMAND_COMPLETE)
Expand Down
Loading
Loading