mirror of
https://github.com/python-kasa/python-kasa.git
synced 2024-12-22 19:23:34 +00:00
Merge remote-tracking branch 'upstream/master' into feat/light_module_feats
This commit is contained in:
commit
cc9930ecc6
@ -25,7 +25,7 @@ from typing import Any
|
||||
|
||||
import asyncclick as click
|
||||
|
||||
from devtools.helpers.smartcamerarequests import SMARTCAMERA_REQUESTS
|
||||
from devtools.helpers.smartcamrequests import SMARTCAM_REQUESTS
|
||||
from devtools.helpers.smartrequests import SmartRequest, get_component_requests
|
||||
from kasa import (
|
||||
AuthenticationError,
|
||||
@ -42,19 +42,19 @@ from kasa.deviceconfig import DeviceEncryptionType, DeviceFamily
|
||||
from kasa.discover import DiscoveryResult
|
||||
from kasa.exceptions import SmartErrorCode
|
||||
from kasa.protocols import IotProtocol
|
||||
from kasa.protocols.smartcameraprotocol import (
|
||||
SmartCameraProtocol,
|
||||
from kasa.protocols.smartcamprotocol import (
|
||||
SmartCamProtocol,
|
||||
_ChildCameraProtocolWrapper,
|
||||
)
|
||||
from kasa.protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper
|
||||
from kasa.smart import SmartChildDevice, SmartDevice
|
||||
from kasa.smartcamera import SmartCamera
|
||||
from kasa.smartcam import SmartCamDevice
|
||||
|
||||
Call = namedtuple("Call", "module method")
|
||||
FixtureResult = namedtuple("FixtureResult", "filename, folder, data")
|
||||
|
||||
SMART_FOLDER = "tests/fixtures/smart/"
|
||||
SMARTCAMERA_FOLDER = "tests/fixtures/smartcamera/"
|
||||
SMARTCAM_FOLDER = "tests/fixtures/smartcam/"
|
||||
SMART_CHILD_FOLDER = "tests/fixtures/smart/child/"
|
||||
IOT_FOLDER = "tests/fixtures/iot/"
|
||||
|
||||
@ -65,7 +65,7 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SmartCall:
|
||||
"""Class for smart and smartcamera calls."""
|
||||
"""Class for smart and smartcam calls."""
|
||||
|
||||
module: str
|
||||
request: dict
|
||||
@ -562,7 +562,7 @@ async def _make_requests_or_exit(
|
||||
# Calling close on child protocol wrappers is a noop
|
||||
protocol_to_close = protocol
|
||||
if child_device_id:
|
||||
if isinstance(protocol, SmartCameraProtocol):
|
||||
if isinstance(protocol, SmartCamProtocol):
|
||||
protocol = _ChildCameraProtocolWrapper(child_device_id, protocol)
|
||||
else:
|
||||
protocol = _ChildProtocolWrapper(child_device_id, protocol)
|
||||
@ -608,7 +608,7 @@ async def get_smart_camera_test_calls(protocol: SmartProtocol):
|
||||
successes: list[SmartCall] = []
|
||||
|
||||
test_calls = []
|
||||
for request in SMARTCAMERA_REQUESTS:
|
||||
for request in SMARTCAM_REQUESTS:
|
||||
method = next(iter(request))
|
||||
if method == "get":
|
||||
module = method + "_" + next(iter(request[method]))
|
||||
@ -693,7 +693,7 @@ async def get_smart_camera_test_calls(protocol: SmartProtocol):
|
||||
click.echo(f"Skipping {component_id}..", nl=False)
|
||||
click.echo(click.style("UNSUPPORTED", fg="yellow"))
|
||||
else: # Not a smart protocol device so assume camera protocol
|
||||
for request in SMARTCAMERA_REQUESTS:
|
||||
for request in SMARTCAM_REQUESTS:
|
||||
method = next(iter(request))
|
||||
if method == "get":
|
||||
method = method + "_" + next(iter(request[method]))
|
||||
@ -858,7 +858,7 @@ async def get_smart_fixtures(
|
||||
protocol: SmartProtocol, *, discovery_info: dict[str, Any] | None, batch_size: int
|
||||
) -> list[FixtureResult]:
|
||||
"""Get fixture for new TAPO style protocol."""
|
||||
if isinstance(protocol, SmartCameraProtocol):
|
||||
if isinstance(protocol, SmartCamProtocol):
|
||||
test_calls, successes = await get_smart_camera_test_calls(protocol)
|
||||
child_wrapper: type[_ChildProtocolWrapper | _ChildCameraProtocolWrapper] = (
|
||||
_ChildCameraProtocolWrapper
|
||||
@ -991,8 +991,8 @@ async def get_smart_fixtures(
|
||||
copy_folder = SMART_FOLDER
|
||||
else:
|
||||
# smart camera protocol
|
||||
model_info = SmartCamera._get_device_info(final, discovery_info)
|
||||
copy_folder = SMARTCAMERA_FOLDER
|
||||
model_info = SmartCamDevice._get_device_info(final, discovery_info)
|
||||
copy_folder = SMARTCAM_FOLDER
|
||||
hw_version = model_info.hardware_version
|
||||
sw_version = model_info.firmware_version
|
||||
model = model_info.long_name
|
||||
|
@ -13,7 +13,7 @@ from typing import Any, NamedTuple
|
||||
from kasa.device_type import DeviceType
|
||||
from kasa.iot import IotDevice
|
||||
from kasa.smart import SmartDevice
|
||||
from kasa.smartcamera import SmartCamera
|
||||
from kasa.smartcam import SmartCamDevice
|
||||
|
||||
|
||||
class SupportedVersion(NamedTuple):
|
||||
@ -48,7 +48,7 @@ README_FILENAME = "README.md"
|
||||
IOT_FOLDER = "tests/fixtures/iot/"
|
||||
SMART_FOLDER = "tests/fixtures/smart/"
|
||||
SMART_CHILD_FOLDER = "tests/fixtures/smart/child"
|
||||
SMARTCAMERA_FOLDER = "tests/fixtures/smartcamera/"
|
||||
SMARTCAM_FOLDER = "tests/fixtures/smartcam/"
|
||||
|
||||
|
||||
def generate_supported(args):
|
||||
@ -65,7 +65,7 @@ def generate_supported(args):
|
||||
_get_supported_devices(supported, IOT_FOLDER, IotDevice)
|
||||
_get_supported_devices(supported, SMART_FOLDER, SmartDevice)
|
||||
_get_supported_devices(supported, SMART_CHILD_FOLDER, SmartDevice)
|
||||
_get_supported_devices(supported, SMARTCAMERA_FOLDER, SmartCamera)
|
||||
_get_supported_devices(supported, SMARTCAM_FOLDER, SmartCamDevice)
|
||||
|
||||
readme_updated = _update_supported_file(
|
||||
README_FILENAME, _supported_summary(supported), print_diffs
|
||||
@ -208,7 +208,7 @@ def _supported_text(
|
||||
def _get_supported_devices(
|
||||
supported: dict[str, Any],
|
||||
fixture_location: str,
|
||||
device_cls: type[IotDevice | SmartDevice | SmartCamera],
|
||||
device_cls: type[IotDevice | SmartDevice | SmartCamDevice],
|
||||
):
|
||||
for file in Path(fixture_location).glob("*.json"):
|
||||
with file.open() as f:
|
||||
|
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
SMARTCAMERA_REQUESTS: list[dict] = [
|
||||
SMARTCAM_REQUESTS: list[dict] = [
|
||||
{"getAlertTypeList": {"msg_alarm": {"name": "alert_type"}}},
|
||||
{"getNightVisionCapability": {"image_capability": {"name": ["supplement_lamp"]}}},
|
||||
{"getDeviceInfo": {"device_info": {"name": ["basic_info"]}}},
|
@ -24,9 +24,9 @@ from .protocols import (
|
||||
IotProtocol,
|
||||
SmartProtocol,
|
||||
)
|
||||
from .protocols.smartcameraprotocol import SmartCameraProtocol
|
||||
from .protocols.smartcamprotocol import SmartCamProtocol
|
||||
from .smart import SmartDevice
|
||||
from .smartcamera.smartcamera import SmartCamera
|
||||
from .smartcam import SmartCamDevice
|
||||
from .transports import (
|
||||
AesTransport,
|
||||
BaseTransport,
|
||||
@ -151,10 +151,10 @@ def get_device_class_from_family(
|
||||
"SMART.TAPOSWITCH": SmartDevice,
|
||||
"SMART.KASAPLUG": SmartDevice,
|
||||
"SMART.TAPOHUB": SmartDevice,
|
||||
"SMART.TAPOHUB.HTTPS": SmartCamera,
|
||||
"SMART.TAPOHUB.HTTPS": SmartCamDevice,
|
||||
"SMART.KASAHUB": SmartDevice,
|
||||
"SMART.KASASWITCH": SmartDevice,
|
||||
"SMART.IPCAMERA.HTTPS": SmartCamera,
|
||||
"SMART.IPCAMERA.HTTPS": SmartCamDevice,
|
||||
"IOT.SMARTPLUGSWITCH": IotPlug,
|
||||
"IOT.SMARTBULB": IotBulb,
|
||||
}
|
||||
@ -189,7 +189,7 @@ def get_protocol(
|
||||
"IOT.KLAP": (IotProtocol, KlapTransport),
|
||||
"SMART.AES": (SmartProtocol, AesTransport),
|
||||
"SMART.KLAP": (SmartProtocol, KlapTransportV2),
|
||||
"SMART.AES.HTTPS": (SmartCameraProtocol, SslAesTransport),
|
||||
"SMART.AES.HTTPS": (SmartCamProtocol, SslAesTransport),
|
||||
}
|
||||
if not (prot_tran_cls := supported_device_protocols.get(protocol_transport_key)):
|
||||
return None
|
||||
|
@ -60,7 +60,7 @@ if TYPE_CHECKING:
|
||||
from .device import Device
|
||||
from .iot import modules as iot
|
||||
from .smart import modules as smart
|
||||
from .smartcamera import modules as smartcamera
|
||||
from .smartcam import modules as smartcam
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -139,8 +139,8 @@ class Module(ABC):
|
||||
)
|
||||
TriggerLogs: Final[ModuleName[smart.TriggerLogs]] = ModuleName("TriggerLogs")
|
||||
|
||||
# SMARTCAMERA only modules
|
||||
Camera: Final[ModuleName[smartcamera.Camera]] = ModuleName("Camera")
|
||||
# SMARTCAM only modules
|
||||
Camera: Final[ModuleName[smartcam.Camera]] = ModuleName("Camera")
|
||||
|
||||
def __init__(self, device: Device, module: str) -> None:
|
||||
self._device = device
|
||||
|
@ -1,4 +1,4 @@
|
||||
"""Module for SmartCamera Protocol."""
|
||||
"""Module for SmartCamProtocol."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@ -46,8 +46,8 @@ class SingleRequest:
|
||||
request: dict[str, Any]
|
||||
|
||||
|
||||
class SmartCameraProtocol(SmartProtocol):
|
||||
"""Class for SmartCamera Protocol."""
|
||||
class SmartCamProtocol(SmartProtocol):
|
||||
"""Class for SmartCam Protocol."""
|
||||
|
||||
async def _handle_response_lists(
|
||||
self, response_result: dict[str, Any], method: str, retry_count: int
|
||||
@ -123,7 +123,7 @@ class SmartCameraProtocol(SmartProtocol):
|
||||
"""
|
||||
method = request
|
||||
method_type = request[:3]
|
||||
snake_name = SmartCameraProtocol._make_snake_name(request)
|
||||
snake_name = SmartCamProtocol._make_snake_name(request)
|
||||
param = snake_name[4:]
|
||||
if (
|
||||
(short_method := method[:3])
|
@ -168,7 +168,7 @@ class SmartProtocol(BaseProtocol):
|
||||
]
|
||||
|
||||
end = len(multi_requests)
|
||||
# The SmartCameraProtocol sends requests with a length 1 as a
|
||||
# The SmartCamProtocol sends requests with a length 1 as a
|
||||
# multipleRequest. The SmartProtocol doesn't so will never
|
||||
# raise_on_error
|
||||
raise_on_error = end == 1
|
||||
|
5
kasa/smartcam/__init__.py
Normal file
5
kasa/smartcam/__init__.py
Normal file
@ -0,0 +1,5 @@
|
||||
"""Package for supporting tapo-branded cameras."""
|
||||
|
||||
from .smartcamdevice import SmartCamDevice
|
||||
|
||||
__all__ = ["SmartCamDevice"]
|
@ -1,4 +1,4 @@
|
||||
"""Modules for SMARTCAMERA devices."""
|
||||
"""Modules for SMARTCAM devices."""
|
||||
|
||||
from .alarm import Alarm
|
||||
from .camera import Camera
|
@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from ...feature import Feature
|
||||
from ..smartcameramodule import SmartCameraModule
|
||||
from ..smartcammodule import SmartCamModule
|
||||
|
||||
DURATION_MIN = 0
|
||||
DURATION_MAX = 6000
|
||||
@ -12,11 +12,11 @@ VOLUME_MIN = 0
|
||||
VOLUME_MAX = 10
|
||||
|
||||
|
||||
class Alarm(SmartCameraModule):
|
||||
class Alarm(SmartCamModule):
|
||||
"""Implementation of alarm module."""
|
||||
|
||||
# Needs a different name to avoid clashing with SmartAlarm
|
||||
NAME = "SmartCameraAlarm"
|
||||
NAME = "SmartCamAlarm"
|
||||
|
||||
REQUIRED_COMPONENT = "siren"
|
||||
QUERY_GETTER_NAME = "getSirenStatus"
|
@ -10,14 +10,14 @@ from ...credentials import Credentials
|
||||
from ...device_type import DeviceType
|
||||
from ...feature import Feature
|
||||
from ...json import loads as json_loads
|
||||
from ..smartcameramodule import SmartCameraModule
|
||||
from ..smartcammodule import SmartCamModule
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
LOCAL_STREAMING_PORT = 554
|
||||
|
||||
|
||||
class Camera(SmartCameraModule):
|
||||
class Camera(SmartCamModule):
|
||||
"""Implementation of device module."""
|
||||
|
||||
QUERY_GETTER_NAME = "getLensMaskConfig"
|
@ -1,10 +1,10 @@
|
||||
"""Module for child devices."""
|
||||
|
||||
from ...device_type import DeviceType
|
||||
from ..smartcameramodule import SmartCameraModule
|
||||
from ..smartcammodule import SmartCamModule
|
||||
|
||||
|
||||
class ChildDevice(SmartCameraModule):
|
||||
class ChildDevice(SmartCamModule):
|
||||
"""Implementation for child devices."""
|
||||
|
||||
REQUIRED_COMPONENT = "childControl"
|
@ -3,10 +3,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from ...feature import Feature
|
||||
from ..smartcameramodule import SmartCameraModule
|
||||
from ..smartcammodule import SmartCamModule
|
||||
|
||||
|
||||
class DeviceModule(SmartCameraModule):
|
||||
class DeviceModule(SmartCamModule):
|
||||
"""Implementation of device module."""
|
||||
|
||||
NAME = "devicemodule"
|
@ -3,10 +3,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from ...interfaces.led import Led as LedInterface
|
||||
from ..smartcameramodule import SmartCameraModule
|
||||
from ..smartcammodule import SmartCamModule
|
||||
|
||||
|
||||
class Led(SmartCameraModule, LedInterface):
|
||||
class Led(SmartCamModule, LedInterface):
|
||||
"""Implementation of led controls."""
|
||||
|
||||
REQUIRED_COMPONENT = "led"
|
@ -3,13 +3,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from ...feature import Feature
|
||||
from ..smartcameramodule import SmartCameraModule
|
||||
from ..smartcammodule import SmartCamModule
|
||||
|
||||
DEFAULT_PAN_STEP = 30
|
||||
DEFAULT_TILT_STEP = 10
|
||||
|
||||
|
||||
class PanTilt(SmartCameraModule):
|
||||
class PanTilt(SmartCamModule):
|
||||
"""Implementation of device_local_time."""
|
||||
|
||||
REQUIRED_COMPONENT = "ptz"
|
@ -9,10 +9,10 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
from ...cachedzoneinfo import CachedZoneInfo
|
||||
from ...feature import Feature
|
||||
from ...interfaces import Time as TimeInterface
|
||||
from ..smartcameramodule import SmartCameraModule
|
||||
from ..smartcammodule import SmartCamModule
|
||||
|
||||
|
||||
class Time(SmartCameraModule, TimeInterface):
|
||||
class Time(SmartCamModule, TimeInterface):
|
||||
"""Implementation of device_local_time."""
|
||||
|
||||
QUERY_GETTER_NAME = "getTimezone"
|
@ -1,4 +1,4 @@
|
||||
"""Module for smartcamera."""
|
||||
"""Module for SmartCamDevice."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@ -8,15 +8,15 @@ from typing import Any
|
||||
from ..device import _DeviceInfo
|
||||
from ..device_type import DeviceType
|
||||
from ..module import Module
|
||||
from ..protocols.smartcameraprotocol import _ChildCameraProtocolWrapper
|
||||
from ..protocols.smartcamprotocol import _ChildCameraProtocolWrapper
|
||||
from ..smart import SmartChildDevice, SmartDevice
|
||||
from .modules import ChildDevice, DeviceModule
|
||||
from .smartcameramodule import SmartCameraModule
|
||||
from .smartcammodule import SmartCamModule
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SmartCamera(SmartDevice):
|
||||
class SmartCamDevice(SmartDevice):
|
||||
"""Class for smart cameras."""
|
||||
|
||||
# Modules that are called as part of the init procedure on first update
|
||||
@ -41,7 +41,7 @@ class SmartCamera(SmartDevice):
|
||||
basic_info = info["getDeviceInfo"]["device_info"]["basic_info"]
|
||||
short_name = basic_info["device_model"]
|
||||
long_name = discovery_info["device_model"] if discovery_info else short_name
|
||||
device_type = SmartCamera._get_device_type_from_sysinfo(basic_info)
|
||||
device_type = SmartCamDevice._get_device_type_from_sysinfo(basic_info)
|
||||
fw_version_full = basic_info["sw_version"]
|
||||
firmware_version, firmware_build = fw_version_full.split(" ", maxsplit=1)
|
||||
return _DeviceInfo(
|
||||
@ -73,7 +73,7 @@ class SmartCamera(SmartDevice):
|
||||
async def _initialize_smart_child(
|
||||
self, info: dict, child_components: dict
|
||||
) -> SmartDevice:
|
||||
"""Initialize a smart child device attached to a smartcamera."""
|
||||
"""Initialize a smart child device attached to a smartcam device."""
|
||||
child_id = info["device_id"]
|
||||
child_protocol = _ChildCameraProtocolWrapper(child_id, self.protocol)
|
||||
try:
|
||||
@ -122,7 +122,7 @@ class SmartCamera(SmartDevice):
|
||||
|
||||
async def _initialize_modules(self) -> None:
|
||||
"""Initialize modules based on component negotiation response."""
|
||||
for mod in SmartCameraModule.REGISTERED_MODULES.values():
|
||||
for mod in SmartCamModule.REGISTERED_MODULES.values():
|
||||
if (
|
||||
mod.REQUIRED_COMPONENT
|
||||
and mod.REQUIRED_COMPONENT not in self._components
|
@ -11,15 +11,15 @@ from ..smart.smartmodule import SmartModule
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from . import modules
|
||||
from .smartcamera import SmartCamera
|
||||
from .smartcamdevice import SmartCamDevice
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SmartCameraModule(SmartModule):
|
||||
"""Base class for SMARTCAMERA modules."""
|
||||
class SmartCamModule(SmartModule):
|
||||
"""Base class for SMARTCAM modules."""
|
||||
|
||||
SmartCameraAlarm: Final[ModuleName[modules.Alarm]] = ModuleName("SmartCameraAlarm")
|
||||
SmartCamAlarm: Final[ModuleName[modules.Alarm]] = ModuleName("SmartCamAlarm")
|
||||
|
||||
#: Query to execute during the main update cycle
|
||||
QUERY_GETTER_NAME: str
|
||||
@ -30,7 +30,7 @@ class SmartCameraModule(SmartModule):
|
||||
|
||||
REGISTERED_MODULES = {}
|
||||
|
||||
_device: SmartCamera
|
||||
_device: SmartCamDevice
|
||||
|
||||
def query(self) -> dict:
|
||||
"""Query to execute during the update cycle.
|
@ -1,5 +0,0 @@
|
||||
"""Package for supporting tapo-branded cameras."""
|
||||
|
||||
from .smartcamera import SmartCamera
|
||||
|
||||
__all__ = ["SmartCamera"]
|
@ -13,11 +13,11 @@ from kasa import (
|
||||
)
|
||||
from kasa.iot import IotBulb, IotDimmer, IotLightStrip, IotPlug, IotStrip, IotWallSwitch
|
||||
from kasa.smart import SmartDevice
|
||||
from kasa.smartcamera.smartcamera import SmartCamera
|
||||
from kasa.smartcam import SmartCamDevice
|
||||
|
||||
from .fakeprotocol_iot import FakeIotProtocol
|
||||
from .fakeprotocol_smart import FakeSmartProtocol
|
||||
from .fakeprotocol_smartcamera import FakeSmartCameraProtocol
|
||||
from .fakeprotocol_smartcam import FakeSmartCamProtocol
|
||||
from .fixtureinfo import (
|
||||
FIXTURE_DATA,
|
||||
ComponentFilter,
|
||||
@ -317,16 +317,16 @@ device_smart = parametrize(
|
||||
device_iot = parametrize(
|
||||
"devices iot", model_filter=ALL_DEVICES_IOT, protocol_filter={"IOT"}
|
||||
)
|
||||
device_smartcamera = parametrize("devices smartcamera", protocol_filter={"SMARTCAMERA"})
|
||||
camera_smartcamera = parametrize(
|
||||
"camera smartcamera",
|
||||
device_smartcam = parametrize("devices smartcam", protocol_filter={"SMARTCAM"})
|
||||
camera_smartcam = parametrize(
|
||||
"camera smartcam",
|
||||
device_type_filter=[DeviceType.Camera],
|
||||
protocol_filter={"SMARTCAMERA"},
|
||||
protocol_filter={"SMARTCAM"},
|
||||
)
|
||||
hub_smartcamera = parametrize(
|
||||
"hub smartcamera",
|
||||
hub_smartcam = parametrize(
|
||||
"hub smartcam",
|
||||
device_type_filter=[DeviceType.Hub],
|
||||
protocol_filter={"SMARTCAMERA"},
|
||||
protocol_filter={"SMARTCAM"},
|
||||
)
|
||||
|
||||
|
||||
@ -344,8 +344,8 @@ def check_categories():
|
||||
+ hubs_smart.args[1]
|
||||
+ sensors_smart.args[1]
|
||||
+ thermostats_smart.args[1]
|
||||
+ camera_smartcamera.args[1]
|
||||
+ hub_smartcamera.args[1]
|
||||
+ camera_smartcam.args[1]
|
||||
+ hub_smartcam.args[1]
|
||||
)
|
||||
diffs: set[FixtureInfo] = set(FIXTURE_DATA) - set(categorized_fixtures)
|
||||
if diffs:
|
||||
@ -363,8 +363,8 @@ check_categories()
|
||||
def device_for_fixture_name(model, protocol):
|
||||
if protocol in {"SMART", "SMART.CHILD"}:
|
||||
return SmartDevice
|
||||
elif protocol == "SMARTCAMERA":
|
||||
return SmartCamera
|
||||
elif protocol == "SMARTCAM":
|
||||
return SmartCamDevice
|
||||
else:
|
||||
for d in STRIPS_IOT:
|
||||
if d in model:
|
||||
@ -420,8 +420,8 @@ async def get_device_for_fixture(
|
||||
d.protocol = FakeSmartProtocol(
|
||||
fixture_data.data, fixture_data.name, verbatim=verbatim
|
||||
)
|
||||
elif fixture_data.protocol == "SMARTCAMERA":
|
||||
d.protocol = FakeSmartCameraProtocol(
|
||||
elif fixture_data.protocol == "SMARTCAM":
|
||||
d.protocol = FakeSmartCamProtocol(
|
||||
fixture_data.data, fixture_data.name, verbatim=verbatim
|
||||
)
|
||||
else:
|
||||
@ -460,8 +460,8 @@ def get_fixture_info(fixture, protocol):
|
||||
def get_nearest_fixture_to_ip(dev):
|
||||
if isinstance(dev, SmartDevice):
|
||||
protocol_fixtures = filter_fixtures("", protocol_filter={"SMART"})
|
||||
elif isinstance(dev, SmartCamera):
|
||||
protocol_fixtures = filter_fixtures("", protocol_filter={"SMARTCAMERA"})
|
||||
elif isinstance(dev, SmartCamDevice):
|
||||
protocol_fixtures = filter_fixtures("", protocol_filter={"SMARTCAM"})
|
||||
else:
|
||||
protocol_fixtures = filter_fixtures("", protocol_filter={"IOT"})
|
||||
assert protocol_fixtures, "Unknown device type"
|
||||
|
@ -11,7 +11,7 @@ from kasa.transports.xortransport import XorEncryption
|
||||
|
||||
from .fakeprotocol_iot import FakeIotProtocol
|
||||
from .fakeprotocol_smart import FakeSmartProtocol, FakeSmartTransport
|
||||
from .fakeprotocol_smartcamera import FakeSmartCameraProtocol
|
||||
from .fakeprotocol_smartcam import FakeSmartCamProtocol
|
||||
from .fixtureinfo import FixtureInfo, filter_fixtures, idgenerator
|
||||
|
||||
DISCOVERY_MOCK_IP = "127.0.0.123"
|
||||
@ -194,8 +194,8 @@ def patch_discovery(fixture_infos: dict[str, FixtureInfo], mocker):
|
||||
protos = {
|
||||
ip: FakeSmartProtocol(fixture_info.data, fixture_info.name)
|
||||
if fixture_info.protocol in {"SMART", "SMART.CHILD"}
|
||||
else FakeSmartCameraProtocol(fixture_info.data, fixture_info.name)
|
||||
if fixture_info.protocol in {"SMARTCAMERA", "SMARTCAMERA.CHILD"}
|
||||
else FakeSmartCamProtocol(fixture_info.data, fixture_info.name)
|
||||
if fixture_info.protocol in {"SMARTCAM", "SMARTCAM.CHILD"}
|
||||
else FakeIotProtocol(fixture_info.data, fixture_info.name)
|
||||
for ip, fixture_info in fixture_infos.items()
|
||||
}
|
||||
@ -221,8 +221,8 @@ def patch_discovery(fixture_infos: dict[str, FixtureInfo], mocker):
|
||||
protos[host] = (
|
||||
FakeSmartProtocol(fixture_info.data, fixture_info.name)
|
||||
if fixture_info.protocol in {"SMART", "SMART.CHILD"}
|
||||
else FakeSmartCameraProtocol(fixture_info.data, fixture_info.name)
|
||||
if fixture_info.protocol in {"SMARTCAMERA", "SMARTCAMERA.CHILD"}
|
||||
else FakeSmartCamProtocol(fixture_info.data, fixture_info.name)
|
||||
if fixture_info.protocol in {"SMARTCAM", "SMARTCAM.CHILD"}
|
||||
else FakeIotProtocol(fixture_info.data, fixture_info.name)
|
||||
)
|
||||
port = (
|
||||
|
@ -5,16 +5,16 @@ from json import loads as json_loads
|
||||
from typing import Any
|
||||
|
||||
from kasa import Credentials, DeviceConfig, SmartProtocol
|
||||
from kasa.protocols.smartcameraprotocol import SmartCameraProtocol
|
||||
from kasa.protocols.smartcamprotocol import SmartCamProtocol
|
||||
from kasa.transports.basetransport import BaseTransport
|
||||
|
||||
from .fakeprotocol_smart import FakeSmartTransport
|
||||
|
||||
|
||||
class FakeSmartCameraProtocol(SmartCameraProtocol):
|
||||
class FakeSmartCamProtocol(SmartCamProtocol):
|
||||
def __init__(self, info, fixture_name, *, is_child=False, verbatim=False):
|
||||
super().__init__(
|
||||
transport=FakeSmartCameraTransport(
|
||||
transport=FakeSmartCamTransport(
|
||||
info, fixture_name, is_child=is_child, verbatim=verbatim
|
||||
),
|
||||
)
|
||||
@ -25,7 +25,7 @@ class FakeSmartCameraProtocol(SmartCameraProtocol):
|
||||
return resp_dict
|
||||
|
||||
|
||||
class FakeSmartCameraTransport(BaseTransport):
|
||||
class FakeSmartCamTransport(BaseTransport):
|
||||
def __init__(
|
||||
self,
|
||||
info,
|
@ -13,7 +13,7 @@ import pytest
|
||||
from kasa.device_type import DeviceType
|
||||
from kasa.iot import IotDevice
|
||||
from kasa.smart.smartdevice import SmartDevice
|
||||
from kasa.smartcamera.smartcamera import SmartCamera
|
||||
from kasa.smartcam import SmartCamDevice
|
||||
|
||||
|
||||
class FixtureInfo(NamedTuple):
|
||||
@ -53,10 +53,10 @@ SUPPORTED_SMART_CHILD_DEVICES = [
|
||||
)
|
||||
]
|
||||
|
||||
SUPPORTED_SMARTCAMERA_DEVICES = [
|
||||
(device, "SMARTCAMERA")
|
||||
SUPPORTED_SMARTCAM_DEVICES = [
|
||||
(device, "SMARTCAM")
|
||||
for device in glob.glob(
|
||||
os.path.dirname(os.path.abspath(__file__)) + "/fixtures/smartcamera/*.json"
|
||||
os.path.dirname(os.path.abspath(__file__)) + "/fixtures/smartcam/*.json"
|
||||
)
|
||||
]
|
||||
|
||||
@ -64,7 +64,7 @@ SUPPORTED_DEVICES = (
|
||||
SUPPORTED_IOT_DEVICES
|
||||
+ SUPPORTED_SMART_DEVICES
|
||||
+ SUPPORTED_SMART_CHILD_DEVICES
|
||||
+ SUPPORTED_SMARTCAMERA_DEVICES
|
||||
+ SUPPORTED_SMARTCAM_DEVICES
|
||||
)
|
||||
|
||||
|
||||
@ -179,14 +179,14 @@ def filter_fixtures(
|
||||
IotDevice._get_device_type_from_sys_info(fixture_data.data)
|
||||
in device_type
|
||||
)
|
||||
elif fixture_data.protocol == "SMARTCAMERA":
|
||||
elif fixture_data.protocol == "SMARTCAM":
|
||||
info = fixture_data.data["getDeviceInfo"]["device_info"]["basic_info"]
|
||||
return SmartCamera._get_device_type_from_sysinfo(info) in device_type
|
||||
return SmartCamDevice._get_device_type_from_sysinfo(info) in device_type
|
||||
return False
|
||||
|
||||
filtered = []
|
||||
if protocol_filter is None:
|
||||
protocol_filter = {"IOT", "SMART", "SMARTCAMERA"}
|
||||
protocol_filter = {"IOT", "SMART", "SMARTCAM"}
|
||||
for fixture_data in fixture_list:
|
||||
if data_root_filter and data_root_filter not in fixture_data.data:
|
||||
continue
|
||||
|
@ -5,21 +5,21 @@ from __future__ import annotations
|
||||
import pytest
|
||||
|
||||
from kasa import Device
|
||||
from kasa.smartcamera.modules.alarm import (
|
||||
from kasa.smartcam.modules.alarm import (
|
||||
DURATION_MAX,
|
||||
DURATION_MIN,
|
||||
VOLUME_MAX,
|
||||
VOLUME_MIN,
|
||||
)
|
||||
from kasa.smartcamera.smartcameramodule import SmartCameraModule
|
||||
from kasa.smartcam.smartcammodule import SmartCamModule
|
||||
|
||||
from ...conftest import hub_smartcamera
|
||||
from ...conftest import hub_smartcam
|
||||
|
||||
|
||||
@hub_smartcamera
|
||||
@hub_smartcam
|
||||
async def test_alarm(dev: Device):
|
||||
"""Test device alarm."""
|
||||
alarm = dev.modules.get(SmartCameraModule.SmartCameraAlarm)
|
||||
alarm = dev.modules.get(SmartCamModule.SmartCamAlarm)
|
||||
assert alarm
|
||||
|
||||
original_duration = alarm.alarm_duration
|
||||
@ -70,10 +70,10 @@ async def test_alarm(dev: Device):
|
||||
await dev.update()
|
||||
|
||||
|
||||
@hub_smartcamera
|
||||
@hub_smartcam
|
||||
async def test_alarm_invalid_setters(dev: Device):
|
||||
"""Test device alarm invalid setter values."""
|
||||
alarm = dev.modules.get(SmartCameraModule.SmartCameraAlarm)
|
||||
alarm = dev.modules.get(SmartCamModule.SmartCamAlarm)
|
||||
assert alarm
|
||||
|
||||
# test set sound invalid
|
||||
@ -92,10 +92,10 @@ async def test_alarm_invalid_setters(dev: Device):
|
||||
await alarm.set_alarm_duration(-3)
|
||||
|
||||
|
||||
@hub_smartcamera
|
||||
@hub_smartcam
|
||||
async def test_alarm_features(dev: Device):
|
||||
"""Test device alarm features."""
|
||||
alarm = dev.modules.get(SmartCameraModule.SmartCameraAlarm)
|
||||
alarm = dev.modules.get(SmartCamModule.SmartCamAlarm)
|
||||
assert alarm
|
||||
|
||||
original_duration = alarm.alarm_duration
|
||||
|
@ -12,10 +12,10 @@ from freezegun.api import FrozenDateTimeFactory
|
||||
|
||||
from kasa import Credentials, Device, DeviceType, Module
|
||||
|
||||
from ..conftest import camera_smartcamera, device_smartcamera, hub_smartcamera
|
||||
from ..conftest import camera_smartcam, device_smartcam, hub_smartcam
|
||||
|
||||
|
||||
@device_smartcamera
|
||||
@device_smartcam
|
||||
async def test_state(dev: Device):
|
||||
if dev.device_type is DeviceType.Hub:
|
||||
pytest.skip("Hubs cannot be switched on and off")
|
||||
@ -26,7 +26,7 @@ async def test_state(dev: Device):
|
||||
assert dev.is_on is not state
|
||||
|
||||
|
||||
@camera_smartcamera
|
||||
@camera_smartcam
|
||||
async def test_stream_rtsp_url(dev: Device):
|
||||
camera_module = dev.modules.get(Module.Camera)
|
||||
assert camera_module
|
||||
@ -85,7 +85,7 @@ async def test_stream_rtsp_url(dev: Device):
|
||||
assert url is None
|
||||
|
||||
|
||||
@device_smartcamera
|
||||
@device_smartcam
|
||||
async def test_alias(dev):
|
||||
test_alias = "TEST1234"
|
||||
original = dev.alias
|
||||
@ -100,7 +100,7 @@ async def test_alias(dev):
|
||||
assert dev.alias == original
|
||||
|
||||
|
||||
@hub_smartcamera
|
||||
@hub_smartcam
|
||||
async def test_hub(dev):
|
||||
assert dev.children
|
||||
for child in dev.children:
|
||||
@ -112,7 +112,7 @@ async def test_hub(dev):
|
||||
assert child.time
|
||||
|
||||
|
||||
@device_smartcamera
|
||||
@device_smartcam
|
||||
async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory):
|
||||
"""Test a child device gets the time from it's parent module."""
|
||||
fallback_time = datetime.now(UTC).astimezone().replace(microsecond=0)
|
||||
|
@ -45,7 +45,7 @@ from kasa.cli.wifi import wifi
|
||||
from kasa.discover import Discover, DiscoveryResult
|
||||
from kasa.iot import IotDevice
|
||||
from kasa.smart import SmartDevice
|
||||
from kasa.smartcamera import SmartCamera
|
||||
from kasa.smartcam import SmartCamDevice
|
||||
|
||||
from .conftest import (
|
||||
device_smart,
|
||||
@ -181,7 +181,7 @@ async def test_state(dev, turn_on, runner):
|
||||
|
||||
@turn_on
|
||||
async def test_toggle(dev, turn_on, runner):
|
||||
if isinstance(dev, SmartCamera) and dev.device_type == DeviceType.Hub:
|
||||
if isinstance(dev, SmartCamDevice) and dev.device_type == DeviceType.Hub:
|
||||
pytest.skip(reason="Hub cannot toggle state")
|
||||
|
||||
await handle_turn_on(dev, turn_on)
|
||||
@ -214,7 +214,7 @@ async def test_raw_command(dev, mocker, runner):
|
||||
update = mocker.patch.object(dev, "update")
|
||||
from kasa.smart import SmartDevice
|
||||
|
||||
if isinstance(dev, SmartCamera):
|
||||
if isinstance(dev, SmartCamDevice):
|
||||
params = ["na", "getDeviceInfo"]
|
||||
elif isinstance(dev, SmartDevice):
|
||||
params = ["na", "get_device_info"]
|
||||
@ -917,7 +917,7 @@ async def test_type_param(device_type, mocker, runner):
|
||||
|
||||
mocker.patch("kasa.cli.device.state", new=_state)
|
||||
if device_type == "camera":
|
||||
expected_type = SmartCamera
|
||||
expected_type = SmartCamDevice
|
||||
elif device_type == "smart":
|
||||
expected_type = SmartDevice
|
||||
else:
|
||||
|
@ -30,7 +30,7 @@ from kasa.iot.iottimezone import (
|
||||
)
|
||||
from kasa.iot.modules import IotLightPreset
|
||||
from kasa.smart import SmartChildDevice, SmartDevice
|
||||
from kasa.smartcamera import SmartCamera
|
||||
from kasa.smartcam import SmartCamDevice
|
||||
|
||||
|
||||
def _get_subclasses(of_class):
|
||||
@ -115,7 +115,7 @@ async def test_device_class_repr(device_class_name_obj):
|
||||
IotLightStrip: DeviceType.LightStrip,
|
||||
SmartChildDevice: DeviceType.Unknown,
|
||||
SmartDevice: DeviceType.Unknown,
|
||||
SmartCamera: DeviceType.Camera,
|
||||
SmartCamDevice: DeviceType.Camera,
|
||||
}
|
||||
type_ = CLASS_TO_DEFAULT_TYPE[klass]
|
||||
child_repr = "<DeviceType.Unknown(child) of <DeviceType.Unknown at 127.0.0.2 - update() needed>>"
|
||||
|
@ -20,7 +20,7 @@ from kasa import (
|
||||
from kasa.device_factory import (
|
||||
Device,
|
||||
IotDevice,
|
||||
SmartCamera,
|
||||
SmartCamDevice,
|
||||
SmartDevice,
|
||||
connect,
|
||||
get_device_class_from_family,
|
||||
@ -178,8 +178,8 @@ async def test_connect_http_client(discovery_mock, mocker):
|
||||
|
||||
async def test_device_types(dev: Device):
|
||||
await dev.update()
|
||||
if isinstance(dev, SmartCamera):
|
||||
res = SmartCamera._get_device_type_from_sysinfo(dev.sys_info)
|
||||
if isinstance(dev, SmartCamDevice):
|
||||
res = SmartCamDevice._get_device_type_from_sysinfo(dev.sys_info)
|
||||
elif isinstance(dev, SmartDevice):
|
||||
assert dev._discovery_info
|
||||
device_type = cast(str, dev._discovery_info["device_type"])
|
||||
|
@ -6,7 +6,7 @@ from devtools.dump_devinfo import get_legacy_fixture, get_smart_fixtures
|
||||
from kasa.iot import IotDevice
|
||||
from kasa.protocols import IotProtocol
|
||||
from kasa.smart import SmartDevice
|
||||
from kasa.smartcamera import SmartCamera
|
||||
from kasa.smartcam import SmartCamDevice
|
||||
|
||||
from .conftest import (
|
||||
FixtureInfo,
|
||||
@ -17,8 +17,8 @@ from .conftest import (
|
||||
smart_fixtures = parametrize(
|
||||
"smart fixtures", protocol_filter={"SMART"}, fixture_name="fixture_info"
|
||||
)
|
||||
smartcamera_fixtures = parametrize(
|
||||
"smartcamera fixtures", protocol_filter={"SMARTCAMERA"}, fixture_name="fixture_info"
|
||||
smartcam_fixtures = parametrize(
|
||||
"smartcam fixtures", protocol_filter={"SMARTCAM"}, fixture_name="fixture_info"
|
||||
)
|
||||
iot_fixtures = parametrize(
|
||||
"iot fixtures", protocol_filter={"IOT"}, fixture_name="fixture_info"
|
||||
@ -27,8 +27,8 @@ iot_fixtures = parametrize(
|
||||
|
||||
async def test_fixture_names(fixture_info: FixtureInfo):
|
||||
"""Test that device info gets the right fixture names."""
|
||||
if fixture_info.protocol in {"SMARTCAMERA"}:
|
||||
device_info = SmartCamera._get_device_info(
|
||||
if fixture_info.protocol in {"SMARTCAM"}:
|
||||
device_info = SmartCamDevice._get_device_info(
|
||||
fixture_info.data, fixture_info.data.get("discovery_result")
|
||||
)
|
||||
elif fixture_info.protocol in {"SMART"}:
|
||||
@ -62,11 +62,11 @@ async def test_smart_fixtures(fixture_info: FixtureInfo):
|
||||
assert fixture_info.data == fixture_result.data
|
||||
|
||||
|
||||
@smartcamera_fixtures
|
||||
async def test_smartcamera_fixtures(fixture_info: FixtureInfo):
|
||||
"""Test that smartcamera fixtures are created the same."""
|
||||
@smartcam_fixtures
|
||||
async def test_smartcam_fixtures(fixture_info: FixtureInfo):
|
||||
"""Test that smartcam fixtures are created the same."""
|
||||
dev = await get_device_for_fixture(fixture_info, verbatim=True)
|
||||
assert isinstance(dev, SmartCamera)
|
||||
assert isinstance(dev, SmartCamDevice)
|
||||
if dev.children:
|
||||
pytest.skip("Test not currently implemented for devices with children.")
|
||||
fixtures = await get_smart_fixtures(
|
||||
|
Loading…
Reference in New Issue
Block a user