Merge pull request #210 from flacjacket/stream_set

Allow specifying stream to turn audio and video on or off

src/amcrest/audio.py

@@ -195,14 +195,22 @@ async def async_is_audio_enabled(
  )
  return is_enabled[channel]
 
- def set_audio_enabled(self, enable: bool, *, channel: int = 0) -> None:
+ def set_audio_enabled(
+ self, enable: bool, *, channel: int = 0, stream: str = "Main"
+ ) -> None:
  """Enable/disable all audio streams on given channel."""
- self.command(utils.enable_audio_video_cmd("Audio", enable, channel))
+ self.command(
+ utils.enable_audio_video_cmd(
+ "Audio", enable, channel, stream=stream
+ )
+ )
 
  async def async_set_audio_enabled(
- self, enable: bool, *, channel: int = 0
+ self, enable: bool, *, channel: int = 0, stream: str = "Main"
  ) -> None:
  """Enable/disable all audio streams on given channel."""
  await self.async_command(
- utils.enable_audio_video_cmd("Audio", enable, channel)
+ utils.enable_audio_video_cmd(
+ "Audio", enable, channel, stream=stream
+ )
  )

src/amcrest/http.py

@@ -335,7 +335,7 @@ async def _async_command(
  follow_redirects=True,
  auth=self._async_token,
  verify=self._verify,
- timeout=httpx_timeout
+ timeout=httpx_timeout,
  ) as client:
  for loop in range(1, 2 + retries):
  _LOGGER.debug(
@@ -399,7 +399,7 @@ async def _async_stream_command(
  follow_redirects=True,
  auth=self._async_token,
  verify=self._verify,
- timeout=httpx_timeout
+ timeout=httpx_timeout,
  ) as client:
  async with client.stream("GET", url) as resp:
  try:

src/amcrest/utils.py

@@ -11,6 +11,7 @@
 
 import re
 from datetime import datetime
+from typing import List
 
 # pylint: disable=no-name-in-module
 from distutils import util
@@ -89,12 +90,19 @@ def extract_audio_video_enabled(param: str, resp: str) -> List[bool]:
  return parts
 
 
-def enable_audio_video_cmd(param: str, enable: bool, channel: int) -> str:
+def enable_audio_video_cmd(
+ param: str, enable: bool, channel: int, *, stream: str
+) -> str:
  """Return command to enable/disable all audio/video streams."""
  formats = [("Extra", 3), ("Main", 4)]
  if param == "Video":
  formats.append(("Snap", 3))
 
+ if stream is not None:
+ formats = [x for x in formats if x[0] == stream]
+ if not formats:
+ raise RuntimeError(f"Bad stream specified: {stream}")
+
  set_enable = str(enable).lower()
  cmds = ["configManager.cgi?action=setConfig"]
  cmds.extend(

src/amcrest/video.py

@@ -380,15 +380,23 @@ async def async_is_video_enabled(
  )
  return is_enabled[channel]
 
- def set_video_enabled(self, enable: bool, channel: int) -> None:
- self.command(utils.enable_audio_video_cmd("Video", enable, channel))
+ def set_video_enabled(
+ self, enable: bool, channel: int, *, stream: str = "Main"
+ ) -> None:
+ self.command(
+ utils.enable_audio_video_cmd(
+ "Video", enable, channel, stream=stream
+ )
+ )
  self.set_smart_ir(enable, channel)
 
  async def async_set_video_enabled(
- self, enable: bool, channel: int
+ self, enable: bool, channel: int, *, stream: str = "Main"
  ) -> None:
  await self.async_command(
- utils.enable_audio_video_cmd("Video", enable, channel)
+ utils.enable_audio_video_cmd(
+ "Video", enable, channel, stream=stream
+ )
  )
  await self.async_set_smart_ir(enable, channel)