# Path: src/amcrest/exceptions.py
# class CommError(AmcrestError):
#     """A communication error occurred."""
# 
# Path: src/amcrest/http.py
# _LOGGER = logging.getLogger(__name__)
# _KEEPALIVE_OPTS = HTTPConnection.default_socket_options + [
#     (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
# ]
# class SOHTTPAdapter(HTTPAdapter):
# class Http:
#     def __init__(self, *args, **kwargs) -> None:
#     def init_poolmanager(self, *args, **kwargs) -> None:
#     def __init__(
#         self,
#         host: str,
#         port: int,
#         user: str,
#         password: str,
#         *,
#         verbose: bool = True,
#         protocol: str = "http",
#         ssl_verify: bool = True,
#         retries_connection: Optional[int] = None,
#         timeout_protocol: TimeoutT = None,
#     ) -> None:
#     def _generate_token(self) -> None:
#     async def _async_generate_token(self) -> None:
#     def __repr__(self) -> str:
#     def as_dict(self) -> dict:
#     def __base_url(self, param: str = "") -> str:
#     def get_base_url(self) -> str:
#     def command(self, *args, **kwargs) -> requests.Response:
#     async def async_command(self, *args, **kwargs) -> httpx.Response:
#     async def async_stream_command(
#         self, *args, **kwargs
#     ) -> AsyncIterator[httpx.Response]:
#     def _command(
#         self,
#         cmd: str,
#         retries: Optional[int] = None,
#         timeout_cmd: TimeoutT = None,
#         stream: bool = False,
#     ) -> requests.Response:
#     async def _async_command(
#         self,
#         cmd: str,
#         retries: Optional[int] = None,
#         timeout_cmd: TimeoutT = None,
#     ) -> httpx.Response:
#     async def _async_stream_command(
#         self,
#         cmd: str,
#         timeout_cmd: TimeoutT = None,
#     ) -> AsyncIterator[httpx.Response]:
#     def command_audio(
#         self,
#         cmd: str,
#         file_content,
#         http_header,
#         timeout: TimeoutT = None,
#     ) -> None:
#     def _get_config(self, config_name: str) -> str:
#     async def _async_get_config(self, config_name: str) -> str:
#     def _magic_box(self, action: str) -> str:
#     async def _async_magic_box(self, action: str) -> str:


import logging
import shutil
from typing import Optional, Union, overload
from typing_extensions import Literal
from urllib3.exceptions import HTTPError
from urllib3.response import HTTPResponse
from .exceptions import CommError
from .http import Http, TimeoutT
            channel:
                Video input channel number

                If no channel param is used, don't send channel parameter
                so camera will use its default channel

            path_file:
                If path_file is provided, save the snapshot
                in the path

        Return:
            raw from http request if stream is True
            response content if stream is False
        """
        cmd = "snapshot.cgi"
        if channel is not None:
            cmd += f"?channel={channel}"
        ret = self.command(cmd, timeout_cmd=timeout, stream=stream)

        if path_file:
            with open(path_file, "wb") as out_file:
                if stream:
                    try:
                        shutil.copyfileobj(ret.raw, out_file)
                    except HTTPError as error:
                        _LOGGER.debug(
                            "%s Snapshot to file failed due to error: %s",
                            self,
                            repr(error),
                        )

# Path: src/amcrest/exceptions.py
# class CommError(AmcrestError):
#     """A communication error occurred."""
# 
# Path: src/amcrest/http.py
# _LOGGER = logging.getLogger(__name__)
# _KEEPALIVE_OPTS = HTTPConnection.default_socket_options + [
#     (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
# ]
# class SOHTTPAdapter(HTTPAdapter):
# class Http:
#     def __init__(self, *args, **kwargs) -> None:
#     def init_poolmanager(self, *args, **kwargs) -> None:
#     def __init__(
#         self,
#         host: str,
#         port: int,
#         user: str,
#         password: str,
#         *,
#         verbose: bool = True,
#         protocol: str = "http",
#         ssl_verify: bool = True,
#         retries_connection: Optional[int] = None,
#         timeout_protocol: TimeoutT = None,
#     ) -> None:
#     def _generate_token(self) -> None:
#     async def _async_generate_token(self) -> None:
#     def __repr__(self) -> str:
#     def as_dict(self) -> dict:
#     def __base_url(self, param: str = "") -> str:
#     def get_base_url(self) -> str:
#     def command(self, *args, **kwargs) -> requests.Response:
#     async def async_command(self, *args, **kwargs) -> httpx.Response:
#     async def async_stream_command(
#         self, *args, **kwargs
#     ) -> AsyncIterator[httpx.Response]:
#     def _command(
#         self,
#         cmd: str,
#         retries: Optional[int] = None,
#         timeout_cmd: TimeoutT = None,
#         stream: bool = False,
#     ) -> requests.Response:
#     async def _async_command(
#         self,
#         cmd: str,
#         retries: Optional[int] = None,
#         timeout_cmd: TimeoutT = None,
#     ) -> httpx.Response:
#     async def _async_stream_command(
#         self,
#         cmd: str,
#         timeout_cmd: TimeoutT = None,
#     ) -> AsyncIterator[httpx.Response]:
#     def command_audio(
#         self,
#         cmd: str,
#         file_content,
#         http_header,
#         timeout: TimeoutT = None,
#     ) -> None:
#     def _get_config(self, config_name: str) -> str:
#     async def _async_get_config(self, config_name: str) -> str:
#     def _magic_box(self, action: str) -> str:
#     async def _async_magic_box(self, action: str) -> str:


# Path: src/amcrest/snapshot.py
import logging
import shutil
from typing import Optional, Union, overload
from typing_extensions import Literal
from urllib3.exceptions import HTTPError
from urllib3.response import HTTPResponse
from .exceptions import CommError
from .http import Http, TimeoutT

            channel:
                Video input channel number

                If no channel param is used, don't send channel parameter
                so camera will use its default channel

            path_file:
                If path_file is provided, save the snapshot
                in the path

        Return:
            raw from http request if stream is True
            response content if stream is False
        """
        cmd = "snapshot.cgi"
        if channel is not None:
            cmd += f"?channel={channel}"
        ret = self.command(cmd, timeout_cmd=timeout, stream=stream)

        if path_file:
            with open(path_file, "wb") as out_file:
                if stream:
                    try:
                        shutil.copyfileobj(ret.raw, out_file)
                    except HTTPError as error:
                        _LOGGER.debug(
                            "%s Snapshot to file failed due to error: %s",
                            self,
                            repr(error),
                        )

                        raise CommError(error) from error