Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
events = self._get_events(timeout)
# Check for downlink
for event in events:
# Format: ,[,][,],[,<data>]
event_items = event.split(',')
status = event_items.pop(0)
status = self._int(status)
if status == EventCode.RECV_DATA:
self._add_downlink(event_items)
# Check for errors
for event in events:
status = event.split(',')[0]
status = self._int(status)
if status not in (EventCode.RECV_DATA,
EventCode.TX_COMFIRMED,
EventCode.TX_UNCOMFIRMED):
raise Rak811EventError(status)
</data>
Process event queue looking for incoming (downlink) messages. Raise
errors when unexpected events are encountered.
Parameter:
timeout: maximum time to wait for event
"""
events = self._get_events(timeout)
# Check for downlink
for event in events:
# Format: ,[,][,],[,<data>]
event_items = event.split(',')
status = event_items.pop(0)
status = self._int(status)
if status == EventCode.RECV_DATA:
self._add_downlink(event_items)
# Check for errors
for event in events:
status = event.split(',')[0]
status = self._int(status)
if status not in (EventCode.RECV_DATA,
EventCode.TX_COMFIRMED,
EventCode.TX_UNCOMFIRMED):
raise Rak811EventError(status)
</data>
WAKE_UP = 8
P2PTX_COMPLETE = 9
UNKNOWN = 100
EVENT_MESSAGE = {
EventCode.RECV_DATA: 'Received data',
EventCode.TX_COMFIRMED: 'Tx confirmed',
EventCode.TX_UNCOMFIRMED: 'Tx unconfirmed',
EventCode.JOINED_SUCCESS: 'Join succeded',
EventCode.JOINED_FAILED: 'Join failed',
EventCode.TX_TIMEOUT: 'Tx timeout',
EventCode.RX2_TIMEOUT: 'Rx2 timeout',
EventCode.DOWNLINK_REPEATED: 'Downlink repeated',
EventCode.WAKE_UP: 'Wake up',
EventCode.P2PTX_COMPLETE: 'P2P tx complete',
EventCode.UNKNOWN: 'Unknown',
}
class Mode(IntEnum):
"""Module operation mode (LoRaWan or Peer to Peer)."""
LoRaWan = 0
LoRaP2P = 1
class Reset(IntEnum):
"""Module reset type."""
Module = 0
LoRa = 1
RX2_TIMEOUT = 6
DOWNLINK_REPEATED = 7
WAKE_UP = 8
P2PTX_COMPLETE = 9
UNKNOWN = 100
EVENT_MESSAGE = {
EventCode.RECV_DATA: 'Received data',
EventCode.TX_COMFIRMED: 'Tx confirmed',
EventCode.TX_UNCOMFIRMED: 'Tx unconfirmed',
EventCode.JOINED_SUCCESS: 'Join succeded',
EventCode.JOINED_FAILED: 'Join failed',
EventCode.TX_TIMEOUT: 'Tx timeout',
EventCode.RX2_TIMEOUT: 'Rx2 timeout',
EventCode.DOWNLINK_REPEATED: 'Downlink repeated',
EventCode.WAKE_UP: 'Wake up',
EventCode.P2PTX_COMPLETE: 'P2P tx complete',
EventCode.UNKNOWN: 'Unknown',
}
class Mode(IntEnum):
"""Module operation mode (LoRaWan or Peer to Peer)."""
LoRaWan = 0
LoRaP2P = 1
class Reset(IntEnum):
"""Module reset type."""
TX_UNCOMFIRMED = 2
JOINED_SUCCESS = 3
JOINED_FAILED = 4
TX_TIMEOUT = 5
RX2_TIMEOUT = 6
DOWNLINK_REPEATED = 7
WAKE_UP = 8
P2PTX_COMPLETE = 9
UNKNOWN = 100
EVENT_MESSAGE = {
EventCode.RECV_DATA: 'Received data',
EventCode.TX_COMFIRMED: 'Tx confirmed',
EventCode.TX_UNCOMFIRMED: 'Tx unconfirmed',
EventCode.JOINED_SUCCESS: 'Join succeded',
EventCode.JOINED_FAILED: 'Join failed',
EventCode.TX_TIMEOUT: 'Tx timeout',
EventCode.RX2_TIMEOUT: 'Rx2 timeout',
EventCode.DOWNLINK_REPEATED: 'Downlink repeated',
EventCode.WAKE_UP: 'Wake up',
EventCode.P2PTX_COMPLETE: 'P2P tx complete',
EventCode.UNKNOWN: 'Unknown',
}
class Mode(IntEnum):
"""Module operation mode (LoRaWan or Peer to Peer)."""
LoRaWan = 0
LoRaP2P = 1
def __init__(self, status):
"""Just assign return status."""
try:
self.errno = int(status)
except ValueError:
self.errno = status
if self.errno in EVENT_MESSAGE:
self.strerror = EVENT_MESSAGE[self.errno]
else:
self.strerror = EVENT_MESSAGE[EventCode.UNKNOWN]
super().__init__(('[Errno {}] {}').format(self.errno, self.strerror))
DOWNLINK_REPEATED = 7
WAKE_UP = 8
P2PTX_COMPLETE = 9
UNKNOWN = 100
EVENT_MESSAGE = {
EventCode.RECV_DATA: 'Received data',
EventCode.TX_COMFIRMED: 'Tx confirmed',
EventCode.TX_UNCOMFIRMED: 'Tx unconfirmed',
EventCode.JOINED_SUCCESS: 'Join succeded',
EventCode.JOINED_FAILED: 'Join failed',
EventCode.TX_TIMEOUT: 'Tx timeout',
EventCode.RX2_TIMEOUT: 'Rx2 timeout',
EventCode.DOWNLINK_REPEATED: 'Downlink repeated',
EventCode.WAKE_UP: 'Wake up',
EventCode.P2PTX_COMPLETE: 'P2P tx complete',
EventCode.UNKNOWN: 'Unknown',
}
class Mode(IntEnum):
"""Module operation mode (LoRaWan or Peer to Peer)."""
LoRaWan = 0
LoRaP2P = 1
class Reset(IntEnum):
"""Module reset type."""
Module = 0
RECV_DATA = 0
TX_COMFIRMED = 1
TX_UNCOMFIRMED = 2
JOINED_SUCCESS = 3
JOINED_FAILED = 4
TX_TIMEOUT = 5
RX2_TIMEOUT = 6
DOWNLINK_REPEATED = 7
WAKE_UP = 8
P2PTX_COMPLETE = 9
UNKNOWN = 100
EVENT_MESSAGE = {
EventCode.RECV_DATA: 'Received data',
EventCode.TX_COMFIRMED: 'Tx confirmed',
EventCode.TX_UNCOMFIRMED: 'Tx unconfirmed',
EventCode.JOINED_SUCCESS: 'Join succeded',
EventCode.JOINED_FAILED: 'Join failed',
EventCode.TX_TIMEOUT: 'Tx timeout',
EventCode.RX2_TIMEOUT: 'Rx2 timeout',
EventCode.DOWNLINK_REPEATED: 'Downlink repeated',
EventCode.WAKE_UP: 'Wake up',
EventCode.P2PTX_COMPLETE: 'P2P tx complete',
EventCode.UNKNOWN: 'Unknown',
}
class Mode(IntEnum):
"""Module operation mode (LoRaWan or Peer to Peer)."""
LoRaWan = 0
JOINED_FAILED = 4
TX_TIMEOUT = 5
RX2_TIMEOUT = 6
DOWNLINK_REPEATED = 7
WAKE_UP = 8
P2PTX_COMPLETE = 9
UNKNOWN = 100
EVENT_MESSAGE = {
EventCode.RECV_DATA: 'Received data',
EventCode.TX_COMFIRMED: 'Tx confirmed',
EventCode.TX_UNCOMFIRMED: 'Tx unconfirmed',
EventCode.JOINED_SUCCESS: 'Join succeded',
EventCode.JOINED_FAILED: 'Join failed',
EventCode.TX_TIMEOUT: 'Tx timeout',
EventCode.RX2_TIMEOUT: 'Rx2 timeout',
EventCode.DOWNLINK_REPEATED: 'Downlink repeated',
EventCode.WAKE_UP: 'Wake up',
EventCode.P2PTX_COMPLETE: 'P2P tx complete',
EventCode.UNKNOWN: 'Unknown',
}
class Mode(IntEnum):
"""Module operation mode (LoRaWan or Peer to Peer)."""
LoRaWan = 0
LoRaP2P = 1
class Reset(IntEnum):
TX_TIMEOUT = 5
RX2_TIMEOUT = 6
DOWNLINK_REPEATED = 7
WAKE_UP = 8
P2PTX_COMPLETE = 9
UNKNOWN = 100
EVENT_MESSAGE = {
EventCode.RECV_DATA: 'Received data',
EventCode.TX_COMFIRMED: 'Tx confirmed',
EventCode.TX_UNCOMFIRMED: 'Tx unconfirmed',
EventCode.JOINED_SUCCESS: 'Join succeded',
EventCode.JOINED_FAILED: 'Join failed',
EventCode.TX_TIMEOUT: 'Tx timeout',
EventCode.RX2_TIMEOUT: 'Rx2 timeout',
EventCode.DOWNLINK_REPEATED: 'Downlink repeated',
EventCode.WAKE_UP: 'Wake up',
EventCode.P2PTX_COMPLETE: 'P2P tx complete',
EventCode.UNKNOWN: 'Unknown',
}
class Mode(IntEnum):
"""Module operation mode (LoRaWan or Peer to Peer)."""
LoRaWan = 0
LoRaP2P = 1
class Reset(IntEnum):
"""Module reset type."""