-
Notifications
You must be signed in to change notification settings - Fork 5
/
device.py
671 lines (568 loc) · 25.6 KB
/
device.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
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
"""aiobafi6 device.
Provides functionality to query and control BAF i6 protocol devices.
"""
from __future__ import annotations
import asyncio
import copy
import inspect
import logging
import time
import typing as t
from google.protobuf import json_format
from google.protobuf.message import Message
from . import wireutils
from .const import DELAY_BETWEEN_CONNECTS_SECONDS, OCCUPANCY_MIN_API_VERSION
from .discovery import Service
from .exceptions import DeviceUUIDMismatchError
from .proto import aiobafi6_pb2
from .protoprop import (
ClosedIntervalValidator,
OffOnAuto,
ProtoProp,
from_proto_humidity,
from_proto_temperature,
maybe_proto_field,
to_proto_temperature,
)
__all__ = (
"VOLATILE_PROPERTIES",
"Device",
)
VOLATILE_PROPERTIES = (
"current_rpm",
"local_datetime",
"stats",
"utc_datetime",
)
_LOGGER = logging.getLogger(__name__)
_MAX_SPEED = 7
_RECV_BUFFER_LIMIT = 4096 # No message is ever expected to be > 4K
_PROPS_REQUIRED_FOR_AVAILABLE = (
"name",
"model",
"firmware_version",
"mac_address",
"dns_sd_uuid",
"capabilities",
"ip_address",
)
def _clear_volatile_props(props: aiobafi6_pb2.Properties):
"""Clear volatile properties from `props`."""
for field in VOLATILE_PROPERTIES:
props.ClearField(field)
class Device:
"""A BAF i6 protocol device.
The design the of class is relatively simple. Since the protocol is based on
protofbuf, the majority of a device's state can be stored in a `Properties` message.
The query loop simply updates this message using `MergeFrom`, with unknown fields
removed (as they are otherwise treated as repeated fields and would lead to unbound
memory growth). Synthetic properties expose the protobuf to clients.
A device must be initialized with a `Service`, either obtained using the `discovery`
module or manually created. The only required fields are `ip_addresses` and `port`.
A `Device` object is initially inert. A client must call its `async_run` method to
connect the device, process state changes and handle device updates. A
`asyncio.Future` is returned to monitor and stop the device.
A device has an `available` property which is true when the device is connected and
has received basic properties from the firmware. The `async_wait_available` coroutine
can be used to wait for a device to become available (which may never happen).
If the `uuid` field of the `Service` used to initialize a device is set, the library
will validate it against the device's `dns_sd_uuid` property after connecting to the
device and receiving basic properties. If the UUIDs do not match, the device is
stopped and a `DeviceUUIDMismatchError` exception is set on the run future and raised
in `async_wait_available` coroutines.
To disable periodic properties queries, set `query_interval_seconds` to 0.
Clients can register callbacks to be notified when one or more device properties
have changed. Callbacks are suppressed when no actual changes are observed (i.e.
the device is in a steady state). Callbacks are also be suppressed when only
so-called volatile properties have changed, such as fan RPM, device uptime or the
device's internal clock. This can be disabled by setting `ignore_volatile_props` to
False. These properties are still queried and available to read from the device.
"""
def __init__(
self,
service: Service,
query_interval_seconds: int = 60,
ignore_volatile_props: bool = True,
delay_between_connects_seconds: int = DELAY_BETWEEN_CONNECTS_SECONDS,
):
if len(service.ip_addresses) == 0 or service.port == 0:
raise ValueError(
f"Invalid service: must have at least one address and a port: {service}"
)
self._service = copy.deepcopy(service)
self._query_interval_seconds = query_interval_seconds
self._ignore_volatile_props = ignore_volatile_props
# Permanent Properties protobuf into which query results are merged.
self._properties = aiobafi6_pb2.Properties() # pylint: disable=no-member
# Device update callbacks.
self._callbacks: list[t.Callable[[Device], None]] = []
self._coro_callbacks: list[t.Coroutine] = []
self._dispatch_coro_callback_tasks: t.Set[asyncio.Task] = set()
# Connection and periodic queries.
self._loop = asyncio.get_running_loop()
if self._loop is None:
raise RuntimeError("no running loop")
self._run_fut: t.Optional[asyncio.Future] = None
self._next_connect_ts: float = time.monotonic()
self._connect_timer: t.Optional[asyncio.TimerHandle] = None
self._connect_task: t.Optional[asyncio.Task] = None
self._delay_between_connects_seconds = delay_between_connects_seconds
self._transport: t.Optional[asyncio.Transport] = None
self._protocol: t.Optional[Protocol] = None
self._query_timer: t.Optional[asyncio.TimerHandle] = None
# Availability.
self._available_fut: asyncio.Future = self._loop.create_future()
def __eq__(self, other: t.Any) -> bool:
if isinstance(other, Device):
return self.dns_sd_uuid == t.cast(Device, other).dns_sd_uuid
if isinstance(other, str):
return other == self.name
return NotImplemented
def __hash__(self) -> int:
return hash(self.dns_sd_uuid)
def __str__(self) -> str:
string = f"Name: {self.name}"
string += f", Model: {self.model}"
string += f", DNS SD UUID: {self.dns_sd_uuid}"
string += f", IP: {self.ip_address}"
string += f", MAC: {self.mac_address}"
if self.firmware_version is not None:
string += f", Firmware: {self.firmware_version}"
if self.has_light is not None:
string += f", Has Light: {self.has_light}"
return string
@property
def service(self) -> Service: # pylint: disable=missing-function-docstring
return copy.deepcopy(self._service)
@property
def properties_dict(self) -> dict[str, t.Any]:
"""Return a dict created by merging the device's service and properties."""
propsd = {
"dns_sd_uuid": self._service.uuid,
"service_name": self._service.service_name,
"name": self._service.device_name,
"model": self._service.model,
"api_version": self._service.api_version,
"ip_addresses": self._service.ip_addresses,
"port": self._service.port,
}
propsd.update(
json_format.MessageToDict(
t.cast(Message, self._properties), preserving_proto_field_name=True
)
)
return propsd
@property
def properties_proto( # pylint: disable=missing-function-docstring
self,
) -> aiobafi6_pb2.Properties:
props = aiobafi6_pb2.Properties() # pylint: disable=no-member
props.CopyFrom(self._properties)
return props
# Client callbacks
def add_callback(self, callback: t.Callable[[Device], None]) -> None:
"""Add a device update callback.
The callback must be a `Callable` with a `Device` argument.
"""
is_coroutine = inspect.iscoroutinefunction(callback)
if is_coroutine:
if callback not in self._coro_callbacks:
self._coro_callbacks.append(callback) # type: ignore
_LOGGER.debug("%s: Added coroutine callback.", self.name)
return
if callback not in self._callbacks:
self._callbacks.append(callback)
_LOGGER.debug("%s: Added function callback.", self.name)
def remove_callback(self, callback) -> None:
"""Remove a device update callback."""
if callback in self._coro_callbacks:
self._coro_callbacks.remove(callback)
_LOGGER.debug("%s: Removed coroutine callback.", self.name)
return
if callback in self._callbacks:
self._callbacks.remove(callback)
_LOGGER.debug("%s: Removed function callback.", self.name)
return
def _dispatch_callbacks(self) -> None:
"""Dispatch registered device update callbacks.
An async task is created for coroutine callbacks. Function callbacks are
executed synchronously inside a try-except block to swallow any error."""
for callback in self._callbacks:
try:
callback(self)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Exception raised during callback.")
for coro in self._coro_callbacks:
task = asyncio.create_task(coro(self)) # type: ignore
self._dispatch_coro_callback_tasks.add(task)
task.add_done_callback(self._dispatch_coro_callback_tasks.remove)
_LOGGER.debug(
"%s: Dispatched %s client callbacks.",
self.name,
len(self._callbacks) + len(self._coro_callbacks),
)
# protoprop support
def _maybe_property(self, field: str) -> t.Optional[t.Any]:
return maybe_proto_field(t.cast(Message, self._properties), field)
def _commit_property(self, prop: aiobafi6_pb2.Properties) -> None:
"""Commit a property to the device.
This does not update the properties reflected by the `Device` object. That will
happen once the device confirms the change by doing a properties push.
Is it unknown if the firmware generally supports writing more than one property
in one transaction.
"""
if self._transport is None:
_LOGGER.warning(
"%s: Dropping property commit because device is not connected: %s",
self.name,
prop,
)
return
root = aiobafi6_pb2.Root() # pylint: disable=no-member
root.root2.commit.properties.CopyFrom(prop)
_LOGGER.debug("%s: Sending commit:\n%s", self.name, root)
self._transport.write(wireutils.serialize(root))
# Connection and query machinery
def _sched_connect_or_reset(self):
"""Schedule a `_connect` invocation or reset the device to be run again.
This function is the entrypoint of the internal run state machine. It is called
when there is no connection (`_start`), when a connection could not be
established (error or timeout) or the connection has been closed, or when the
device is stopped (`_stop`)."""
assert self._connect_timer is None
assert self._connect_task is None
assert self._query_timer is None
assert self._transport is None
assert self._protocol is None
assert self._run_fut
# If the run future is done, then reset it to None and return.
# `_sched_connect_or_reset` is the entrypoint of the internal run state machine
# and therefore this is the right place to make the device runnable again (by
# clearing the run future).
if self._run_fut.done():
_LOGGER.debug("%s: Resetting device for new run.", self.name)
self._run_fut = None
return
_LOGGER.debug("%s: Scheduling next connect invocation.", self.name)
self._connect_timer = self._loop.call_at(
self._next_connect_ts,
self._connect,
)
def _connect(self) -> None:
self._connect_timer = None
self._next_connect_ts = time.monotonic() + self._delay_between_connects_seconds
_LOGGER.debug(
"%s: Connecting to %s:%s.",
self.name,
self._service.ip_addresses[0],
self._service.port,
)
connect_task = asyncio.create_task(
self._loop.create_connection(
lambda: Protocol(self),
self._service.ip_addresses[0],
self._service.port,
)
)
connect_task.add_done_callback(self._finish_connect)
self._loop.call_later(self._delay_between_connects_seconds, connect_task.cancel)
self._connect_task = connect_task
def _finish_connect(self, task: asyncio.Task) -> None:
assert self._connect_task is task
self._connect_task = None
try:
transport, protocol = task.result()
_LOGGER.debug(
"%s: Connected to %s.", self.name, transport.get_extra_info("peername")
)
self._transport = transport
self._protocol = protocol
self._loop.call_soon(self._query)
except (OSError, asyncio.CancelledError) as err:
_LOGGER.debug("%s: Connection failed: %s", self.name, err)
self._sched_connect_or_reset()
def _handle_connection_lost(self, exc: t.Optional[Exception]) -> None:
_LOGGER.debug("%s: Connection lost: %s", self.name, exc)
if self._query_timer is not None:
self._query_timer.cancel()
self._query_timer = None
self._transport = None
self._protocol = None
self._sched_connect_or_reset()
def _process_message(self, data: bytes) -> None:
root = aiobafi6_pb2.Root() # pylint: disable=no-member
root.ParseFromString(data)
_LOGGER.debug("%s: Received message: %s", self.name, root)
# Discard unknown fields because `MergeFrom` treats them as repeated.
root.DiscardUnknownFields() # type: ignore
previous = self.properties_proto
for prop in root.root2.query_result.properties:
self._properties.MergeFrom(prop)
if not self.available:
self._maybe_set_available()
current = self.properties_proto
if self._ignore_volatile_props:
_clear_volatile_props(previous)
_clear_volatile_props(current)
if self.available and current != previous:
self._dispatch_callbacks()
def _maybe_set_available(self):
"""Set the device as available if all required properties are set."""
for pname in _PROPS_REQUIRED_FOR_AVAILABLE:
if not self._properties.HasField(pname):
return
if self._service.uuid is not None and self._service.uuid != self.dns_sd_uuid:
_LOGGER.error(
"%s: Device UUID (%s) does not match service UUID (%s): stopping.",
self.name,
self.dns_sd_uuid,
self._service.uuid,
)
assert self._run_fut
if not self._run_fut.done():
self._run_fut.set_exception(DeviceUUIDMismatchError)
return
_LOGGER.debug("%s: Setting device as available.", self.name)
self._available_fut.set_result(True)
def _query(self) -> None:
self._query_timer = None
# The first `_query` of a connection is scheduled with `call_soon` and can't
# be cancelled, so it's possible (though unlikely) for `_transport` to be None.
# If that's the case, just bail out.
if self._transport is None:
return
root = aiobafi6_pb2.Root() # pylint: disable=no-member
root.root2.query.property_query = aiobafi6_pb2.ALL # pylint: disable=no-member
_LOGGER.debug("%s: Sending query:\n%s", self.name, root)
self._transport.write(wireutils.serialize(root))
if self._query_interval_seconds > 0:
self._query_timer = self._loop.call_later(
self._query_interval_seconds, self._query
)
def async_run(self) -> asyncio.Future:
"""Run the device asynchronously.
A running `Device` schedules functions on the run loop to maintain a connection
to the device, sends periodic property queries, and services query commits.
Returns a future that will resolve when the device stops. Cancelling this future
will stop the device.
"""
if self._run_fut is None:
self._start()
assert self._run_fut is not None
return self._run_fut
def _start(self):
"""Start the device.
This function schedules the device to connect on the next run loop iteration.
From there on, the device will continue scheduling functions to maintain the
connection, send periodic property queries, and service query commits.
"""
assert self._run_fut is None
_LOGGER.debug("%s: Starting.", self.name)
self._run_fut = self._loop.create_future()
def stop_on_done(_: asyncio.Future):
self._stop()
self._run_fut.add_done_callback(stop_on_done)
self._sched_connect_or_reset()
def _stop(self) -> None:
"""Stop the device.
This function ultimately causes `_sched_connect_or_reset` to be called by
cancelling the appropriate in-flight task.
This function also creates a new available future, thus marking the device as
unavailable. If the run future became done because of an exception or because
it was cancelled, that is propagated to the prior available future. Otherwise,
the prior available future is never signalled."""
_LOGGER.debug("%s: Stopping.", self.name)
# Propagate run exception or cancellation to the available future, then reset it
# to set the device as unavailable.
assert self._run_fut
if self._run_fut.cancelled():
self._available_fut.cancel()
else:
run_exc = self._run_fut.exception()
if run_exc is not None:
self._available_fut.set_exception(run_exc)
self._available_fut = self._loop.create_future()
# Dispatch client callbacks, since some clients may observe the `available`
# property through a callback.
self._dispatch_callbacks()
# If there is an active connection, close it.
if self._transport is not None:
self._transport.close()
# Otherwise, if the device is opening a connection, cancel that.
elif self._connect_task is not None:
self._connect_task.cancel()
# Otherwise, if `_connect` is scheduled, cancel that and call
# `_sched_connect_or_reset` directly because nothing else will.
elif self._connect_timer is not None:
self._connect_timer.cancel()
self._connect_timer = None
self._sched_connect_or_reset()
# Availability
@property
def available(self) -> bool:
"""Return True when device is running and has values for critical properties."""
available_fut = self._available_fut
return available_fut.done() and not available_fut.exception()
async def async_wait_available(self) -> None:
"""Asynchronously wait for the device to be available."""
await self._available_fut
# General
@property
def name(self) -> str: # pylint: disable=missing-function-docstring
if len(self._properties.name) > 0:
return self._properties.name
if (
self._service.service_name is not None
and len(self._service.service_name) > 0
):
return self._service.service_name
if len(self._properties.mac_address) > 0:
return self._properties.mac_address
return self._service.ip_addresses[0]
@property
def model(self) -> t.Optional[str]: # pylint: disable=missing-function-docstring
if len(self._properties.model) > 0:
return self._properties.model
return self._service.model
firmware_version = ProtoProp[t.Optional[str]]()
mac_address = ProtoProp[t.Optional[str]]()
# API
@property
def dns_sd_uuid( # pylint: disable=missing-function-docstring
self,
) -> t.Optional[str]:
if len(self._properties.dns_sd_uuid) > 0:
return self._properties.dns_sd_uuid
return self._service.uuid
@property
def api_version( # pylint: disable=missing-function-docstring
self,
) -> t.Optional[str]:
if len(self._properties.api_version) > 0:
return self._properties.api_version
return self._service.api_version
@property
def has_fan(self) -> bool: # pylint: disable=missing-function-docstring
# TODO(#1): Support light-only devices.
return True
@property
def has_light( # pylint: disable=missing-function-docstring
self,
) -> t.Optional[bool]:
return maybe_proto_field(self._properties.capabilities, "has_light")
@property
def has_auto_comfort(self) -> bool: # pylint: disable=missing-function-docstring
# https://github.com/home-assistant/core/issues/72934
hc1 = maybe_proto_field(self._properties.capabilities, "has_comfort1") or False
hc3 = maybe_proto_field(self._properties.capabilities, "has_comfort3") or False
return hc1 and hc3
@property
def has_occupancy(self) -> bool: # pylint: disable=missing-function-docstring
try:
api_version = int(self.api_version or 0)
except ValueError:
api_version = 0
# There is probably a capability flag for this but it is unknown. Speculatively,
# a device that supports auto comfort is assumed to supports occupancy.
return api_version >= OCCUPANCY_MIN_API_VERSION and self.has_auto_comfort
# Fan
# pylint: disable=unnecessary-lambda
fan_mode = ProtoProp[OffOnAuto](writable=True, from_proto=lambda v: OffOnAuto(v))
reverse_enable = ProtoProp[bool](writable=True)
speed_percent = ProtoProp[int]()
speed = ProtoProp[int](
writable=True, to_proto=ClosedIntervalValidator[int](0, _MAX_SPEED)
)
whoosh_enable = ProtoProp[bool](writable=True)
eco_enable = ProtoProp[bool](writable=True)
auto_comfort_enable = ProtoProp[bool](writable=True)
comfort_ideal_temperature = ProtoProp[float](
writable=True,
to_proto=to_proto_temperature,
from_proto=from_proto_temperature,
)
comfort_heat_assist_enable = ProtoProp[bool](writable=True)
comfort_heat_assist_speed = ProtoProp[int](writable=True)
comfort_heat_assist_reverse_enable = ProtoProp[bool](writable=True)
comfort_min_speed = ProtoProp[int](
writable=True, to_proto=ClosedIntervalValidator[int](0, _MAX_SPEED)
)
comfort_max_speed = ProtoProp[int](
writable=True, to_proto=ClosedIntervalValidator[int](0, _MAX_SPEED)
)
motion_sense_enable = ProtoProp[bool](writable=True)
motion_sense_timeout = ProtoProp[int](writable=True)
return_to_auto_enable = ProtoProp[bool](writable=True)
return_to_auto_timeout = ProtoProp[int](writable=True)
target_rpm = ProtoProp[int]()
current_rpm = ProtoProp[int]()
fan_occupancy_detected = ProtoProp[bool](min_api_version=OCCUPANCY_MIN_API_VERSION)
# Light
# pylint: disable=unnecessary-lambda
light_mode = ProtoProp[OffOnAuto](writable=True, from_proto=lambda v: OffOnAuto(v))
light_brightness_percent = ProtoProp[int](writable=True)
light_brightness_level = ProtoProp[int](writable=True)
light_color_temperature = ProtoProp[int](writable=True)
light_dim_to_warm_enable = ProtoProp[bool](writable=True)
light_auto_motion_timeout = ProtoProp[int](writable=True)
light_return_to_auto_enable = ProtoProp[bool](writable=True)
light_return_to_auto_timeout = ProtoProp[int](writable=True)
light_warmest_color_temperature = ProtoProp[int]()
light_coolest_color_temperature = ProtoProp[int]()
light_occupancy_detected = ProtoProp[bool](
min_api_version=OCCUPANCY_MIN_API_VERSION
)
# Sensors
temperature = ProtoProp[float](
to_proto=to_proto_temperature,
from_proto=from_proto_temperature,
)
humidity = ProtoProp[int](from_proto=from_proto_humidity)
# Connectivity
@property
def ip_address(self) -> str: # pylint: disable=missing-function-docstring
if len(self._properties.ip_address) > 0:
return self._properties.ip_address
return self._service.ip_addresses[0]
@property
def wifi_ssid( # pylint: disable=missing-function-docstring
self,
) -> t.Optional[str]:
return maybe_proto_field(self._properties.wifi, "ssid")
# More
led_indicators_enable = ProtoProp[bool](writable=True)
fan_beep_enable = ProtoProp[bool](writable=True)
legacy_ir_remote_enable = ProtoProp[bool](writable=True)
class Protocol(asyncio.Protocol):
"""AsyncIO Protocol for BAF i6."""
__slots__ = ("_device", "_transport", "_buffer")
def __init__(self, device: Device):
self._device = device
self._transport: t.Optional[asyncio.Transport] = None
self._buffer = bytearray()
def connection_made(self, transport: asyncio.Transport) -> None:
self._transport = transport
def connection_lost(self, exc: t.Optional[Exception]) -> None:
self._device._handle_connection_lost(exc) # pylint: disable=protected-access
self._transport = None
def data_received(self, data: bytes) -> None:
assert self._transport is not None
if len(self._buffer) + len(data) > _RECV_BUFFER_LIMIT:
raise RuntimeError("Exceeded buffering limit.")
self._buffer.extend(data)
while len(self._buffer) > 1:
if self._buffer[0] != 0xC0:
_LOGGER.error("Receive buffer does not start with sync byte.")
self._transport.abort()
break
end = self._buffer.find(0xC0, 1)
if end == -1:
break
if end == 1:
_LOGGER.error("Empty message found in receive buffer.")
self._transport.abort()
break
self._device._process_message( # pylint: disable=protected-access
wireutils.remove_emulation_prevention(self._buffer[1:end])
)
self._buffer = self._buffer[end + 1 :]