| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -569,6 +569,26 @@ | |
| return res | ||
|
|
||
|
|
||
| @cli.command() | ||
| @click.option("--list", is_flag=True) | ||
| @click.argument("index", type=int, required=False) | ||
| @pass_dev | ||
| async def set_timezone(dev, list, index): | ||
| """Set the device timezone. | ||
|
|
||
| pass --list to see valid timezones | ||
| """ | ||
| time = dev.modules["time"] | ||
| if list: | ||
| for timezone in time.get_timezones(): | ||
| echo( | ||
| f"Index: {timezone['index']:3}\tName: {timezone['zone_str']:65}\tRule: {timezone['tz_str']}" | ||
| ) | ||
| return | ||
| if index: | ||
| return await time.set_timezone(index) | ||
|
|
||
|
|
||
| @cli.command() | ||
| @click.option("--index", type=int, required=False) | ||
| @click.option("--name", type=str, required=False) | ||
| Expand Down Expand Up | @@ -630,18 +650,136 @@ | |
|
|
||
| @schedule.command(name="list") | ||
| @pass_dev | ||
| @click.option("--json", is_flag=True) | ||
| @click.argument("type", default="schedule") | ||
| def _schedule_list(dev, type): | ||
| def _schedule_list(dev, json, type): | ||
| """Return the list of schedule actions for the given type.""" | ||
| sched = dev.modules[type] | ||
| for rule in sched.rules: | ||
| print(rule) | ||
| else: | ||
| if sched.rules == []: | ||
| echo(f"No rules of type {type}") | ||
| for rule in sched.rules: | ||
| if json: | ||
| print(rule.json()) | ||
| else: | ||
| print(rule) | ||
|
|
||
| return sched.rules | ||
|
|
||
|
|
||
| @schedule.command(name="enable") | ||
| @pass_dev | ||
| @click.argument("enable", type=click.BOOL) | ||
| async def _schedule_enable(dev, enable): | ||
| """Enable or disable schedule.""" | ||
| schedule = dev.modules["schedule"] | ||
| return await schedule.set_enabled(1 if state else 0) | ||
|
|
||
|
|
||
| @schedule.command(name="add") | ||
| @pass_dev | ||
| @click.option("--name", type=str, required=True) | ||
| @click.option("--enable", type=click.BOOL, default=True, show_default=True) | ||
| @click.option("--repeat", type=click.BOOL, default=True, show_default=True) | ||
| @click.option("--days", type=str, required=True) | ||
| @click.option("--start-action", type=click.IntRange(-1, 2), default=None, required=True) | ||
| @click.option("--start-sun", type=click.IntRange(-1, 2), default=None, required=True) | ||
| @click.option( | ||
| "--start-minutes", type=click.IntRange(0, 1440), default=None, required=True | ||
| ) | ||
| @click.option("--end-action", type=click.IntRange(-1, 2), default=-1) | ||
| @click.option("--end-sun", type=click.IntRange(-1, 2), default=-1) | ||
| @click.option("--end-minutes", type=click.IntRange(0, 1440), default=None) | ||
| async def add_rule( | ||
| dev, | ||
| name, | ||
| enable, | ||
| repeat, | ||
| days, | ||
| start_action, | ||
| start_sun, | ||
| start_minutes, | ||
| end_action, | ||
| end_sun, | ||
| end_minutes, | ||
| ): | ||
| """Add rule to device.""" | ||
| schedule = dev.modules["schedule"] | ||
| rule_to_add = schedule.Rule( | ||
| name=name, | ||
| enable=enable, | ||
| repeat=repeat, | ||
| days=list(map(int, days.split(","))), | ||
| start_action=start_action, | ||
| start_sun=start_sun, | ||
| start_minutes=start_minutes, | ||
| end_action=end_action, | ||
| end_sun=end_sun, | ||
| end_minutes=end_minutes, | ||
| ) | ||
| if rule_to_add: | ||
| echo("Adding rule") | ||
| return await schedule.add_rule(rule_to_add) | ||
| else: | ||
| echo("Invalid rule") | ||
|
|
||
|
|
||
| @schedule.command(name="edit") | ||
| @pass_dev | ||
| @click.option("--id", type=str, required=True) | ||
| @click.option("--name", type=str) | ||
| @click.option("--enable", type=click.BOOL) | ||
| @click.option("--repeat", type=click.BOOL) | ||
| @click.option("--days", type=str) | ||
| @click.option("--start-action", type=click.IntRange(-1, 2)) | ||
| @click.option("--start-sun", type=click.IntRange(-1, 2)) | ||
| @click.option("--start-minutes", type=click.IntRange(0, 1440)) | ||
| @click.option("--end-action", type=click.IntRange(-1, 2)) | ||
| @click.option("--end-sun", type=click.IntRange(-1, 2)) | ||
| @click.option("--end-minutes", type=click.IntRange(0, 1440)) | ||
| async def edit_rule( | ||
| dev, | ||
| id, | ||
| name, | ||
| enable, | ||
| repeat, | ||
| days, | ||
| start_action, | ||
| start_sun, | ||
| start_minutes, | ||
| end_action, | ||
| end_sun, | ||
| end_minutes, | ||
| ): | ||
| """Edit rule from device.""" | ||
| schedule = dev.modules["schedule"] | ||
| rule_to_edit = next(filter(lambda rule: (rule.id == id), schedule.rules), None) | ||
| if rule_to_edit: | ||
| echo(f"Editing rule id {id}") | ||
| if name is not None: | ||
| rule_to_edit.name = name | ||
| if enable is not None: | ||
| rule_to_edit.enable = 1 if enable else 0 | ||
|
Comment thread
Copy link
Copy Markdown
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Qualitypydantic should handle the conversion just fine as long as the types are correct, so there should be no need for manual conversions here.
Sorry, something went wrong.
All reactions
Copy link
Copy Markdown
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityI want that only the user specified options should override the existing rule fields, however if the user does not specify a flag we get a None value that i don't want to pass to pydantic to overwrite the field. If there is a better way to do this with click & pydantic please let me know. (I am not too familiar with either of them)
Sorry, something went wrong.
All reactions
|
||
| if repeat is not None: | ||
| rule_to_edit.repeat = 1 if repeat else 0 | ||
| if days is not None: | ||
| rule_to_edit.wday = list(map(int, days.split(","))) | ||
| if start_action is not None: | ||
| rule_to_edit.sact = start_action | ||
| if start_sun is not None: | ||
| rule_to_edit.stime_opt = start_sun | ||
| if start_minutes is not None: | ||
| rule_to_edit.smin = start_minutes | ||
| if end_action is not None: | ||
| rule_to_edit.eact = end_action | ||
| if end_sun is not None: | ||
| rule_to_edit.etime_opt = end_sun | ||
| if end_minutes is not None: | ||
| rule_to_edit.emin = end_minutes | ||
| return await schedule.edit_rule(rule_to_edit) | ||
| else: | ||
| echo(f"No rule with id {id} was found") | ||
|
|
||
|
|
||
| @schedule.command(name="delete") | ||
| @pass_dev | ||
| @click.option("--id", type=str, required=True) | ||
| Expand All | @@ -656,6 +794,17 @@ | |
| echo(f"No rule with id {id} was found") | ||
|
|
||
|
|
||
| @schedule.command() | ||
| @pass_dev | ||
| @click.option("--prompt", type=click.BOOL, prompt=True, help="Are you sure?") | ||
| async def delete_all(dev, prompt): | ||
| """Delete all rules from device.""" | ||
| schedule = dev.modules["schedule"] | ||
| if prompt: | ||
| echo("Deleting all rules") | ||
| return await schedule.delete_all_rules() | ||
|
|
||
|
|
||
| @cli.group(invoke_without_command=True) | ||
| @click.pass_context | ||
| async def presets(ctx): | ||
| Expand Down | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,9 +1,11 @@ | ||
| """Implementation of the antitheft module.""" | ||
| from .rulemodule import RuleModule | ||
| from .rulemodule import AntitheftRule, RuleModule | ||
|
|
||
|
|
||
| class Antitheft(RuleModule): | ||
| """Implementation of the antitheft module. | ||
|
|
||
| This shares the functionality among other rule-based modules. | ||
| """ | ||
|
|
||
| Rule = AntitheftRule |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,8 @@ | ||
| """Implementation for the countdown timer.""" | ||
| from .rulemodule import RuleModule | ||
| from .rulemodule import CountdownRule, RuleModule | ||
|
|
||
|
|
||
| class Countdown(RuleModule): | ||
| """Implementation of countdown module.""" | ||
|
|
||
| Rule = CountdownRule |
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -1,13 +1,21 @@ | ||||||
| """Base implementation for all rule-based modules.""" | ||||||
| import json | ||||||
| import logging | ||||||
| from enum import Enum | ||||||
| from typing import Dict, List, Optional | ||||||
| from typing import List, Optional | ||||||
|
|
||||||
| from pydantic import BaseModel | ||||||
| from pydantic import BaseModel, Field | ||||||
|
|
||||||
| from .module import Module, merge | ||||||
|
|
||||||
|
|
||||||
| class EnabledOption(Enum): | ||||||
| """Integer enabled option.""" | ||||||
|
|
||||||
| TurnOff = 0 | ||||||
| Enabled = 1 | ||||||
|
|
||||||
|
|
||||||
| class Action(Enum): | ||||||
| """Action to perform.""" | ||||||
|
|
||||||
| Expand All | @@ -26,55 +34,114 @@ | |||||
| AtSunset = 2 | ||||||
|
|
||||||
|
|
||||||
| class Rule(BaseModel): | ||||||
| class BulbModeOption(Enum): | ||||||
| """Bulb mode.""" | ||||||
|
|
||||||
| Customize = "customize_preset" | ||||||
| Last = "last_status" | ||||||
|
|
||||||
|
|
||||||
| class BaseRule(BaseModel): | ||||||
| """Representation of a rule.""" | ||||||
|
|
||||||
| id: str | ||||||
| # not used when adding a rule | ||||||
| id: Optional[str] | ||||||
| name: str | ||||||
| enable: bool | ||||||
| wday: List[int] | ||||||
| repeat: bool | ||||||
| enable: EnabledOption | ||||||
|
|
||||||
| class Config: | ||||||
| """Rule Config.""" | ||||||
|
|
||||||
| validate_assignment = True | ||||||
| allow_population_by_field_name = True | ||||||
|
|
||||||
|
|
||||||
| class CountdownRule(BaseRule): | ||||||
| """Representation of a countdown rule.""" | ||||||
|
|
||||||
| delay: int | ||||||
| act: EnabledOption = Field(alias="action") | ||||||
|
|
||||||
|
|
||||||
| class ScheduleRule(BaseRule): | ||||||
| """Representation of a schedule rule.""" | ||||||
|
|
||||||
| wday: List[int] = Field(alias="days") | ||||||
| repeat: EnabledOption | ||||||
|
|
||||||
| # start action | ||||||
| sact: Optional[Action] | ||||||
| stime_opt: TimeOption | ||||||
| smin: int | ||||||
| sact: Optional[Action] = Field(alias="start_action") | ||||||
|
Comment thread
Copy link
Copy Markdown
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality
Suggested change
This is wrong way around, no? Basically, the alias allows using a nicer name for the field name itself, reading (and serializing?) in this case from/to a json field called sact.
Sorry, something went wrong.
All reactions
|
||||||
| stime_opt: TimeOption = Field(alias="start_sun") | ||||||
| smin: int = Field(alias="start_minutes", ge=0, le=1440) | ||||||
|
|
||||||
| # end action | ||||||
| eact: Optional[Action] = Field(alias="end_action") | ||||||
| # Required to submit, but the device will not return it if set to -1 | ||||||
| etime_opt: TimeOption = Field(default=TimeOption.Disabled, alias="end_sun") | ||||||
| emin: Optional[int] = Field(alias="end_minutes", ge=0, le=1440) | ||||||
|
|
||||||
|
|
||||||
| class BulbRule(BaseRule): | ||||||
| """Representation of a bulb schedule rule.""" | ||||||
|
|
||||||
| saturation: int = Field(ge=0, le=100) | ||||||
| hue: int = Field(ge=0, le=360) | ||||||
| brightness: int = Field(ge=0, le=100) | ||||||
| color_temp: int = Field(ge=2500, le=9000) | ||||||
| mode: BulbModeOption | ||||||
| on_off: EnabledOption | ||||||
|
|
||||||
|
|
||||||
| eact: Optional[Action] | ||||||
| etime_opt: TimeOption | ||||||
| emin: int | ||||||
| class BulbScheduleRule(BaseRule): | ||||||
| """Representation of a bulb schedule rule.""" | ||||||
|
|
||||||
| # Only on bulbs | ||||||
| s_light: Optional[Dict] | ||||||
| s_light: BulbRule = Field(alias="lights") | ||||||
|
|
||||||
|
|
||||||
| class AntitheftRule(BaseRule): | ||||||
| """Representation of a antitheft rule.""" | ||||||
|
|
||||||
| frequency: int = Field(ge=1, le=10) | ||||||
|
|
||||||
|
|
||||||
| _LOGGER = logging.getLogger(__name__) | ||||||
|
|
||||||
|
|
||||||
| class RuleModule(Module): | ||||||
| """Base class for rule-based modules, such as countdown and antitheft.""" | ||||||
| """Base class for rule-based modules, such as antitheft, countdown and schedule.""" | ||||||
|
|
||||||
| Rule = BaseRule | ||||||
|
|
||||||
| def query(self): | ||||||
| """Prepare the query for rules.""" | ||||||
| q = self.query_for_command("get_rules") | ||||||
| return merge(q, self.query_for_command("get_next_action")) | ||||||
|
|
||||||
| @property | ||||||
| def rules(self) -> List[Rule]: | ||||||
| def rules(self) -> List[BaseRule]: | ||||||
| """Return the list of rules for the service.""" | ||||||
| try: | ||||||
| return [ | ||||||
| Rule.parse_obj(rule) for rule in self.data["get_rules"]["rule_list"] | ||||||
| self.Rule.parse_obj(rule) | ||||||
| for rule in self.data["get_rules"]["rule_list"] | ||||||
| ] | ||||||
| except Exception as ex: | ||||||
| _LOGGER.error("Unable to read rule list: %s (data: %s)", ex, self.data) | ||||||
| return [] | ||||||
|
|
||||||
| async def set_enabled(self, state: bool): | ||||||
| async def set_enabled(self, state: int): | ||||||
| """Enable or disable the service.""" | ||||||
| return await self.call("set_overall_enable", state) | ||||||
| return await self.call("set_overall_enable", {"enable": state}) | ||||||
|
|
||||||
| async def add_rule(self, rule: BaseRule): | ||||||
| """Add a new rule.""" | ||||||
| return await self.call("add_rule", json.loads(rule.json(exclude_none=True))) | ||||||
|
|
||||||
| async def edit_rule(self, rule: BaseRule): | ||||||
| """Edit the given rule.""" | ||||||
| return await self.call("edit_rule", json.loads(rule.json(exclude_none=True))) | ||||||
|
|
||||||
| async def delete_rule(self, rule: Rule): | ||||||
| async def delete_rule(self, rule: BaseRule): | ||||||
| """Delete the given rule.""" | ||||||
| return await self.call("delete_rule", {"id": rule.id}) | ||||||
|
|
||||||
| Expand Down | ||||||
| Back | FazBrowse Home | New Git URL |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Qualityjsonifying is handled on the cli automatically for the return value, so there should be no need for special handling.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityThis way i get the actual json that is passed to the device, not the python rule wrapped in json
$ kasa --host 192.168.0.xxx --type plug schedule list id='B9A82B5BA5AB6F6DA4277429A226D632' name='7 Off 11 On' enable=<EnabledOption.Enabled: 1> wday=[1, 1, 1, 1, 1, 1, 1] repeat=<EnabledOption.Enabled: 1> sact=<Action.TurnOff: 0> stime_opt=<TimeOption.Enabled: 0> smin=420 eact=<Action.TurnOn: 1> etime_opt=<TimeOption.Enabled: 0> emin=660 s_light=None $ kasa --host 192.168.0.xxx --type plug --json schedule list id='B9A82B5BA5AB6F6DA4277429A226D632' name='7 Off 11 On' enable=<EnabledOption.Enabled: 1> wday=[1, 1, 1, 1, 1, 1, 1] repeat=<EnabledOption.Enabled: 1> sact=<Action.TurnOff: 0> stime_opt=<TimeOption.Enabled: 0> smin=420 eact=<Action.TurnOn: 1> etime_opt=<TimeOption.Enabled: 0> emin=660 s_light=None [ "id='B9A82B5BA5AB6F6DA4277429A226D632' name='7 Off 11 On' enable=<EnabledOption.Enabled: 1> wday=[1, 1, 1, 1, 1, 1, 1] repeat=<EnabledOption.Enabled: 1> sact=<Action.TurnOff: 0> stime_opt=<TimeOption.Enabled: 0> smin=420 eact=<Action.TurnOn: 1> etime_opt=<TimeOption.Enabled: 0> emin=660 s_light=None" ] $ kasa --host 192.168.0.xxx --type plug schedule list --json {"id": "B9A82B5BA5AB6F6DA4277429A226D632", "name": "7 Off 11 On", "enable": 1, "wday": [1, 1, 1, 1, 1, 1, 1], "repeat": 1, "sact": 0, "stime_opt": 0, "smin": 420, "eact": 1, "etime_opt": 0, "emin": 660, "s_light": null}Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityThis needs to be fixed separately then (I think the proper way would be extending the json_formatter_cb by defining a serializer for pydantic BaseModel), but let's do that in a separate PR.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.