Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def as_arrow(self):
value = self.v()
if not value:
return obj.NoneObject("")
try:
# Return a data time object in UTC.
return arrow.Arrow.utcfromtimestamp(self.v())
except (ValueError, TypeError ,OSError) as e:
return obj.NoneObject("Datetime conversion failure: " + str(e))
self._consecutive_errors = 0
except KeyError as e:
# Unknown, Unsupported, or vendor-specific ME. Key is the unknown classID
# TODO: Can we create a temporary one to hold it so upload does not always fail on new ME's?
self.log.exception('frame-decode-key-error', msg=hexlify(msg), e=e)
return
except Exception as e:
self.log.exception('frame-decode', msg=hexlify(msg), e=e)
return
try:
(ts, d, _, _) = self._requests.pop(rx_tid)
ts_diff = now - arrow.Arrow.utcfromtimestamp(ts)
secs = ts_diff.total_seconds()
self._reply_sum += secs
if secs < self._reply_min:
self._reply_min = secs
if secs > self._reply_max:
self._reply_max = secs
# TODO: Could also validate response type based on request action
except KeyError:
# Possible late Rx on a message that timed-out
self._rx_unknown_tid += 1
self.log.warn('tx-message-missing', rx_id=rx_tid, msg=hexlify(msg))
return
def _parse_date(self, date):
"""Returns Arrow object from timestamp."""
return arrow.Arrow.utcfromtimestamp(date).to('local')
def _update_rx_tx_stats(self, now, ts):
ts_diff = now - arrow.Arrow.utcfromtimestamp(ts)
secs = ts_diff.total_seconds()
self._reply_sum += secs
if secs < self._reply_min:
self._reply_min = secs
if secs > self._reply_max:
self._reply_max = secs
return secs
def dc_utc_timestamp_to_dt(dc_timestamp_in_milleseconds):
"""Return a UTC datetime object"""
return arrow.Arrow.utcfromtimestamp(dc_timestamp_in_milleseconds / 1000).datetime