[WIP] Kasa Cam support #537
Conversation
f9179ab to
b2c78a1
Compare
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #537 +/- ##
==========================================
+ Coverage 80.36% 81.35% +0.99%
==========================================
Files 27 30 +3
Lines 2093 2274 +181
Branches 639 671 +32
==========================================
+ Hits 1682 1850 +168
- Misses 366 376 +10
- Partials 45 48 +3 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Thanks for the PR, @bstrdsmkr! I added some initial review comments inline.
| @@ -379,7 +386,7 @@ def update_from_discover_info(self, info: Dict[str, Any]) -> None: | |||
|
|
|||
| def _set_sys_info(self, sys_info: Dict[str, Any]) -> None: | |||
| """Set sys_info.""" | |||
| self._sys_info = sys_info | |||
| self._sys_info = sys_info if "model" in sys_info else sys_info["system"] | |||
There was a problem hiding this comment.
Add a quick command why this is done.
There was a problem hiding this comment.
I saw you did check earlier if the key system is inside the sysinfo, this probably should do the same for consistency?
There was a problem hiding this comment.
This was buried, so just a friendly ping :-)
2ea2141 to
3a1db0c
Compare
| @@ -379,7 +386,7 @@ def update_from_discover_info(self, info: Dict[str, Any]) -> None: | |||
|
|
|||
| def _set_sys_info(self, sys_info: Dict[str, Any]) -> None: | |||
| """Set sys_info.""" | |||
| self._sys_info = sys_info | |||
| self._sys_info = sys_info if "model" in sys_info else sys_info["system"] | |||
There was a problem hiding this comment.
I saw you did check earlier if the key system is inside the sysinfo, this probably should do the same for consistency?
7f83115 to
26ce51a
Compare
63f7206 to
ba63501
Compare
ba63501 to
e2d8f52
Compare
|
@rytilahti the ptz stuff is most of what I personally care about for this but I've pulled a List of known commands from the android app# smartlife.cam.ipcamera.audio
async def get_mic_config(self):
# , volume:int
pass
async def get_quickres_list(self):
# , files:list, max_files:int
return await self.call("smartlife.cam.ipcamera.audio", "get_quickres_list", {})
async def set_chime(self, duration: int):
pass
async def set_mic_config(self, volume: int):
pass
async def set_quickres_off(self):
pass
async def set_quickres_state(self, file_id: int, repeat: int):
pass
# smartlife.cam.ipcamera.battery
async def get_power_save(self):
pass
async def get_status(self):
# public Integer battery_capacity;
# public String battery_charging;
# public Boolean battery_installed;
# public String battery_model;
# public Integer battery_percent;
# public Integer battery_temperature;
# public String battery_vendor;
# public Integer battery_voltage;
# public Boolean low_battery;
# public String power_mode;
pass
async def set_power_save(self):
pass
# smartlife.cam.ipcamera.siren
async def get_config(self):
# public Integer duration;
# public Boolean enable;
# public Integer volume;
return await self.call("smartlife.cam.ipcamera.siren", "get_config", {})
async def get_state(self):
# public Integer time_left;
# public String value;
pass
async def set_config(self, duration: int, enable: bool, volume: int):
pass
async def set_state(self, state: str):
pass
async def set_state_with_duration(self, state: str, duration: int):
pass
# smartlife.cam.ipcamera.switch
async def get_is_enable(self):
pass
async def set_is_enable(self, enabled: bool):
pass
# smartlife.cam.ipcamera.cloud
async def get_info(self):
# public Boolean binded;
# public Boolean cld_connection;
# public String default_svr;
# public String fw_dl_page;
# public Integer fw_notify_type;
# public Integer illegal_type;
# public String sef_domain;
# public Integer stop_connect;
# public String tcsp_info;
# public Integer tcsp_status;
# public String username;
return await self.call("smartlife.cam.ipcamera.cloud", "get_info", {})
async def set_server_url(self):
# Deprecated
pass
async def set_n_server_url(self):
# public String default_svr;
# public String ipcserv_svr;
# public String sef_domain
pass
async def set_bind(self):
pass
async def set_unbind(self, username: str, password: str):
pass
# smartlife.cam.ipcamera.dateTime
async def get_status(self):
pass
async def get_time(self):
# public Long epoch_sec;
pass
async def get_time_zone(self):
# public String area;
# public String timezone;
pass
async def set_time(self, epoch_sec: int, mode: str):
pass
async def set_time_zone(self, area: str, timezone: str):
pass
# smartlife.cam.ipcamera.dayNight
async def get_force_lamp_state(self):
pass
async def get_force_lamp_time(self):
pass
async def get_lamp_status(self):
# public Integer forcetime;
# public String value;
pass
async def get_mode(self):
pass
async def get_night_vision_mode(self):
pass
async def set_force_lamp_state(self):
pass
async def set_force_lamp_time(self):
pass
async def set_mode(self):
pass
async def set_night_vision_mode(self):
pass
# smartlife.cam.ipcamera.debug
async def set_http_server_switch(self):
# public String av;
# public String httpd;
pass
# smartlife.cam.ipcamera.delivery
async def get_clip_audio_is_enable(self):
pass
async def get_local_clip_is_enable(self):
return await self.call(
"smartlife.cam.ipcamera.delivery", "get_local_clip_is_enable", {}
)
async def set_clip_audio_is_enable(self):
pass
async def set_local_clip_is_enable(self):
pass
# smartlife.cam.ipcamera.dndSchedule
async def add_rule(self):
pass
async def delete_rule(self):
pass
async def edit_rule(self):
pass
async def get_dnd_enable(self):
pass
async def get_rules(self):
pass
async def set_dnd_enable(self):
pass
# smartlife.cam.ipcamera.intelligence
async def get_all_pd_area(self):
pass
async def get_bcd_enable(self):
pass
async def get_bcd_sensitivity_level(self):
pass
async def get_pd_area(self):
# public List area;
# public Integer max;
# public Integer viewpoint
pass
async def get_pd_enable(self):
pass
async def get_pd_sensitivity_level(self):
# public Integer day_mode_level;
# public Integer night_mode_level
pass
async def set_bcd_enable(self):
pass
async def set_bcd_sensitivity_level(self):
pass
async def set_pd_area(self, area: list, viewpoint: int):
pass
async def set_pd_enable(self):
pass
async def set_pd_sensitivity_level(
self, day_mode_level: int, night_mode_level: int
):
pass
# smartlife.cam.ipcamera.led
async def get_buttonled_status(self):
pass
async def set_buttonled_status(self):
pass
async def set_status(self):
pass
async def get_status(self):
pass
# smartlife.cam.ipcamera.motionDetect
async def get_detect_area(self):
# public List area;
# public Integer max;
# public Integer viewpoint;
pass
async def get_is_enable(self):
pass
async def get_min_trigger_time(self):
# public Integer day_mode_value;
# public Integer night_mode_value;
pass
async def get_sensitivity(self):
pass
async def get_sensitivity_level(self):
# public Integer day_mode_level;
# public Integer night_mode_level;
pass
async def set_detect_area(self, area: list, viewpoint: int):
pass
async def set_is_enable(self):
pass
async def set_min_trigger_time(self, day_mode_level: int, night_mode_level: int):
pass
async def set_sensitivity(self):
pass
async def set_sensitivity_level(self, day_mode_level: int, night_mode_level: int):
pass
# smartlife.cam.ipcamera.OSD
async def get_logo_is_enable(self):
pass
async def get_time_is_enable(self):
pass
async def set_logo_is_enable(self):
pass
async def set_time_is_enable(self):
pass
# smartlife.cam.ipcamera.ptz
async def add_preset(self):
# public String api_srv_url;
# public Integer index;
# public String name;
# public String path;
# public Integer wait_time;
pass
async def delete_preset(self, index: int):
pass
async def edit_preset(self, index: int, name: str, wait_time: int):
pass
async def get_all_preset(self):
# public Integer maximum;
# public List preset_attr;
pass
async def get_patrol_is_enable(self):
pass
async def get_position(self):
# public Integer x;
# public Integer y;
pass
async def get_ptz_rectify_state(self):
pass
async def get_ptz_tracking_is_enable(self):
pass
async def set_motor_rectify(self):
pass
async def set_move(self, x: int, y: int):
pass
async def set_patrol_is_enable(self):
pass
async def set_ptz_tracking_is_enable(self):
pass
async def set_run_to_preset(self, index: int):
pass
async def set_stop(self):
pass
async def set_target(self, direction: str, speed: int):
pass
# smartlife.cam.ipcamera.relay
async def get_preview_snapshot(self):
# public String api_srv_url;
# public String path;
# public String timestamp;
pass
async def set_frame_type(self, frame_type: str):
pass
async def set_prepare_relay(
self,
audio_type: str,
cookie: str,
frame_type: str,
max_video_res: str,
player_id: str,
record_preview: int,
resolution: str,
start_time: int,
stream_url: str,
type: str,
video_type: str,
):
pass
# smartlife.cam.ipcamera.rtp
async def set_prepare_rtc_session(self, sessionId: str, speaker_occupied: bool):
pass
async def set_rtc_session_status(
self, connected: bool, sessionId: str, speaker_occupied: bool
):
pass
# smartlife.cam.ipcamera.camSchedule
async def add_rule(self):
pass
async def delete_all_rules(self):
pass
async def delete_rule(self, id: str):
pass
async def edit_rule(self):
# public String conflict_id;
# public Integer day;
# public Integer eact;
# public Integer emin;
# public Boolean enable;
# public Integer etime_opt;
# public String id;
# public Integer month;
# public String name;
# public Integer repeat;
# public Integer sact;
# public Integer smin;
# public Integer stime_opt;
# public List wday;
# public Integer year;
pass
async def get_rules(self):
# public Boolean enable;
# public List rule_list;
pass
# smartlife.cam.ipcamera.sdCard
async def get_sd_enc_config(self):
# public String sd_enc_enable;
# public String user_key_enable;
pass
async def get_sd_card_state(self):
# public String detect_state;
# public String free;
# public String reserve_capacity;
# public String sd_capacity;
# public String state;
# public String total;
# public String used;
pass
async def set_format_sd_card(self):
pass
async def set_sd_enc_config(
self, key: str, sd_enc_enable: str, user_key_enable: str
):
pass
# smartlife.cam.ipcamera.soundDetect
async def get_is_enable(self):
pass
async def get_sensitivity(self):
pass
async def set_is_enable(self):
pass
async def set_sensitivity(self):
pass
# system
async def get_sysinfo(self):
# public Integer a_type;
# public String account_id;
# public String alias;
# public String battery_charging;
# public Integer battery_percent;
# public List c_opt;
# public String camera_switch;
# public String dev_name;
# public String deviceId;
# public String dnd_switch;
# public List f_list;
# public String hwId;
# public String hw_ver;
# public Long last_activity_timestamp;
# public String mic_mac;
# public String model;
# public List new_feature;
# public String oemId;
# public Boolean online;
# public String power;
# public String resolution;
# public Integer rssi;
# public Integer stream_version;
# public String sw_ver;
# final LinkieCameraCommand.SysInfo this$0;
# public String type;
# public Boolean updating;
pass
# smartlife.cam.ipcamera.system
async def set_alias(self):
pass
async def set_change_local_password(self, c_opt: int, passphrase: str):
pass
async def set_dev_location(self, lattitude: int, longitude: int):
pass
async def set_onboarding_status(self):
pass
async def set_reboot(self, delay_sec: int):
pass
async def set_reset_config(self, isfullreset: bool):
pass
# smartlife.cam.ipcamera.upgrade
async def set_download_firmware(self, flash_sec: int, reboot_sec: int):
pass
async def get_download_status(self):
pass
# smartlife.cam.ipcamera.upnpc
async def get_pub_ip(self):
pass
async def get_upnp_commstatus(self):
pass
async def get_upnp_info(self):
return await self.call("smartlife.cam.ipcamera.upnpc", "get_upnp_info", {})
async def get_upnp_status(self):
pass
async def set_upnp_commstatus(self, comm_status: str, desc: str, timestamp: int):
pass
async def set_upnp_info(self, enabled: str, mode: str):
pass
# smartlife.cam.ipcamera.videoControl
async def get_channel_quality(self):
return await self.call(
"smartlife.cam.ipcamera.videoControl", "get_channel_quality", {}
)
async def get_power_frequency(self):
pass
async def get_resolution(self):
pass
async def get_rotation_degree(self):
pass
async def set_channel_quality(self):
pass
async def set_power_frequency(self):
pass
async def set_resolution(self):
pass
async def set_rotation_degree(self):
pass
# smartlife.cam.ipcamera.vod
async def get_detect_event_state(self):
pass
async def get_detect_zone_brief(self):
pass
async def get_detect_zone_list(self):
pass
async def get_detect_zone_range(self):
pass
async def get_is_enable(self):
return await self.call("smartlife.cam.ipcamera.vod", "get_is_enable", {})
async def get_playback_info(self):
return await self.call("smartlife.cam.ipcamera.vod", "get_playback_info", {})
async def get_record_zone_brief(self):
pass
async def get_record_zone_list(self):
pass
async def get_record_zone_range(self):
pass
async def get_vod_occupied_state(self):
pass
async def reset_vod_connection(self):
pass
async def set_download_playback(
self, player_id: str, start_time: int, user_id: str
):
pass
async def set_heartbeat_playback(self, player_id: str, user_id: str):
pass
async def set_is_enable(self):
pass
async def set_pause_playback(self, player_id: str, user_id: str):
pass
async def set_start_playback(self, player_id: str, start_time: int, user_id: str):
pass
# smartlife.cam.ipcamera.wireless
async def get_ap_list(self):
pass
async def get_connect_status(self):
pass
async def get_uplink(self):
pass
async def set_apply_changes(self, delay: int):
pass
async def set_start_scan(self, band: str, scan_type: str):
pass
async def set_uplink(
self,
apply: bool,
band: str,
encryption: str,
key_index: str,
passphrase: str,
ssid: str,
wpa_mode: str,
):
passI'd like to implement them eventually but don't want that to hold up this PR and likely won't have time for a bit. What are your thoughts on something like a |
| @@ -13,12 +13,16 @@ def query(self): | |||
| q = self.query_for_command("get_time") | |||
|
|
|||
| merge(q, self.query_for_command("get_timezone")) | |||
| merge(q, self.query_for_command("get_time_zone")) | |||
There was a problem hiding this comment.
@rytilahti the time module for my camera needs get_time_zone vs get_timezone for all other devices. This merge seems to work, but I'm concerned whether:
- This works with all other devices (I only have the camera and a couple of lights)
- Is this "The Right Way" to do this?
There was a problem hiding this comment.
Maybe it's better to create a separate class that inherits from the Time, that is only used for the camera class? This way we would not send unnecessary requests to devices that do not support this.
edit: Or maybe just a keyword argument to __init__ that takes the name of the method, something like this:
def __init__(self, device: "SmartDevice", module: str, *, query_command="get_timezone"):
self._query_command = query_command
super().__init__(device, module)
and initializing it with get_time_zone for the camera.
There was a problem hiding this comment.
@rytilahti This was produced by support I added for cameras to the dump_dev_info script but backed it out to a separate, future PR after review discussion. It's not clear to me how much info these fixtures should contain. Should I manually add example responses for the PTZ module commands?
There was a problem hiding this comment.
If you wish to add, that could be helpful for the future. The whole idea behind the fixtures is to have some test responses for commands to hopefully avoid breaking and enabling some development by users (like me) who have just a few devices to test the library against.
It's indeed a good idea to take this in small steps. Instead adding some unused code, how about creating an issue to the issue tracker? |
| credentials: Credentials, | ||
| *, |
There was a problem hiding this comment.
| credentials: Credentials, | |
| *, | |
| *, | |
| credentials: Credentials, |
For consistency, all other classes expect it as kwarg-only.
|
|
||
| Examples: | ||
| >>> import asyncio | ||
| >>> camera = KasaCam("127.0.0.1") |
There was a problem hiding this comment.
Old name used here :-) Please add test_camera_examples to kasa/tests/test_readme_examples.py which will catch similar issues in the future.
There was a problem hiding this comment.
Can you clarify or suggest the fix here?
There was a problem hiding this comment.
The issue is that the name of the class is SmartCamera but the example uses KasaCam.
There was a problem hiding this comment.
Hopefullly I'm not to ignorant here, but is it using "KasaCam" because of the TYPE_TO_CLASS in the kasa/cli.py ?
Line 51 in e2d8f52
There was a problem hiding this comment.
| >>> camera = KasaCam("127.0.0.1") | |
| >>> camera = SmartCamera("127.0.0.1") |
Is that what you are suggesting?
| self, | ||
| target: str, | ||
| cmd: str, | ||
| arg: Optional[Dict] = None, |
There was a problem hiding this comment.
If the Kasa Cameras requires {} in place of "None", why not just put that as the default? I'm not a super Python expert, so open to being wrong here if I misunderstand.
| arg: Optional[Dict] = None, | |
| arg: Optional[Dict] = {}, |
There was a problem hiding this comment.
Generally speaking it is a good idea to avoid passing mutable objects as parameters, as they are initialized at the time of declaration. I.e., if the method body would modify the arg somehow, all future calls would default to having that modified dictionary as input.
In this case it does not make any difference, but given this function overloads the base class one, it should keep using the same signature.
| @@ -156,6 +161,9 @@ def device_for_file(model): | |||
| if d in model: | |||
| return SmartDimmer | |||
|
|
|||
| for d in CAMERAS: | |||
| if d in model: | |||
| return SmartCamera | |||
There was a problem hiding this comment.
| return SmartCamera | |
| return SmartCamera | |
There was a problem hiding this comment.
Adding blank line back in for consistency.
| @@ -68,7 +75,9 @@ def mock_discover(self): | |||
| mocker.patch.object(_DiscoverProtocol, "do_discover", mock_discover) | |||
| mocker.patch("kasa.TPLinkSmartHomeProtocol.query", return_value=discovery_data) | |||
|
|
|||
| x = await Discover.discover_single(host, port=custom_port) | |||
| x = await Discover.discover_single( | |||
| host, port=custom_port, credentials=Credentials("user", "pass") | |||
There was a problem hiding this comment.
| host, port=custom_port, credentials=Credentials("user", "pass") | |
| host, | |
| port=custom_port, | |
| credentials=Credentials("user", "pass") |
There was a problem hiding this comment.
I'm not sure if this is applicable for a test, but the default username/password for the EC70 appears to be admin/admin. But the password is base64 encoded. Which I believe works out to YWRtaW4=.
There was a problem hiding this comment.
This can be tested with
curl -vv -k -u admin:YWRtaW4= "https://192.168.0.10:19443/https/stream/mixed?video=h264&audio=g711&resolution=hd"
or
curl -vv -k -u admin:$(echo -n "admin" | base64) "https://192.168.0.10:19443/https/stream/mixed?video=h264&audio=g711&resolution=hd"
| @@ -80,7 +89,9 @@ async def test_connect_single(discovery_data: dict, mocker, custom_port): | |||
| host = "127.0.0.1" | |||
| mocker.patch("kasa.TPLinkSmartHomeProtocol.query", return_value=discovery_data) | |||
|
|
|||
| dev = await Discover.connect_single(host, port=custom_port) | |||
| dev = await Discover.connect_single( | |||
| host, port=custom_port, credentials=Credentials("user", "pass") | |||
There was a problem hiding this comment.
| host, port=custom_port, credentials=Credentials("user", "pass") | |
| host, | |
| port=custom_port, | |
| credentials=Credentials("user", "pass") |
|
|
||
| Examples: | ||
| >>> import asyncio | ||
| >>> camera = KasaCam("127.0.0.1") |
There was a problem hiding this comment.
The issue is that the name of the class is SmartCamera but the example uses KasaCam.
| self, | ||
| target: str, | ||
| cmd: str, | ||
| arg: Optional[Dict] = None, |
There was a problem hiding this comment.
Generally speaking it is a good idea to avoid passing mutable objects as parameters, as they are initialized at the time of declaration. I.e., if the method body would modify the arg somehow, all future calls would default to having that modified dictionary as input.
In this case it does not make any difference, but given this function overloads the base class one, it should keep using the same signature.
To help here, if I understand the |
|
It is also necessary to convert this to use aiohttp instead of httpx for performance reasons, see #635. |
| @@ -47,6 +48,7 @@ def wrapper(message=None, *args, **kwargs): | |||
| "bulb": SmartBulb, | |||
| "dimmer": SmartDimmer, | |||
| "strip": SmartStrip, | |||
| "kasacam": SmartCamera, | |||
There was a problem hiding this comment.
| "kasacam": SmartCamera, | |
| "camera": SmartCamera, |
|
I have an EC70, and I'd really like to get it working with this. @bstrdsmkr hoping you have some time to wrap this up. Excited to see it included. |
|
So, not sure how much this helps, but the default user name and password for the EC70 is admin:admin. But the password must be base64 encoded when sent. From this Reddit: https://www.reddit.com/r/homeautomation/comments/sej7vi/comment/iee9t0s/ I have verified authentication with Actually, base64('admin') is just This returns a self signed SSL certificate. Hope this may be of some help for the discovery portion. |
|
@bstrdsmkr just pinging to see if you are still interested in this. I'd like to see this implemented, but it's over my head. Plus, I'm probably going to be forced off GitHub due to their MFA requirements tomorrow. |
|
Hi @bstrdsmkr. It's been a while since you commented on this PR and a lot of core library changes have since gone in to support a broader range of tplink devices. That may have made you think it'll be too much trouble to get this PR working but I'd be happy to help you go through it if you're still interested. |
|
There hasn't been any activity on this pull request recently. This pull request has been automatically marked as stale because of that and will be closed if no further activity occurs within 7 days. |
Supports controlling Kasa Cam devices such as the EC70(US)
Closes #530