Skip to content

Commit 54ef91d

Browse files
brefrabouwew
brefra
authored and
bouwew
committed
Use context manager for locking
1 parent 7b6a1a6 commit 54ef91d

File tree

1 file changed

+144
-149
lines changed

1 file changed

+144
-149
lines changed

‎plugwise/controller.py‎

Lines changed: 144 additions & 149 deletions
Original file line numberDiff line numberDiff line change
@@ -152,68 +152,67 @@ def send(
152152
def resend(self, seq_id):
153153
"""Resend message."""
154154
_mac = "<unknown>"
155-
self.lock_expected_responses.acquire()
156-
if not self.expected_responses.get(seq_id):
157-
_LOGGER.warning(
158-
"Cannot resend unknown request %s",
159-
str(seq_id),
160-
)
161-
else:
162-
if self.expected_responses[seq_id][0].mac:
163-
_mac = self.expected_responses[seq_id][0].mac.decode(UTF8_DECODE)
164-
_request = self.expected_responses[seq_id][0].__class__.__name__
155+
with self.lock_expected_responses:
156+
if not self.expected_responses.get(seq_id):
157+
_LOGGER.warning(
158+
"Cannot resend unknown request %s",
159+
str(seq_id),
160+
)
161+
else:
162+
if self.expected_responses[seq_id][0].mac:
163+
_mac = self.expected_responses[seq_id][0].mac.decode(UTF8_DECODE)
164+
_request = self.expected_responses[seq_id][0].__class__.__name__
165165

166-
if self.expected_responses[seq_id][2] == -1:
167-
_LOGGER.debug("Drop single %s to %s ", _request, _mac)
168-
elif self.expected_responses[seq_id][2] <= MESSAGE_RETRY:
169-
if (
170-
isinstance(self.expected_responses[seq_id][0], NodeInfoRequest)
171-
and not self.discovery_finished
172-
):
173-
# Time out for node which is not discovered yet
174-
# to speedup the initial discover phase skip retries and mark node as not discovered.
175-
_LOGGER.debug(
176-
"Skip retry %s to %s to speedup discover process",
177-
_request,
178-
_mac,
179-
)
180-
if self.expected_responses[seq_id][1]:
181-
self.expected_responses[seq_id][1]()
166+
if self.expected_responses[seq_id][2] == -1:
167+
_LOGGER.debug("Drop single %s to %s ", _request, _mac)
168+
elif self.expected_responses[seq_id][2] <= MESSAGE_RETRY:
169+
if (
170+
isinstance(self.expected_responses[seq_id][0], NodeInfoRequest)
171+
and not self.discovery_finished
172+
):
173+
# Time out for node which is not discovered yet
174+
# to speedup the initial discover phase skip retries and mark node as not discovered.
175+
_LOGGER.debug(
176+
"Skip retry %s to %s to speedup discover process",
177+
_request,
178+
_mac,
179+
)
180+
if self.expected_responses[seq_id][1]:
181+
self.expected_responses[seq_id][1]()
182+
else:
183+
_LOGGER.info(
184+
"Resend %s for %s, retry %s of %s",
185+
_request,
186+
_mac,
187+
str(self.expected_responses[seq_id][2] + 1),
188+
str(MESSAGE_RETRY + 1),
189+
)
190+
self.send(
191+
self.expected_responses[seq_id][0],
192+
self.expected_responses[seq_id][1],
193+
self.expected_responses[seq_id][2] + 1,
194+
)
182195
else:
183-
_LOGGER.info(
184-
"Resend %s for %s, retry %s of %s",
196+
_LOGGER.warning(
197+
"Drop %s to %s because max retries %s reached",
185198
_request,
186199
_mac,
187-
str(self.expected_responses[seq_id][2] + 1),
188200
str(MESSAGE_RETRY + 1),
189201
)
190-
self.send(
191-
self.expected_responses[seq_id][0],
192-
self.expected_responses[seq_id][1],
193-
self.expected_responses[seq_id][2] + 1,
194-
)
195-
else:
196-
_LOGGER.warning(
197-
"Drop %s to %s because max retries %s reached",
198-
_request,
199-
_mac,
200-
str(MESSAGE_RETRY + 1),
201-
)
202-
# Report node as unavailable for missing NodePingRequest
203-
if isinstance(self.expected_responses[seq_id][0], NodePingRequest):
204-
self.node_state(_mac, False)
205-
else:
206-
_LOGGER.debug(
207-
"Do a single ping request to %s to validate if node is reachable",
208-
_mac,
209-
)
210-
self.send(
211-
NodePingRequest(self.expected_responses[seq_id][0].mac),
212-
None,
213-
MESSAGE_RETRY + 1,
214-
)
215-
del self.expected_responses[seq_id]
216-
self.lock_expected_responses.release()
202+
# Report node as unavailable for missing NodePingRequest
203+
if isinstance(self.expected_responses[seq_id][0], NodePingRequest):
204+
self.node_state(_mac, False)
205+
else:
206+
_LOGGER.debug(
207+
"Do a single ping request to %s to validate if node is reachable",
208+
_mac,
209+
)
210+
self.send(
211+
NodePingRequest(self.expected_responses[seq_id][0].mac),
212+
None,
213+
MESSAGE_RETRY + 1,
214+
)
215+
del self.expected_responses[seq_id]
217216

218217
def _send_message_loop(self):
219218
"""Daemon to send messages waiting in queue."""
@@ -228,27 +227,26 @@ def _send_message_loop(self):
228227
# Calc next seq_id based last received ack message
229228
# if previous seq_id is unknown use fake b"0000"
230229
seq_id = inc_seq_id(self.last_seq_id)
231-
self.lock_expected_responses.acquire()
232-
self.expected_responses[seq_id] = request_set
233-
if self.expected_responses[seq_id][2] == 0:
234-
_LOGGER.info(
235-
"Send %s to %s using seq_id %s",
236-
self.expected_responses[seq_id][0].__class__.__name__,
237-
self.expected_responses[seq_id][0].mac.decode(UTF8_DECODE),
238-
str(seq_id),
239-
)
240-
else:
241-
_LOGGER.info(
242-
"Resend %s to %s using seq_id %s, retry %s",
243-
self.expected_responses[seq_id][0].__class__.__name__,
244-
self.expected_responses[seq_id][0].mac.decode(UTF8_DECODE),
245-
str(seq_id),
246-
str(self.expected_responses[seq_id][2]),
247-
)
248-
self.expected_responses[seq_id][3] = datetime.now()
249-
# Send request
250-
self.connection.send(self.expected_responses[seq_id][0])
251-
self.lock_expected_responses.release()
230+
with self.lock_expected_responses:
231+
self.expected_responses[seq_id] = request_set
232+
if self.expected_responses[seq_id][2] == 0:
233+
_LOGGER.info(
234+
"Send %s to %s using seq_id %s",
235+
self.expected_responses[seq_id][0].__class__.__name__,
236+
self.expected_responses[seq_id][0].mac.decode(UTF8_DECODE),
237+
str(seq_id),
238+
)
239+
else:
240+
_LOGGER.info(
241+
"Resend %s to %s using seq_id %s, retry %s",
242+
self.expected_responses[seq_id][0].__class__.__name__,
243+
self.expected_responses[seq_id][0].mac.decode(UTF8_DECODE),
244+
str(seq_id),
245+
str(self.expected_responses[seq_id][2]),
246+
)
247+
self.expected_responses[seq_id][3] = datetime.now()
248+
# Send request
249+
self.connection.send(self.expected_responses[seq_id][0])
252250
time.sleep(SLEEP_TIME)
253251
timeout_counter = 0
254252
# Wait max 1 second for acknowledge response from USB-stick
@@ -292,66 +290,64 @@ def message_handler(self, message):
292290

293291
def _post_message_action(self, seq_id, ack_response=None, request="unknown"):
294292
"""Execute action if request has been successful."""
295-
self.lock_expected_responses.acquire()
296-
if seq_id in self.expected_responses:
297-
if ack_response in (*REQUEST_SUCCESS, None):
298-
if self.expected_responses[seq_id][1]:
299-
_LOGGER.debug(
300-
"Execute action %s of request with seq_id %s",
301-
self.expected_responses[seq_id][1].__name__,
302-
str(seq_id),
303-
)
304-
try:
305-
self.expected_responses[seq_id][1]()
306-
# TODO: narrow exception
307-
except Exception as err: # pylint: disable=broad-except
308-
_LOGGER.error(
309-
"Execution of %s for request with seq_id %s failed: %s",
293+
with self.lock_expected_responses:
294+
if seq_id in self.expected_responses:
295+
if ack_response in (*REQUEST_SUCCESS, None):
296+
if self.expected_responses[seq_id][1]:
297+
_LOGGER.debug(
298+
"Execute action %s of request with seq_id %s",
310299
self.expected_responses[seq_id][1].__name__,
311300
str(seq_id),
312-
err,
313301
)
314-
del self.expected_responses[seq_id]
315-
elif ack_response in REQUEST_FAILED:
316-
self.resend(seq_id)
317-
else:
318-
if not self.last_seq_id:
319-
if b"0000" in self.expected_responses:
320-
self.expected_responses[seq_id] = self.expected_responses[b"0000"]
321-
del self.expected_responses[b"0000"]
322-
self.last_seq_id = seq_id
302+
try:
303+
self.expected_responses[seq_id][1]()
304+
# TODO: narrow exception
305+
except Exception as err: # pylint: disable=broad-except
306+
_LOGGER.error(
307+
"Execution of %s for request with seq_id %s failed: %s",
308+
self.expected_responses[seq_id][1].__name__,
309+
str(seq_id),
310+
err,
311+
)
312+
del self.expected_responses[seq_id]
313+
elif ack_response in REQUEST_FAILED:
314+
self.resend(seq_id)
323315
else:
324-
_LOGGER.info(
325-
"Drop unexpected %s%s using seq_id %s",
326-
STATUS_RESPONSES.get(ack_response, "") + " ",
327-
request,
328-
str(seq_id),
329-
)
330-
self.lock_expected_responses.release()
316+
if not self.last_seq_id:
317+
if b"0000" in self.expected_responses:
318+
self.expected_responses[seq_id] = self.expected_responses[b"0000"]
319+
del self.expected_responses[b"0000"]
320+
self.last_seq_id = seq_id
321+
else:
322+
_LOGGER.info(
323+
"Drop unexpected %s%s using seq_id %s",
324+
STATUS_RESPONSES.get(ack_response, "") + " ",
325+
request,
326+
str(seq_id),
327+
)
331328

332329
def _receive_timeout_loop(self):
333330
"""Daemon to time out open requests without any (n)ack response message."""
334331
while self._receive_timeout_thread_state:
335-
self.lock_expected_responses.acquire()
336-
for seq_id in list(self.expected_responses.keys()):
337-
if self.expected_responses[seq_id][3] is not None:
338-
if self.expected_responses[seq_id][3] < (
339-
datetime.now() - timedelta(seconds=MESSAGE_TIME_OUT)
340-
):
341-
_mac = "<unknown>"
342-
if self.expected_responses[seq_id][0].mac:
343-
_mac = self.expected_responses[seq_id][0].mac.decode(
344-
UTF8_DECODE
332+
with self.lock_expected_responses:
333+
for seq_id in list(self.expected_responses.keys()):
334+
if self.expected_responses[seq_id][3] is not None:
335+
if self.expected_responses[seq_id][3] < (
336+
datetime.now() - timedelta(seconds=MESSAGE_TIME_OUT)
337+
):
338+
_mac = "<unknown>"
339+
if self.expected_responses[seq_id][0].mac:
340+
_mac = self.expected_responses[seq_id][0].mac.decode(
341+
UTF8_DECODE
342+
)
343+
_LOGGER.info(
344+
"No response within %s seconds timeout for %s to %s with sequence ID %s",
345+
str(MESSAGE_TIME_OUT),
346+
self.expected_responses[seq_id][0].__class__.__name__,
347+
_mac,
348+
str(seq_id),
345349
)
346-
_LOGGER.info(
347-
"No response within %s seconds timeout for %s to %s with sequence ID %s",
348-
str(MESSAGE_TIME_OUT),
349-
self.expected_responses[seq_id][0].__class__.__name__,
350-
_mac,
351-
str(seq_id),
352-
)
353-
self.resend(seq_id)
354-
self.lock_expected_responses.release()
350+
self.resend(seq_id)
355351
receive_timeout_checker = 0
356352
while (
357353
receive_timeout_checker < MESSAGE_TIME_OUT
@@ -372,27 +368,26 @@ def _log_status_message(self, message, status=None):
372368
str(message.seq_id),
373369
)
374370
else:
375-
self.lock_expected_responses.acquire()
376-
if self.expected_responses.get(message.seq_id):
377-
_LOGGER.warning(
378-
"Received unmanaged (%s) %s in response to %s with seq_id %s",
379-
str(status),
380-
message.__class__.__name__,
381-
str(
382-
self.expected_responses[message.seq_id][
383-
1
384-
].__class__.__name__
385-
),
386-
str(message.seq_id),
387-
)
388-
else:
389-
_LOGGER.warning(
390-
"Received unmanaged (%s) %s for unknown request with seq_id %s",
391-
str(status),
392-
message.__class__.__name__,
393-
str(message.seq_id),
394-
)
395-
self.lock_expected_responses.release()
371+
with self.lock_expected_responses:
372+
if self.expected_responses.get(message.seq_id):
373+
_LOGGER.warning(
374+
"Received unmanaged (%s) %s in response to %s with seq_id %s",
375+
str(status),
376+
message.__class__.__name__,
377+
str(
378+
self.expected_responses[message.seq_id][
379+
1
380+
].__class__.__name__
381+
),
382+
str(message.seq_id),
383+
)
384+
else:
385+
_LOGGER.warning(
386+
"Received unmanaged (%s) %s for unknown request with seq_id %s",
387+
str(status),
388+
message.__class__.__name__,
389+
str(message.seq_id),
390+
)
396391
else:
397392
_LOGGER.info(
398393
"Received %s from %s with sequence id %s",

0 commit comments

Comments
 (0)