-
Notifications
You must be signed in to change notification settings - Fork 0
/
shopping_list.py
266 lines (230 loc) · 11.2 KB
/
shopping_list.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
# The MIT License (MIT)
#
# Copyright © 2023 Xavier Berger
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software
# and associated documentation files (the “Software”), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge, publish, distribute,
# sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all copies or
# substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT
# NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import json
import shutil
import time
import appdaemon.plugins.hass.hassapi as hass
# ----------------------------------------------------------------------------------------------------------------------
# Multiple shopping list manager
# ----------------------------------------------------------------------------------------------------------------------
#
# Manage multiple shopping and notification based on Zone.
# Notification gives an access to shopping list when entered into a shop.
#
# ----------------------------------------------------------------------------------------------------------------------
#
# To configure the application follow the instruction below:
#
# Create an input_select gathering the list of shops.
# Options of this input_select are used to select the active list.
#
# Create zones
# Zone are used to define shops area. These shop area are used to automatically select active list and trigger
# notification. The beginning of zone's friendly_name has to match the shop name as defined into options of
# input_select described upper.
#
# Example:
# Zone "zone.Biocoop_Grenoble" and "zone.Biocoop_Modane" will both use the shoppinglist named "Biocoop"
#
# Notifier is a dependency
# Refer to notifier.py documentation activate notification
#
# Configure an AppDeamon application with:
# shopping_list:
# module: shopping_list
# class : ShoppingList
# shops: input_select gathering the shops to manage
# tempo: delay ins seconds between list population and item complete update (recommended: 0.1)
# if complete item are not set corectly, increase this value
# notificationurl: url of shopping list's lovelace card used in notification
# notification_title: title display in notification. This text will be prefixed by the zone name.
# notification_message: message to display in notification
# persons: List of person to notify when they enter into shop zone. At least one person has to be defined.
# - name: username as defined in notifier application (used for notification)
# id: a user as defined in notifier application (used for zone tracking)
#
# Appdaemon configuration example:
# shopping_list:
# module: shopping_list
# class: ShoppingList
# log: shopping_list_log
# shops: input_select.shoppinglist
# tempo: 0.1
# notification_url: "/shopping-list-extended/"
# notification_title: "Shopping list"
# notification_message: "Show shopping list"
# persons:
# - name: user1
# id: person.user1
#
# Lovelace configuration
# Create a new card with a vertical_layout and add the shops' input_select and shopping list card
# as in the yaml example below:
#
# title: Shopping list
# views:
# - cards:
# - type: vertical-stack
# cards:
# - type: entities
# entities:
# - entity: input_select.shops
# - type: shopping-list
#
class ShoppingList(hass.Hass):
def initialize(self):
"""
Initialize the shopping list manager application.
This method sets up the necessary listeners and event handlers for the shopping list manager.
It initializes the active shop change callback, shopping list update callback, and zone change
callbacks for each specified person.
Returns:
None
"""
self.log("Starting multiple shopping list manager")
self.listen_state(self.callback_active_shop_changed, self.args["shops"])
self.listen_event(self.callback_shopping_list_changed, "shopping_list_updated")
if "persons" in self.args:
for person in self.args["persons"]:
self.listen_state(
self.callback_zone_changed,
person["id"],
name=person["name"],
)
# Note: cancel_listen_event has no effect when executed within callback_active_shop_changed
# A workaround to avoid burst call to callback_shopping_list_changed is to manage
# a flag raised during update which deactivate the callback and clear this flag with a timer
self.updating = False
def update_completed(self, cb_args):
"""
Update completed callback for the shopping list manager.
This method is called when the shopping list update process is completed. It sets the 'updating'
flag to False, indicating that the update process is finished.
Args:
cb_args: Callback arguments (not used in this method).
Returns:
None
"""
self.updating = False
self.log("Shopping list updated")
def activate_shop(self, shop):
"""
Activate a new shop and initialize its shopping list.
This method is responsible for changing the active shop and initializing its shopping list.
It first checks if the shopping list is currently being updated and returns early if so.
Then, it performs shopping list update using call_service to homeassistant' shoppinglist plugin.
Args:
shop (str): The shop identifier to activate.
Returns:
bool: True if the shop's shopping list contains incomplete items, False otherwise.
"""
if self.updating is True:
# A shop change has just occurs lets ignore this call
return False
self.log(f"Active shop has changed to {shop}")
# Stop listen on shopping list change since all call_service bellow will add a callback_shopping_list_changed
# call in a FIFO which will be processed once current callback will be completed
self.updating = True
# Clear current shopping list
self.call_service("shopping_list/complete_all")
self.call_service("shopping_list/clear_completed_items")
has_incomplete = False
# Open shop's sopping list backup
with open(f"/config/.shopping_list_{shop}.json", "r") as file:
data = json.load(file)
for item in data:
# Add items from backup
self.call_service("shopping_list/add_item", name=item["name"])
# Note: Complete is set in a second loop and after a tempo because I notice that sometime the list was not
# recreated correctly maybe because of too fast service calls
# /!\ sleep should be avoided in appdaemon application but it mandatory to make update works
# I prefer not using 'run_in' to have have to open the file a second time
time.sleep(self.args["tempo"])
for item in data:
if item["complete"]:
# Set completion from backup
self.call_service("shopping_list/complete_item", name=item["name"])
else:
has_incomplete = True
# Reactivate listen on shopping list change in one second (when callback burst will be finished)
self.run_in(self.update_completed, 1)
return has_incomplete
def callback_active_shop_changed(self, entity, attribute, old, new, kwargs):
"""
Callback for handling changes in the active shop.
This method is called when the active shop changes and triggers the activation of the new shop.
Args:
Arguments as define into Appdaemon callback documentation.
Returns:
None
"""
self.activate_shop(new)
def callback_shopping_list_changed(self, event_name, data, kwargs):
"""
Callback for handling changes in the shopping list.
This method is called when an itel of the shoppinglist is updated and triggers the creation or update of
the backup shopping list for the active shop.
Args:
Arguments as define into Appdaemon event documentation.
Returns:
None
"""
if self.updating is True:
# A shop change has just occurs lets ignore this call
return
# Copy active shopping list to shop's backup
shop = self.get_state(self.args["shops"])
shutil.copyfile("/config/.shopping_list.json", f"/config/.shopping_list_{shop}.json")
def callback_zone_changed(self, entity, attribute, old, new, kwargs):
"""
Callback for handling changes in the zone of a person.
This method is called when the zone of a person changes. It handles actions based on entering
or leaving a zone, such as loading the appropriate shopping list and sending or clearing notifications.
Args:
Arguments as define into Appdaemon callback documentation.
Returns:
None
"""
self.log(f"Zone changed to {new} for {entity}")
if self.get_state(f"zone.{new.lower()}") is not None:
# Entering in a shop
shop_zone = self.get_state(f"zone.{new.lower()}", attribute="friendly_name")
for shop in self.get_state(self.args["shops"], attribute="options"):
if shop_zone.startswith(shop):
self.log(f"{shop} > loading shopping list")
has_incomplete = self.activate_shop(shop)
self.log(f"{shop} > shopping list loaded.")
self.select_option(self.args["shops"], shop)
self.log(f"{shop} > input_select updated.")
# Send notification only if incomplete item are present in the list
if has_incomplete:
self.log("Send notification")
self.fire_event(
"NOTIFIER",
action=f"send_to_{kwargs['name']}",
title=f"{shop}: {self.args['notification_title']}",
message=self.args["notification_message"],
icon="mdi-cart",
color="deep-orange",
tag="shoppinglist",
click_url=self.args["notification_url"],
until=[
{"entity_id": entity, "old_state": shop_zone},
],
)