bckelley/tconnectsync
1
0
Fork 0
mirror of https://github.com/bckelley/tconnectsync.git synced 2026-08-24 03:34:12 -05:00
Code Issues Packages Projects Releases Wiki Activity

revert nightscout.py

Browse Source
This commit is contained in:
Beshoy Girgis
2026-03-27 11:08:23 -07:00
committed by James Woglom
parent bf894bce02
commit f0f94baf02
1 changed files with 82 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tconnectsync/nightscout.py
+54 -14
View File
@@ -12,8 +12,7 @@ from .api.common import ApiException
from .parser.nightscout import ENTERED_BY
def format_datetime(date):
# Convert to UTC and format as ISO-8601 with Z suffix (Nightscout expects UTC)
return arrow.get(date).to('UTC').replace(microsecond=0).isoformat().replace('+00:00', 'Z')
return arrow.get(date).isoformat()
def time_range(field_name, start_time, end_time, t_to_space=False):
def fmt(date):
@@ -66,21 +65,37 @@ class NightscoutApi:
raise ApiException(r.status_code, "Nightscout put %s response: %s" % (r.status_code, r.text))
def last_uploaded_entry(self, eventType, time_start=None, time_end=None):
dateFilter = time_range('created_at', time_start, time_end, t_to_space=False)
def internal(t_to_space):
dateFilter = time_range('created_at', time_start, time_end, t_to_space=t_to_space)
latest = requests.get(urljoin(self.url, 'api/v1/treatments?count=1&find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[eventType]=' + urllib.parse.quote(eventType) + dateFilter + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
}, verify=self.verify)
if latest.status_code != 200:
if 'as a valid ISO-8601 date' in latest.text:
logger.warning("Nightscout last_uploaded_entry %s could not process ISO-8601 date: start=%s end=%s dateFilter=%s response=%s" % (eventType, time_start, time_end, dateFilter, latest.text))
logger.warning("Nightscout last_uploaded_entry %s could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (eventType, time_start, time_end, dateFilter))
return None
raise ApiException(latest.status_code, "Nightscout last_uploaded_entry %s response: %s" % (latest.status_code, latest.text))
try:
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
try:
ret = None
try:
ret = internal(False)
except ApiException as e:
#logger.warning("last_uploaded_entry with no t_to_space: %s", e)
ret = None
if ret is None and (time_start or time_end):
try:
ret = internal(True)
except ApiException as e:
#logger.warning("last_uploaded_entry with t_to_space: %s", e)
ret = None
if ret is not None:
logger.warning("last_uploaded_entry with eventType=%s time_start=%s time_end=%s only returned data when timestamps contained a space" % (eventType, time_start, time_end))
return ret
except requests.exceptions.ConnectionError as e:
if self.ignore_conn_errors:
logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e)
@@ -88,21 +103,30 @@ class NightscoutApi:
raise e
def last_uploaded_bg_entry(self, time_start=None, time_end=None):
dateFilter = time_range('dateString', time_start, time_end, t_to_space=False)
def internal(t_to_space):
dateFilter = time_range('dateString', time_start, time_end, t_to_space=t_to_space)
latest = requests.get(urljoin(self.url, 'api/v1/entries.json?count=1&find[device]=' + urllib.parse.quote(ENTERED_BY) + dateFilter + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
}, verify=self.verify)
if latest.status_code != 200:
if 'as a valid ISO-8601 date' in latest.text:
logger.warning("Nightscout last_uploaded_bg_entry could not process ISO-8601 date: start=%s end=%s dateFilter=%s response=%s" % (time_start, time_end, dateFilter, latest.text))
logger.warning("Nightscout last_uploaded_bg_entry could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (time_start, time_end, dateFilter))
return None
raise ApiException(latest.status_code, "Nightscout last_uploaded_bg_entry %s response: %s" % (latest.status_code, latest.text))
try:
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
try:
ret = internal(False)
if ret is None and (time_start or time_end):
ret = internal(True)
if ret is not None:
logger.warning("last_uploaded_bg_entry with time_start=%s time_end=%s only returned data when timestamps contained a space" % (time_start, time_end))
return ret
except requests.exceptions.ConnectionError as e:
if self.ignore_conn_errors:
logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e)
@@ -110,21 +134,29 @@ class NightscoutApi:
raise e
def last_uploaded_activity(self, activityType, time_start=None, time_end=None):
dateFilter = time_range('created_at', time_start, time_end, t_to_space=False)
def internal(t_to_space):
dateFilter = time_range('created_at', time_start, time_end, t_to_space=t_to_space)
latest = requests.get(urljoin(self.url, 'api/v1/activity?find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[activityType]=' + urllib.parse.quote(activityType) + dateFilter + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
}, verify=self.verify)
if latest.status_code != 200:
if 'as a valid ISO-8601 date' in latest.text:
logger.warning("Nightscout activity %s could not process ISO-8601 date: start=%s end=%s dateFilter=%s response=%s" % (activityType, time_start, time_end, dateFilter, latest.text))
logger.warning("Nightscout activity %s could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (activityType, time_start, time_end, dateFilter))
return None
raise ApiException(latest.status_code, "Nightscout activity %s response: %s" % (latest.status_code, latest.text))
try:
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
try:
ret = internal(False)
if ret is None and (time_start or time_end):
ret = internal(True)
if ret is not None:
logger.warning("last_uploaded_activity with activityType=%s time_start=%s time_end=%s only returned data when timestamps contained a space" % (activityType, time_start, time_end))
return ret
except requests.exceptions.ConnectionError as e:
if self.ignore_conn_errors:
logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e)
@@ -132,21 +164,29 @@ class NightscoutApi:
raise e
def last_uploaded_devicestatus(self, time_start=None, time_end=None):
dateFilter = time_range('created_at', time_start, time_end, t_to_space=False)
def internal(t_to_space):
dateFilter = time_range('created_at', time_start, time_end, t_to_space=t_to_space)
latest = requests.get(urljoin(self.url, 'api/v1/devicestatus?find[device]=' + urllib.parse.quote(ENTERED_BY) + dateFilter + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
}, verify=self.verify)
if latest.status_code != 200:
if 'as a valid ISO-8601 date' in latest.text:
logger.warning("Nightscout devicestatus could not process ISO-8601 date: start=%s end=%s dateFilter=%s response=%s" % (time_start, time_end, dateFilter, latest.text))
logger.warning("Nightscout devicestatus could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (time_start, time_end, dateFilter))
return None
raise ApiException(latest.status_code, "Nightscout devicestatus %s response: %s" % (latest.status_code, latest.text))
try:
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
try:
ret = internal(False)
if ret is None and (time_start or time_end):
ret = internal(True)
if ret is not None:
logger.warning("devicestatus time_start=%s time_end=%s only returned data when timestamps contained a space" % (time_start, time_end))
return ret
except requests.exceptions.ConnectionError as e:
if self.ignore_conn_errors:
logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e)