mirror of
https://github.com/python-kasa/python-kasa.git
synced 2025-02-02 10:07:03 +00:00
extract shared types (exceptions, enums), add module level doc, rename exception to be generic
This commit is contained in:
parent
93d24281c6
commit
7457ad48f8
@ -1,5 +1,20 @@
|
|||||||
|
"""
|
||||||
|
This module provides a way to interface with TP-Link's smart home devices,
|
||||||
|
such as smart plugs (HS1xx), wall switches (HS2xx), and light bulbs (LB1xx).
|
||||||
|
|
||||||
|
All common, shared functionalities are available through `SmartDevice` class::
|
||||||
|
|
||||||
|
x = SmartDevice("192.168.1.1")
|
||||||
|
print(x.sys_info)
|
||||||
|
|
||||||
|
For device type specific actions `SmartBulb` or `SmartPlug` must be used instead.
|
||||||
|
|
||||||
|
Module-specific errors are raised as `SmartDeviceException` and are expected
|
||||||
|
to be handled by the user of the library.
|
||||||
|
"""
|
||||||
# flake8: noqa
|
# flake8: noqa
|
||||||
from .smartplug import SmartPlug
|
from .smartplug import SmartPlug
|
||||||
from .pyHS100 import SmartPlugException, SmartDevice
|
from .pyHS100 import SmartDevice
|
||||||
|
from .types import SmartDeviceException
|
||||||
from .smartbulb import SmartBulb
|
from .smartbulb import SmartBulb
|
||||||
from .protocol import TPLinkSmartHomeProtocol
|
from .protocol import TPLinkSmartHomeProtocol
|
||||||
|
@ -16,27 +16,13 @@ http://www.apache.org/licenses/LICENSE-2.0
|
|||||||
import datetime
|
import datetime
|
||||||
import logging
|
import logging
|
||||||
import socket
|
import socket
|
||||||
import enum
|
|
||||||
|
|
||||||
|
from .types import SmartDeviceException
|
||||||
from .protocol import TPLinkSmartHomeProtocol
|
from .protocol import TPLinkSmartHomeProtocol
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class SmartPlugException(Exception):
|
|
||||||
"""
|
|
||||||
SmartPlugException gets raised for errors reported by the plug.
|
|
||||||
"""
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class DeviceType(enum.Enum):
|
|
||||||
Unknown = -1,
|
|
||||||
Plug = 0,
|
|
||||||
Switch = 1
|
|
||||||
Bulb = 2
|
|
||||||
|
|
||||||
|
|
||||||
class SmartDevice(object):
|
class SmartDevice(object):
|
||||||
# possible device features
|
# possible device features
|
||||||
FEATURE_ENERGY_METER = 'ENE'
|
FEATURE_ENERGY_METER = 'ENE'
|
||||||
@ -76,16 +62,16 @@ class SmartDevice(object):
|
|||||||
request={target: {cmd: arg}}
|
request={target: {cmd: arg}}
|
||||||
)
|
)
|
||||||
except Exception as ex:
|
except Exception as ex:
|
||||||
raise SmartPlugException('Communication error') from ex
|
raise SmartDeviceException('Communication error') from ex
|
||||||
|
|
||||||
if target not in response:
|
if target not in response:
|
||||||
raise SmartPlugException("No required {} in response: {}"
|
raise SmartDeviceException("No required {} in response: {}"
|
||||||
.format(target, response))
|
.format(target, response))
|
||||||
|
|
||||||
result = response[target]
|
result = response[target]
|
||||||
if "err_code" in result and result["err_code"] != 0:
|
if "err_code" in result and result["err_code"] != 0:
|
||||||
raise SmartPlugException("Error on {}.{}: {}"
|
raise SmartDeviceException("Error on {}.{}: {}"
|
||||||
.format(target, cmd, result))
|
.format(target, cmd, result))
|
||||||
|
|
||||||
result = result[cmd]
|
result = result[cmd]
|
||||||
del result["err_code"]
|
del result["err_code"]
|
||||||
|
@ -1,5 +1,5 @@
|
|||||||
from ..protocol import TPLinkSmartHomeProtocol
|
from ..protocol import TPLinkSmartHomeProtocol
|
||||||
from .. import SmartPlugException
|
from .. import SmartDeviceException
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
|
||||||
@ -250,7 +250,7 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
|
|||||||
|
|
||||||
def query(self, host, request, port=9999):
|
def query(self, host, request, port=9999):
|
||||||
if self.invalid:
|
if self.invalid:
|
||||||
raise SmartPlugException("Invalid connection, can't query!")
|
raise SmartDeviceException("Invalid connection, can't query!")
|
||||||
|
|
||||||
proto = self.proto
|
proto = self.proto
|
||||||
|
|
||||||
|
@ -2,7 +2,7 @@ from unittest import TestCase, skip, skipIf
|
|||||||
from voluptuous import Schema, Invalid, All, Range
|
from voluptuous import Schema, Invalid, All, Range
|
||||||
from functools import partial
|
from functools import partial
|
||||||
|
|
||||||
from .. import SmartBulb, SmartPlugException
|
from .. import SmartBulb, SmartDeviceException
|
||||||
from .fakes import FakeTransportProtocol, sysinfo_lb130
|
from .fakes import FakeTransportProtocol, sysinfo_lb130
|
||||||
|
|
||||||
BULB_IP = '192.168.250.186'
|
BULB_IP = '192.168.250.186'
|
||||||
@ -93,11 +93,11 @@ class TestSmartBulb(TestCase):
|
|||||||
bulb = SmartBulb('127.0.0.1',
|
bulb = SmartBulb('127.0.0.1',
|
||||||
protocol=FakeTransportProtocol(sysinfo_lb130,
|
protocol=FakeTransportProtocol(sysinfo_lb130,
|
||||||
invalid=True))
|
invalid=True))
|
||||||
with self.assertRaises(SmartPlugException):
|
with self.assertRaises(SmartDeviceException):
|
||||||
bulb.sys_info['model']
|
bulb.sys_info['model']
|
||||||
|
|
||||||
def test_query_helper(self):
|
def test_query_helper(self):
|
||||||
with self.assertRaises(SmartPlugException):
|
with self.assertRaises(SmartDeviceException):
|
||||||
self.bulb._query_helper("test", "testcmd", {})
|
self.bulb._query_helper("test", "testcmd", {})
|
||||||
# TODO check for unwrapping?
|
# TODO check for unwrapping?
|
||||||
|
|
||||||
|
@ -4,7 +4,7 @@ from functools import partial
|
|||||||
import datetime
|
import datetime
|
||||||
import re
|
import re
|
||||||
|
|
||||||
from .. import SmartPlug, SmartPlugException
|
from .. import SmartPlug, SmartDeviceException
|
||||||
from .fakes import FakeTransportProtocol, sysinfo_hs110, sysinfo_hs105
|
from .fakes import FakeTransportProtocol, sysinfo_hs110, sysinfo_hs105
|
||||||
|
|
||||||
PLUG_IP = '192.168.250.186'
|
PLUG_IP = '192.168.250.186'
|
||||||
@ -87,11 +87,11 @@ class TestSmartPlug(TestCase):
|
|||||||
plug = SmartPlug('127.0.0.1',
|
plug = SmartPlug('127.0.0.1',
|
||||||
protocol=FakeTransportProtocol(sysinfo_hs110,
|
protocol=FakeTransportProtocol(sysinfo_hs110,
|
||||||
invalid=True))
|
invalid=True))
|
||||||
with self.assertRaises(SmartPlugException):
|
with self.assertRaises(SmartDeviceException):
|
||||||
plug.sys_info['model']
|
plug.sys_info['model']
|
||||||
|
|
||||||
def test_query_helper(self):
|
def test_query_helper(self):
|
||||||
with self.assertRaises(SmartPlugException):
|
with self.assertRaises(SmartDeviceException):
|
||||||
self.plug._query_helper("test", "testcmd", {})
|
self.plug._query_helper("test", "testcmd", {})
|
||||||
# TODO check for unwrapping?
|
# TODO check for unwrapping?
|
||||||
|
|
||||||
|
15
pyHS100/types.py
Normal file
15
pyHS100/types.py
Normal file
@ -0,0 +1,15 @@
|
|||||||
|
import enum
|
||||||
|
|
||||||
|
|
||||||
|
class SmartDeviceException(Exception):
|
||||||
|
"""
|
||||||
|
SmartPlugException gets raised for errors reported by the plug.
|
||||||
|
"""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class DeviceType(enum.Enum):
|
||||||
|
Unknown = -1,
|
||||||
|
Plug = 0,
|
||||||
|
Switch = 1
|
||||||
|
Bulb = 2
|
Loading…
Reference in New Issue
Block a user