Compare commits

..
56 Commits
Author SHA1 Message Date
James Woglom 90bede8b75 bump version to v0.6.2 2021-12-23 20:57:50 -05:00
James Woglom 7a8d00da88 secret: put environment variables at higher precedence than .env file 2021-12-23 20:46:58 -05:00
James Woglom 52cb287ce8 bump version to 0.6.1 2021-12-23 20:38:23 -05:00
James Woglom 27ba5d9a64 Update dotenv secret parsing logic, and read from .config/tconnectsync/.env path 2021-12-23 19:39:51 -05:00
James WoglomandGitHub 2140bd356e Update README.md with note about forking the repo 2021-12-15 15:31:33 -05:00
James Woglom a74b63ab05 bugfix: check for extended bolus start/completion time presence before parsing, since extended boluses can be in progress 2021-12-12 13:29:08 -05:00
James Woglom b0ffe9bcfc update to 1.6 2021-12-04 22:00:56 -05:00
James Woglom ca4ca72476 add SKIP_NS_LAST_UPLOADED_CHECK workaround when running tconnectsync for past data. see #23 2021-12-04 21:56:30 -05:00
James Woglom 8cef3233c6 features: enable PUMP_EVENTS as optional feature 2021-12-04 21:38:13 -05:00
James Woglom 118ee236ba test_process: add WS2 activity event process tests 2021-12-04 21:37:34 -05:00
James Woglom 1657b1c696 test_process: add test coverage for CIQ pump_events sync 2021-12-04 21:20:47 -05:00
James Woglom 0a9dd8732f bug: use deepcopy to avoid mutating the base dictionary 2021-12-04 21:20:13 -05:00
James Woglom dbb1389254 hack: enable debug logging in tests 2021-12-04 21:18:53 -05:00
James Woglom 163db9807b sync/pump_events: upload events to nightscout 2021-12-04 20:28:03 -05:00
James Woglom b1ca7eee1e add basalsuspension parsing + tests 2021-12-04 19:02:13 -05:00
James Woglom 1fa543e3d6 add PUMP_EVENTS feature, with initial code for parsing CIQ activities (sleep mode, exercise, etc) 2021-12-04 17:41:16 -05:00
James Woglom 037f53e8c4 version: update to 0.5, due to suspension event changes 2021-10-28 01:43:11 -04:00
James Woglom 9635b0bd0f sync/basal: process manual suspension events as manual 0u/hr basal, fixes #17 2021-10-28 01:38:11 -04:00
James Woglom e9f569d2a0 workflows: only publish to pypi on dispatch 2021-10-24 00:41:04 -04:00
James Woglom f5ef34174f in-progress: support BOLUS_BG and CGM features.
BOLUS_BG will add BG readings which are associated with boluses
on the pump into the Nightscout treatment object. It will determine
whether the BG reading was automatically filled via the Dexcom
connection on the pump or was manually entered by seeing if the
BG reading matches the current CGM reading as known to the pump
at that time. Support for this is nearly complete.

CGM will add Dexcom CGM readings from the pump to Nightscout
as SGV (sensor glucose value) entries. This should only be
used in a situation where xDrip is not used and the pump
connection to the CGM will be the only source of CGM
data to Nightscout. This requires additional testing before
it should be considered ready.

Both options are hidden behind the ENABLE_TESTING_MODES=true
environment variable. Run tconnectsync with this environment
variable and BOLUS_BG and CGM will be usable with the
--features argument.
2021-10-24 00:36:48 -04:00
James Woglom 9f488b2c5e support specifying specific features to synchronize (basal, bolus, iob) 2021-10-23 16:31:59 -04:00
James WoglomandGitHub 99341c5443 README: use GitHub Registry instead of Docker Hub 2021-10-20 01:40:20 -04:00
James Woglom a2df266e93 setup.cfg: bump to v0.4.1 2021-10-20 01:30:35 -04:00
James Woglom b53fb9c099 README: update with pip installation mechanism, and clarify installation steps 2021-10-20 01:28:43 -04:00
James Woglom bcd486561f dockerfile: fix build 2021-10-20 01:01:47 -04:00
James Woglom d931242562 setup.cfg: set version to 0.4.0 2021-10-20 00:51:26 -04:00
James Woglom bec1cef1f1 dockerfile: attempt to fix build 2021-10-20 00:48:46 -04:00
James Woglom 82ec1b4c1b pypi: add github action 2021-10-20 00:43:28 -04:00
James Woglom b6569edabc configure console_script for tconnectsync, move main code to inside package 2021-10-20 00:34:58 -04:00
James Woglom 2e5d8fc793 add setup.cfg/setup.py/pyproject.toml 2021-10-20 00:15:23 -04:00
James Woglom c33a674bad iob: add e2e tests for IOB behavior 2021-10-14 02:02:57 -04:00
James Woglom 3be0289177 bolus: add tests for partial entries, propagate requested_insulin 2021-10-14 02:02:57 -04:00
James Woglom 87f531cbb3 tests: add iob test, fix sleep increment in ws2 tests 2021-10-14 02:02:57 -04:00
James WoglomandGitHub 795b5ce240 Merge pull request #15 from policetonyR/patch-1
Remove PatientObjectId
2021-09-10 20:36:02 -04:00
TonyandGitHub 3a004183bb Remove PatientObjectId
After trying to run auto update, I was met with an error saying that patientObjectId was not found, I commented the code out and the auto update ran. I propose that this line is removed as I do not see any use for the variable anywhere.
2021-09-09 15:06:29 -04:00
James Woglom 7d813ee8fb main: update logging config to show timestamps 2021-07-09 21:49:34 -04:00
James Woglom d0366cc064 sync/bolus: fix bug when synchronizing extended bolus 2021-07-09 21:30:23 -04:00
James Woglom f2f7b1c0e1 api: add basalsuspension, pumpfeatures api endpoints 2021-07-09 21:29:06 -04:00
James Woglom 855e50358a bugfix: fix TypeError if last_process_time_range is None, fix printf 2021-05-06 13:30:26 -04:00
James Woglom 4afb35e235 bugfix: autoupdate 2021-05-04 02:03:36 -04:00
James Woglom 655e66a8aa bug: fix string format 2021-05-04 02:00:27 -04:00
James Woglom 380e4f00b6 parser/nightscout: add rate field for temp basals for autotune compatibility (fixes #7) 2021-05-03 23:59:30 -04:00
James Woglom f9ecaedf4d tests/controliq: fix fake access token expiration in tests. remove warns for warnings 2021-05-03 23:59:30 -04:00
James Woglom faf54b1f47 autoupdate: add new config options if no data received, log pretty timestamps 2021-05-03 23:59:30 -04:00
James Woglom 1733b0e139 logging: use logger instead of prints, improve logging throughout 2021-05-03 23:59:30 -04:00
James Woglom acfe44881b add .codecov.yml 2021-04-26 23:14:58 -04:00
James Woglom 3fc9cdeea3 add .codecov.yml 2021-04-26 23:14:43 -04:00
James Woglom 33df18d4c9 api/android: add tests for last_event_uploaded and retry behavior 2021-04-26 23:07:32 -04:00
James Woglom 05c19fffa7 api/controliq: add login tests 2021-04-26 22:50:33 -04:00
James Woglom 878cd5b247 api/controliq: retry/relogin on http errors, add tests 2021-04-26 22:50:33 -04:00
James WoglomandGitHub 9c8e5c150c Add codecov badge 2021-04-24 23:36:38 -04:00
James WoglomandGitHub 01d9517ff4 Add codecov 2021-04-24 23:24:01 -04:00
James Woglom e7816c4aa0 tests/api: test ws2 therapy_timeline_csv parsing 2021-04-24 23:12:36 -04:00
James Woglom 18f1149b28 tests: add ws2 api tests for retries 2021-04-24 22:48:45 -04:00
James Woglom fe941c432f api/ws2: add retry logic for http 500s in therapy_timeline_csv (fix #9) 2021-04-24 22:27:07 -04:00
James WoglomandGitHub 71ceb94270 Merge pull request #8 from jwoglom/develop
version 0.2.0
2021-04-21 01:43:43 -04:00
37 changed files with 2724 additions and 260 deletions
+33
View File
@@ -0,0 +1,33 @@
name: Publish to PyPI
on: workflow_dispatch
jobs:
build-binary:
name: Build Binary and Publish to PyPI
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@master
- name: Set up Python 3.9
uses: actions/setup-python@v1
with:
python-version: 3.9
- name: Install pypa/build
run: >-
python -m
pip install
build
--user
- name: Build a binary wheel and a source tarball
run: >-
python -m
build
--sdist
--wheel
--outdir dist/
.
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@master
with:
password: ${{ secrets.PYPI_API_TOKEN }}
+3
View File
@@ -1,6 +1,9 @@
__pycache__
dist
build
*.pyc
*.pyo
*.swp
*.swa
*.egg-info
.env
+8 -2
View File
@@ -15,16 +15,22 @@ FROM base AS python-deps
RUN pip install pipenv
RUN apt-get update && apt-get install -y --no-install-recommends gcc
RUN mkdir -p /base
WORKDIR /base
# Install python dependencies in /.venv
COPY Pipfile .
COPY Pipfile.lock .
COPY setup.cfg .
COPY setup.py .
COPY pyproject.toml .
RUN PIPENV_VENV_IN_PROJECT=1 pipenv install --deploy
FROM base AS runtime
# Copy virtualenv from python-deps stage
COPY --from=python-deps /.venv /.venv
ENV PATH="/.venv/bin:$PATH"
COPY --from=python-deps /base/.venv /base/.venv
ENV PATH="/base/.venv/bin:$PATH"
# Create and switch to a new user
RUN useradd --create-home appuser
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2021 James Woglom
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+2
View File
@@ -4,8 +4,10 @@ url = "https://pypi.org/simple"
verify_ssl = true
[dev-packages]
ptpython = "*"
[packages]
tconnectsync = {editable = true, path = "."}
requests = "*"
bs4 = "*"
arrow = "*"
Generated
+111 -40
View File
@@ -1,7 +1,7 @@
{
"_meta": {
"hash": {
"sha256": "979feca3aba1b7f43890a94862347738158af5847959c14dc1d5f23ce5db8cdc"
"sha256": "413f8bbd66e702656262deda1b6586d42b5a2e9fa52f6d23416b1ea1efbc2591"
},
"pipfile-spec": 6,
"requires": {},
@@ -16,19 +16,19 @@
"default": {
"arrow": {
"hashes": [
"sha256:3515630f11a15c61dcb4cdd245883270dd334c83f3e639824e65a4b79cc48543",
"sha256:399c9c8ae732270e1aa58ead835a79a40d7be8aa109c579898eb41029b5a231d"
"sha256:16fc29bbd9e425e3eb0fef3018297910a0f4568f21116fc31771e2760a50e074",
"sha256:8fb7d9d3d4bf90e49e734c22fa077bdd0964135c4b8120de2510575a8d1f620c"
],
"index": "pypi",
"version": "==1.0.3"
"version": "==1.2.0"
},
"beautifulsoup4": {
"hashes": [
"sha256:4c98143716ef1cb40bf7f39a8e3eec8f8b009509e74904ba3a7b315431577e35",
"sha256:84729e322ad1d5b4d25f805bfa05b902dd96450f43842c4e99067d5e1369eb25",
"sha256:fff47e031e34ec82bf17e00da8f592fe7de69aeea38be00523c04623c04fb666"
"sha256:9a315ce70049920ea4572a4055bc4bd700c940521d36fc858205ad4fcde149bf",
"sha256:c23ad23c521d818955a4151a67d81580319d4bf548d3d49f4223ae041ff98891"
],
"version": "==4.9.3"
"markers": "python_version >= '3.1'",
"version": "==4.10.0"
},
"bs4": {
"hashes": [
@@ -39,44 +39,51 @@
},
"certifi": {
"hashes": [
"sha256:1a4995114262bffbc2413b159f2a1a480c969de6e6eb13ee966d470af86af59c",
"sha256:719a74fb9e33b9bd44cc7f3a8d94bc35e4049deebe19ba7d8e108280cfd59830"
"sha256:78884e7c1d4b00ce3cea67b44566851c4343c120abd683433ce934a68ea58872",
"sha256:d62a0163eb4c2344ac042ab2bdf75399a71a2d8c7d47eac2e2ee91b9d6339569"
],
"version": "==2020.12.5"
"version": "==2021.10.8"
},
"chardet": {
"charset-normalizer": {
"hashes": [
"sha256:0d6f53a15db4120f2b08c94f11e7d93d2c911ee118b6b30a04ec3ee8310179fa",
"sha256:f864054d66fd9118f2e67044ac8981a54775ec5b67aed0441892edb553d21da5"
"sha256:e019de665e2bcf9c2b64e2e5aa025fa991da8720daa3c1138cadd2fd1856aed0",
"sha256:f7af805c321bfa1ce6714c51f254e0d5bb5e5834039bc17db7ebe3a4cec9492b"
],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'",
"version": "==4.0.0"
"markers": "python_version >= '3'",
"version": "==2.0.7"
},
"idna": {
"hashes": [
"sha256:b307872f855b18632ce0c21c5e45be78c0ea7ae4c15c828c20788b26921eb3f6",
"sha256:b97d804b1e9b523befed77c48dacec60e6dcb0b5391d57af6a65a312a90648c0"
"sha256:84d9dd047ffa80596e0f246e2eab0b391788b0503584e8945f2368256d2735ff",
"sha256:9d643ff0a55b762d5cdb124b8eaa99c66322e2157b69160bc32796e824360e6d"
],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
"version": "==2.10"
"markers": "python_version >= '3'",
"version": "==3.3"
},
"lxml": {
"hashes": [
"sha256:079f3ae844f38982d156efce585bc540c16a926d4436712cf4baee0cce487a3d",
"sha256:0fbcf5565ac01dff87cbfc0ff323515c823081c5777a9fc7703ff58388c258c3",
"sha256:122fba10466c7bd4178b07dba427aa516286b846b2cbd6f6169141917283aae2",
"sha256:1b38116b6e628118dea5b2186ee6820ab138dbb1e24a13e478490c7db2f326ae",
"sha256:1b7584d421d254ab86d4f0b13ec662a9014397678a7c4265a02a6d7c2b18a75f",
"sha256:26e761ab5b07adf5f555ee82fb4bfc35bf93750499c6c7614bd64d12aaa67927",
"sha256:289e9ca1a9287f08daaf796d96e06cb2bc2958891d7911ac7cae1c5f9e1e0ee3",
"sha256:2a9d50e69aac3ebee695424f7dbd7b8c6d6eb7de2a2eb6b0f6c7db6aa41e02b7",
"sha256:3082c518be8e97324390614dacd041bb1358c882d77108ca1957ba47738d9d59",
"sha256:33bb934a044cf32157c12bfcfbb6649807da20aa92c062ef51903415c704704f",
"sha256:3439c71103ef0e904ea0a1901611863e51f50b5cd5e8654a151740fde5e1cade",
"sha256:36108c73739985979bf302006527cf8a20515ce444ba916281d1c43938b8bb96",
"sha256:39b78571b3b30645ac77b95f7c69d1bffc4cf8c3b157c435a34da72e78c82468",
"sha256:4289728b5e2000a4ad4ab8da6e1db2e093c63c08bdc0414799ee776a3f78da4b",
"sha256:4bff24dfeea62f2e56f5bab929b4428ae6caba2d1eea0c2d6eb618e30a71e6d4",
"sha256:4c61b3a0db43a1607d6264166b230438f85bfed02e8cff20c22e564d0faff354",
"sha256:542d454665a3e277f76954418124d67516c5f88e51a900365ed54a9806122b83",
"sha256:5a0a14e264069c03e46f926be0d8919f4105c1623d620e7ec0e612a2e9bf1c04",
"sha256:5c8c163396cc0df3fd151b927e74f6e4acd67160d6c33304e805b84293351d16",
"sha256:64812391546a18896adaa86c77c59a4998f33c24788cadc35789e55b727a37f4",
"sha256:66e575c62792c3f9ca47cb8b6fab9e35bab91360c783d1606f758761810c9791",
"sha256:6f12e1427285008fd32a6025e38e977d44d6382cf28e7201ed10d6c1698d2a9a",
"sha256:74f7d8d439b18fa4c385f3f5dfd11144bb87c1da034a466c5b5577d23a1d9b51",
"sha256:7610b8c31688f0b1be0ef882889817939490a36d0ee880ea562a4e1399c447a1",
"sha256:76fa7b1362d19f8fbd3e75fe2fb7c79359b0af8747e6f7141c338f0bee2f871a",
@@ -89,10 +96,15 @@
"sha256:b007cbb845b28db4fb8b6a5cdcbf65bacb16a8bd328b53cbc0698688a68e1caa",
"sha256:bc4313cbeb0e7a416a488d72f9680fffffc645f8a838bd2193809881c67dd106",
"sha256:bccbfc27563652de7dc9bdc595cb25e90b59c5f8e23e806ed0fd623755b6565d",
"sha256:c1a40c06fd5ba37ad39caa0b3144eb3772e813b5fb5b084198a985431c2f1e8d",
"sha256:c47ff7e0a36d4efac9fd692cfa33fbd0636674c102e9e8d9b26e1b93a94e7617",
"sha256:c4f05c5a7c49d2fb70223d0d5bcfbe474cf928310ac9fa6a7c6dddc831d0b1d4",
"sha256:cdaf11d2bd275bf391b5308f86731e5194a21af45fbaaaf1d9e8147b9160ea92",
"sha256:ce256aaa50f6cc9a649c51be3cd4ff142d67295bfc4f490c9134d0f9f6d58ef0",
"sha256:d2e35d7bf1c1ac8c538f88d26b396e73dd81440d59c1ef8522e1ea77b345ede4",
"sha256:d916d31fd85b2f78c76400d625076d9124de3e4bda8b016d25a050cc7d603f24",
"sha256:df7c53783a46febb0e70f6b05df2ba104610f2fb0d27023409734a3ecbb78fb2",
"sha256:e1cbd3f19a61e27e011e02f9600837b921ac661f0c40560eefb366e4e4fb275e",
"sha256:efac139c3f0bf4f0939f9375af4b02c5ad83a622de52d6dfa8e438e8e01d0eb0",
"sha256:efd7a09678fd8b53117f6bae4fa3825e0a22b03ef0a932e070c0bdbb3a35e654",
"sha256:f2380a6376dfa090227b663f9678150ef27543483055cc327555fb592c5967e2",
@@ -104,60 +116,119 @@
},
"python-dateutil": {
"hashes": [
"sha256:73ebfe9dbf22e832286dafa60473e4cd239f8592f699aa5adaf10050e6e1823c",
"sha256:75bb3f31ea686f1197762692a9ee6a7550b59fc6ca3a1f4b5d7e32fb98e2da2a"
"sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86",
"sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9"
],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
"version": "==2.8.1"
"version": "==2.8.2"
},
"python-dotenv": {
"hashes": [
"sha256:471b782da0af10da1a80341e8438fca5fadeba2881c54360d5fd8d03d03a4f4a",
"sha256:49782a97c9d641e8a09ae1d9af0856cc587c8d2474919342d5104d85be9890b2"
"sha256:14f8185cc8d494662683e6914addcb7e95374771e707601dfc70166946b4c4b8",
"sha256:bbd3da593fc49c249397cbfbcc449cf36cb02e75afc8157fcc6a81df6fb7750a"
],
"index": "pypi",
"version": "==0.17.0"
"version": "==0.19.1"
},
"requests": {
"hashes": [
"sha256:27973dd4a904a4f13b263a19c866c13b92a39ed1c964655f025f3f8d3d75b804",
"sha256:c210084e36a42ae6b9219e00e48287def368a26d03a048ddad7bfee44f75871e"
"sha256:6c1246513ecd5ecd4528a0906f910e8f0f9c6b8ec72030dc9fd154dc1a6efd24",
"sha256:b8aa58f8cf793ffd8782d3d8cb19e66ef36f7aba4353eec859e74678b01b07a7"
],
"index": "pypi",
"version": "==2.25.1"
"version": "==2.26.0"
},
"requests-mock": {
"hashes": [
"sha256:11215c6f4df72702aa357f205cf1e537cffd7392b3e787b58239bde5fb3db53b",
"sha256:e68f46844e4cee9d447150343c9ae875f99fa8037c6dcf5f15bf1fe9ab43d226"
"sha256:0a2d38a117c08bb78939ec163522976ad59a6b7fdd82b709e23bb98004a44970",
"sha256:8d72abe54546c1fc9696fa1516672f1031d72a55a1d66c85184f972a24ba0eba"
],
"index": "pypi",
"version": "==1.8.0"
"version": "==1.9.3"
},
"six": {
"hashes": [
"sha256:30639c035cdb23534cd4aa2dd52c3bf48f06e5f4a941509c8bafd8ce11080259",
"sha256:8b74bedcbbbaca38ff6d7491d76f2b06b3592611af620f8426e82dddb04a5ced"
"sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926",
"sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254"
],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
"version": "==1.15.0"
"version": "==1.16.0"
},
"soupsieve": {
"hashes": [
"sha256:052774848f448cf19c7e959adf5566904d525f33a3f8b6ba6f6f8f26ec7de0cc",
"sha256:c2c1c2d44f158cdbddab7824a9af8c4f83c76b1e23e049479aa432feb6c4c23b"
],
"markers": "python_version >= '3.0'",
"markers": "python_version >= '3.6'",
"version": "==2.2.1"
},
"tconnectsync": {
"editable": true,
"path": "."
},
"urllib3": {
"hashes": [
"sha256:2f4da4594db7e1e110a944bb1b551fdf4e6c136ad42e4234131391e21eb5b0df",
"sha256:e7b021f7241115872f92f43c6508082facffbd1c048e3c6e2bb9c2a157e28937"
"sha256:4987c65554f7a2dbf30c18fd48778ef124af6fab771a377103da0585e2336ece",
"sha256:c4fdf4019605b6e5423637e01bc9fe4daef873709a7973e195ceba0a62bbc844"
],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4' and python_version < '4'",
"version": "==1.26.4"
"version": "==1.26.7"
}
},
"develop": {}
"develop": {
"appdirs": {
"hashes": [
"sha256:7d5d0167b2b1ba821647616af46a749d1c653740dd0d2415100fe26e27afdf41",
"sha256:a841dacd6b99318a741b166adb07e19ee71a274450e68237b4650ca1055ab128"
],
"version": "==1.4.4"
},
"jedi": {
"hashes": [
"sha256:18456d83f65f400ab0c2d3319e48520420ef43b23a086fdc05dff34132f0fb93",
"sha256:92550a404bad8afed881a137ec9a461fed49eca661414be45059329614ed0707"
],
"markers": "python_version >= '3.6'",
"version": "==0.18.0"
},
"parso": {
"hashes": [
"sha256:12b83492c6239ce32ff5eed6d3639d6a536170723c6f3f1506869f1ace413398",
"sha256:a8c4922db71e4fdb90e0d0bc6e50f9b273d3397925e5e60a717e719201778d22"
],
"markers": "python_version >= '3.6'",
"version": "==0.8.2"
},
"prompt-toolkit": {
"hashes": [
"sha256:6076e46efae19b1e0ca1ec003ed37a933dc94b4d20f486235d436e64771dcd5c",
"sha256:eb71d5a6b72ce6db177af4a7d4d7085b99756bf656d98ffcc4fecd36850eea6c"
],
"markers": "python_full_version >= '3.6.2'",
"version": "==3.0.20"
},
"ptpython": {
"hashes": [
"sha256:99636899ab0e4d026e2ecc9368269114f387b4bb5411e57f072b0bde724d9f99",
"sha256:eafd4ced27ca5dc370881d4358d1ab5041b32d88d31af8e3c24167fe4af64ed6"
],
"index": "pypi",
"version": "==3.0.20"
},
"pygments": {
"hashes": [
"sha256:b8e67fe6af78f492b3c4b3e2970c0624cbf08beb1e493b2c99b9fa1b67a20380",
"sha256:f398865f7eb6874156579fdf36bc840a03cab64d1cde9e93d68f46a425ec52c6"
],
"markers": "python_version >= '3.5'",
"version": "==2.10.0"
},
"wcwidth": {
"hashes": [
"sha256:beb4802a9cebb9144e99086eff703a642a13d6a0052920003a230f3294bbe784",
"sha256:c4d647b99872929fdb7bdcaa4fbe7f01413ed3d98077df798530e5b04f116c83"
],
"version": "==0.2.5"
}
}
}
+243 -44
View File
@@ -9,19 +9,52 @@ If you have a t:slim X2 pump with the companion t:connect mobile Android or iOS
At a high level, tconnectsync works by querying Tandem's undocumented APIs to receive basal and bolus data from t:connect, and then uploads that data as treatment objects to Nightscout. It contains features for checking for new Tandem pump data continuously, and updating that data along with the pump's reported IOB value to Nightscout whenever there is new data.
**To get started,** read the setup instructions below and choose whether to run the application via **Pipenv** or **Docker**.
When you run the program with no arguments, it performs a single cycle of the following, and exits after completion:
## Tandem APIs
* Queries for basal information via the t:connect ControlIQ API.
* Queries for bolus, basal, and IOB data via the t:connect non-ControlIQ API.
* Merges the basal information received from the two APIs. (If using ControlIQ, then basal information appears only on the ControlIQ API. If not using ControlIQ, it appears only on the legacy API.)
* Queries Nightscout for the most recently created Temp Basal object by tconnectsync, and uploads all data newer than that.
* Queries Nightscout for the most recently created Bolus object by tconnectsync, and uploads all data newer than that.
* Uploads a single Nightscout Activity object representing the current IOB as reported by the pump.
This application utilizes three separate Tandem APIs for obtaining t:connect data, referenced here by the identifying part of their URLs:
If run with the `--auto-update` flag, then the application performs the following steps:
* [**controliq**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/controliq.py) - Contains Control:IQ related data, namely a timeline of all Basal events uploaded by the pump, separated by type (temp basals, algorithmically-updated basals, or profile-updated basals).
* [**android**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/android.py) - Used internally by the t:connect Android app, these API endpoints were discovered by reverse-engineering the Android app. Most of the API endpoints are used for uploading pump data, and tconnectsync uses one endpoint which returns the most recent event ID uploaded by the pump, so we know when more data has been uploaded.
* [**tconnectws2**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/ws2.py) - More legacy than the others, this seems to power the bulk of the main t:connect website. It is used to retrieve a CSV export of non-ControlIQ basal data, as well as bolus and IOB data. (I haven't found any mentions of bolus or IOB data in the Control:IQ-specific API.)
* Queries an API endpoint used only by the t:connect mobile app which returns an internal event ID, corresponding to the most recent event published by the mobile app.
* Whenever the internal event ID changes (denoting that the mobile app uploaded new data to synchronize), perform all of the above mentioned steps to synchronize data.
The following synchronization features are enabled by default:
* `BASAL`: Basal data
* `BOLUS`: Bolus data
* `IOB`: Insulin-on-board data. Only the most recent IOB entry is saved to Nightscout, as an "activity"
The following synchronization feature is disabled by default, but can be enabled via the `--features` flag:
* `PUMP_EVENTS`: Events reported by the pump. Includes support for the following:
* Site/Cartridge Change (occurs for both a site change and a cartridge change)
* Empty Cartridge/Pump Shutdown (from my investigation, occurs either when the cartridge runs out of insulin OR you hard-shut off the pump)
* User Suspended (occurs when you manually disable insulin delivery)
* Exercise Mode (in Nightscout, appears with a start and end time)
* Sleep Mode (in Nightscout, appears with a start and end time)
## Setup
Create a file named `.env` containing configuration values inside the checked-out tconnectsync folder (the same folder as `main.py`). You should specify the following parameters:
**To get started,** you need to choose whether to install the application via
**Pip**, **Pipenv**, or **Docker**.
After that, you can choose to run the program continuously via **Supervisord**
or on a regular interval with **Cron**.
**NOTE:** If you fork the tconnectsync repository on GitHub, **do not commit your .env file**.
If pushed to GitHub, this will make your tconnect and Nightscout passwords publicly visible and put your data at risk.
## Installation
First, you need to create a file containing configuration values.
The name of this file will be `.env`, and its location will be dependent on which
method of installation you choose.
You should specify the following parameters:
```bash
# Your credentials for t:connect
@@ -39,79 +72,170 @@ NS_SECRET='apisecret'
TIMEZONE_NAME='America/New_York'
```
These values can alternatively be specified via environment variables.
This file contains your t:connect username and password, Tandem pump serial number (which is utilized in API calls to t:connect), your Nightscout URL and secret token (for uploading data to Nightscout), and local timezone (the timezone used in t:connect).
The .env file contains your t:connect username and password, Tandem pump serial number (which is utilized in API calls to t:connect), your Nightscout URL and secret token (for uploading data to Nightscout), and local timezone (the timezone used in t:connect).
(Alternatively, these values can be specified via environment variables.)
I have only tested tconnectsync with a Tandem pump set in the US Eastern timezone. Tandem's (to us, undocumented) APIs are [a bit loose with timezones](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/parser.py#L15), so please let me know if you notice any timezone-related bugs.
### Installation via Pip
When you run the program with no arguments, it performs a single cycle of the following, and exits after completion:
This is the easiest method to install.
* Queries for basal information via the t:connect ControlIQ API.
* Queries for bolus, basal, and IOB data via the t:connect non-ControlIQ API.
* Merges the basal information received from the two APIs. (If using ControlIQ, then basal information appears only on the ControlIQ API. If not using ControlIQ, it appears only on the legacy API.)
* Queries Nightscout for the most recently created Temp Basal object by tconnectsync, and uploads all data newer than that.
* Queries Nightscout for the most recently created Bolus object by tconnectsync, and uploads all data newer than that.
* Uploads a single Nightscout Activity object representing the current IOB as reported by the pump.
First, ensure that you have **Python 3** with **Pip** installed:
If run with the `--auto-update` flag, then the application performs the following steps:
* **On MacOS:** Open Terminal. Install [Homebrew](https://brew.sh/), and then run `brew install python3`
* **On Linux:** Follow your distribution's specific instructions.
For Debian/Ubuntu based distros, `sudo apt install python3 python3-pip`
* **On Windows:** Install Ubuntu under the [Windows Subsystem for Linux](https://ubuntu.com/wsl).
Open the Ubuntu Terminal, then run `sudo apt install python3 python3-pip`.
Perform the remainder of the steps under the Ubuntu environment.
* Queries an API endpoint used only by the t:connect mobile app which returns an internal event ID, corresponding to the most recent event published by the mobile app.
* Whenever the internal event ID changes (denoting that the mobile app uploaded new data to synchronize), perform all of the above mentioned steps to synchronize data.
Now install the `tconnectsync` package with pip:
### Running with Pipenv
```
$ pip3 install tconnectsync
```
You can run the application using Pipenv. Assuming you have only Python 3 and pip installed, install pipenv with `pip3 install pipenv`. Then install tconnectsync's dependencies with `pipenv install`, and you can launch the program with `pipenv run tconnectsync` (which, through an alias defined in `Pipfile`, runs ``pipenv run python3 main.py`).
If the pip3 command is not found, run `python3 -m pip install tconnectsync` instead.
After this, you should be able to view tconnectsync's help with:
```
$ tconnectsync --help
usage: tconnectsync [-h] [--version] [--pretend] [-v] [--start-date START_DATE] [--end-date END_DATE] [--days DAYS] [--auto-update] [--check-login]
[--features {BASAL,BOLUS,IOB,PUMP_EVENTS} [{BASAL,BOLUS,IOB,PUMP_EVENTS} ...]]
Syncs bolus, basal, and IOB data from Tandem Diabetes t:connect to Nightscout.
optional arguments:
-h, --help show this help message and exit
--version show program's version number and exit
--pretend Pretend mode: do not upload any data to Nightscout.
-v, --verbose Verbose mode: show extra logging details
--start-date START_DATE
The oldest date to process data from. Must be specified with --end-date.
--end-date END_DATE The newest date to process data until (inclusive). Must be specified with --start-date.
--days DAYS The number of days of t:connect data to read in. Cannot be used with --from-date and --until-date.
--auto-update If set, continuously checks for updates from t:connect and syncs with Nightscout.
--check-login If set, checks that the provided t:connect credentials can be used to log in.
--features {BASAL,BOLUS,IOB,PUMP_EVENTS} [{BASAL,BOLUS,IOB,PUMP_EVENTS} ...]
Specifies what data should be synchronized between tconnect and Nightscout.
```
Move the `.env` file you created to the following folder:
* **MacOS:** `/Users/<username>/.config/tconnectsync/.env`
* **Linux:** `$HOME/.config/tconnectsync/.env`
* **Windows:** `$HOME/.config/tconnectsync/.env` (inside WSL)
```
$ tconnectsync --check-login
```
If you receive no errors, then you can move on to the **Running Tconnectsync Continuously** section.
### Installing with Pipenv
You can run the application using Pipenv.
First, ensure you have Python 3 and pip installed, then install pipenv with `pip3 install pipenv`.
Clone the Git repository for tconnectsync and cd into it with:
```
$ git clone https://github.com/jwoglom/tconnectsync
$ cd tconnectsync
```
Then install tconnectsync's dependencies with `pipenv install`.
Afterwards, you can launch the program with `pipenv run tconnectsync` so long as
you are inside the checked-out tconnectsync folder.
```bash
$ git clone https://github.com/jwoglom/tconnectsync && cd tconnectsync
$ pip3 install pipenv
$ pipenv install
$ pipenv run tconnectsync --help
usage: main.py [-h] [--pretend] [--start-date START_DATE] [--end-date END_DATE]
[--days DAYS] [--auto-update]
usage: main.py [-h] [--version] [--pretend] [-v] [--start-date START_DATE] [--end-date END_DATE] [--days DAYS] [--auto-update] [--check-login]
[--features {BASAL,BOLUS,IOB,PUMP_EVENTS} [{BASAL,BOLUS,IOB,PUMP_EVENTS} ...]]
Syncs bolus, basal, and IOB data from Tandem Diabetes t:connect to Nightscout.
optional arguments:
-h, --help show this help message and exit
--version show program's version number and exit
--pretend Pretend mode: do not upload any data to Nightscout.
-v, --verbose Verbose mode: show extra logging details
--start-date START_DATE
The oldest date to process data from. Must be specified with
--end-date.
--end-date END_DATE The newest date to process data until (inclusive). Must be
specified with --start-date.
--days DAYS The number of days of t:connect data to read in. Cannot be
used with --from-date and --until-date.
--auto-update If set, continuously checks for updates from t:connect and
syncs with Nightscout.
The oldest date to process data from. Must be specified with --end-date.
--end-date END_DATE The newest date to process data until (inclusive). Must be specified with --start-date.
--days DAYS The number of days of t:connect data to read in. Cannot be used with --from-date and --until-date.
--auto-update If set, continuously checks for updates from t:connect and syncs with Nightscout.
--check-login If set, checks that the provided t:connect credentials can be used to log in.
--features {BASAL,BOLUS,IOB,PUMP_EVENTS} [{BASAL,BOLUS,IOB,PUMP_EVENTS} ...]
Specifies what data should be synchronized between tconnect and Nightscout.
```
You can now continue to either the **Running with Cron** or **Running with Supervisord** sections.
### Running with Docker
Move the `.env` file you created earlier into the `tconnectsync` folder, and run:
```
$ pipenv run tconnectsync --check-login
```
If you receive no errors, then you can move on to the **Running Tconnectsync Continuously** section.
### Installing with Docker
First, [ensure that you have Docker running and installed](https://docs.docker.com/get-started/#download-and-install-docker).
To download and run the `jwoglom/tconnectsync` prebuilt Docker image from [Docker Hub](https://hub.docker.com/r/jwoglom/tconnectsync):
To download and run the prebuilt Docker image from GitHub Packages:
```bash
$ docker pull jwoglom/tconnectsync:latest
$ docker run jwoglom/tconnectsync --help
$ docker pull ghcr.io/jwoglom/tconnectsync/tconnectsync:latest
$ docker run ghcr.io/jwoglom/tconnectsync/tconnectsync --help
```
Move the `.env` file you created earlier into the current folder, and run:
```
$ docker run tconnectsync --check-login
```
If you receive no errors, then you can move on to the **Running Tconnectsync Continuously** section.
#### Building Locally
To instead build the image locally and launch the project:
```bash
$ git clone https://github.com/jwoglom/tconnectsync
$ cd tconnectsync
$ docker build -t tconnectsync .
$ docker run tconnectsync --help
```
You can now continue to either the **Running with Cron** or **Running with Supervisord** sections.
Move the `.env` file you created earlier into this folder, and run:
```
$ docker run tconnectsync --check-login
```
If you receive no errors, then you can move on to the **Running Tconnectsync Continuously** section.
## Running Tconnectsync Continuously
You most likely want tconnectsync to run either continuously (via the auto-update
feature) or on a regular interval (via cron).
The supervisord approach is recommended for simplicity.
### Running with Supervisord (recommended)
To instead configure tconnectsync to run continuously in the background using its `--auto-update` feature, you can use a tool such as Supervisord. Here is an example `tconnectsync.conf` which you can place inside `/etc/supervisor/conf.d`:
To configure tconnectsync to run continuously in the background using its `--auto-update` feature, you can use a tool such as Supervisord.
First, install supervisord via your Linux system's package manager.
(For example, for Ubuntu/Debian-based systems, run `sudo apt install supervisor`)
Supervisord is configured by creating a configuration file in `/etc/supervisor/conf.d`.
Here is an example `tconnectsync.conf` which you can place in that folder:
```
[program:tconnectsync]
@@ -119,12 +243,63 @@ command=/path/to/tconnectsync/run.sh
directory=/path/to/tconnectsync/
stderr_logfile=/path/to/tconnectsync/stderr.log
stdout_logfile=/path/to/tconnectsync/stdout.log
user=tconnectsync
user=<your username>
numprocs=1
autostart=true
autorestart=true
```
In order to create a `run.sh` file, see the section below which aligns with your
choice of installation method.
After the configuration file has been created, ensure that Supervisor is running
and configured to start on boot:
```bash
$ sudo systemctl daemon-reload
$ sudo systemctl start supervisord
$ sudo systemctl enable supervisord
```
Then use the `supervisorctl` command to manage the status of the tconnectsync program:
```bash
$ sudo supervisorctl status
tconnectsync STOPPED
$ sudo supervisorctl start tconnectsync
$ sudo supervisorctl status
tconnectsync RUNNING pid 18810, uptime 00:00:05
```
You can look at the `stderr.log` and `stdout.log` files to check that tconnectsync
is running and has started up properly:
```bash
$ tail -f /path/to/tconnectsync/stdout.log
Starting auto-update between 2021-09-30 00:06:39.942273 and 2021-10-01 00:06:39.942273
2021-10-01 00:06:39 DEBUG Instantiating new AndroidApi
2021-10-01 00:06:39 DEBUG Starting new HTTPS connection (1): tdcservices.tandemdiabetes.com:443
2021-10-01 00:06:40 DEBUG https://tdcservices.tandemdiabetes.com:443 "POST /cloud/oauth2/token HTTP/1.1" 200 404
2021-10-01 00:06:40 INFO Logged in to AndroidApi successfully (expiration: 2021-10-01T08:06:40.362Z, in 7 hours, 59 minutes)
```
#### With Pip Installation
In the `tconnectsync.conf`, you should set `/path/to/tconnectsync` to the folder
containing your `.env` file.
Create a `run.sh` file containing:
```bash
#!/bin/bash
tconnectsync --auto-update
```
#### With Pipenv Installation
In the `tconnectsync.conf`, you should set `/path/to/tconnectsync` to the folder
where you checked-out the GitHub repository.
An example `run.sh` which launches tconnectsync within its pipenv-configured virtual environment:
```bash
@@ -139,17 +314,32 @@ cd /path/to/tconnectsync
exec python3 -u main.py --auto-update
```
An example `run.sh` which uses Docker:
#### With Docker Installation
In the `tconnectsync.conf`, you should set `/path/to/tconnectsync` to the folder
where you checked-out the GitHub repository.
An example `run.sh` if you installed tconnectsync via the GitHub Docker Registry:
```bash
#!/bin/bash
docker run ghcr.io/jwoglom/tconnectsync/tconnectsync --auto-update
```
An example `run.sh` if you built tconnectsync locally:
```bash
#!/bin/bash
docker build -t tconnectsync
docker run tconnectsync --auto-update
```
### Running with Cron
To configure tconnectsync to run at a periodic interval (i.e. every 15 minutes), you can just invoke main.py with no arguments via cron.
If you choose not to run tconnectsync with `--auto-update` continuously,
you can instead run it at a periodic interval (i.e. every 15 minutes) by just
invoking tconnectsync with no arguments via cron.
If using Pipenv or a virtualenv, make sure that you either prefix the call to main.py with `pipenv run` or source the `bin/activate` file within the virtualenv, so that the proper dependencies are loaded. If not using any kind of virtualenv, you can instead just install the necessary dependencies as specified inside Pipfile globally.
@@ -160,8 +350,17 @@ An example configuration in `/etc/crontab` which runs every 15 minutes:
0,15,30,45 * * * * root /path/to/tconnectsync/run.sh
```
You can use one of the same `run.sh` files mentioned above in the Supervisord example, but remove the `--auto-update` flag since you are handling the functionality for running the script periodically yourself.
You can use one of the same `run.sh` files referenced above, but remove the `--auto-update` flag since you are handling the functionality for running the script periodically yourself.
## Tandem APIs
This application utilizes three separate Tandem APIs for obtaining t:connect data, referenced here by the identifying part of their URLs:
* [**controliq**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/controliq.py) - Contains Control:IQ related data, namely a timeline of all Basal events uploaded by the pump, separated by type (temp basals, algorithmically-updated basals, or profile-updated basals).
* [**android**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/android.py) - Used internally by the t:connect Android app, these API endpoints were discovered by reverse-engineering the Android app. Most of the API endpoints are used for uploading pump data, and tconnectsync uses one endpoint which returns the most recent event ID uploaded by the pump, so we know when more data has been uploaded.
* [**tconnectws2**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/ws2.py) - More legacy than the others, this seems to power the bulk of the main t:connect website. It is used to retrieve a CSV export of non-ControlIQ basal data, as well as bolus and IOB data. (I haven't found any mentions of bolus or IOB data in the Control:IQ-specific API.)
I have only tested tconnectsync with a Tandem pump set in the US Eastern timezone. Tandem's (to us, undocumented) APIs are [a bit loose with timezones](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/parser.py#L15), so please let me know if you notice any timezone-related bugs.
## Backfilling t:connect Data
To backfill existing t:connect data in to Nightscout, you can use the `--start-date` and `--end-date` options. For example, the following will upload all t:connect data between January 1st and March 1st, 2020 to Nightscout:
+1 -73
View File
@@ -1,77 +1,5 @@
#!/usr/bin/env python3
import sys
import datetime
import arrow
import argparse
import logging
from tconnectsync.api import TConnectApi
from tconnectsync.process import process_time_range
from tconnectsync.autoupdate import process_auto_update
from tconnectsync.check import check_login
from tconnectsync.nightscout import NightscoutApi
try:
from tconnectsync.secret import (
TCONNECT_EMAIL,
TCONNECT_PASSWORD,
NS_URL,
NS_SECRET
)
except Exception:
print('Unable to read secret.py')
sys.exit(1)
def parse_args():
parser = argparse.ArgumentParser(description="Syncs bolus, basal, and IOB data from Tandem Diabetes t:connect to Nightscout.")
parser.add_argument('--pretend', dest='pretend', action='store_const', const=True, default=False, help='Pretend mode: do not upload any data to Nightscout.')
parser.add_argument('-v', '--verbose', dest='verbose', action='store_const', const=True, default=False, help='Verbose mode: show extra logging details')
parser.add_argument('--start-date', dest='start_date', type=str, default=None, help='The oldest date to process data from. Must be specified with --end-date.')
parser.add_argument('--end-date', dest='end_date', type=str, default=None, help='The newest date to process data until (inclusive). Must be specified with --start-date.')
parser.add_argument('--days', dest='days', type=int, default=1, help='The number of days of t:connect data to read in. Cannot be used with --from-date and --until-date.')
parser.add_argument('--auto-update', dest='auto_update', action='store_const', const=True, default=False, help='If set, continuously checks for updates from t:connect and syncs with Nightscout.')
parser.add_argument('--check-login', dest='check_login', action='store_const', const=True, default=False, help='If set, checks that the provided t:connect credentials can be used to log in.')
return parser.parse_args()
def main():
args = parse_args()
if args.verbose:
logging.basicConfig(level=logging.DEBUG)
logging.root.debug("Set logging level to DEBUG")
else:
logging.basicConfig(level=logging.INFO)
if args.auto_update and (args.start_date or args.end_date):
raise Exception('Auto-update cannot be used with start/end date')
if args.start_date and args.end_date:
time_start = arrow.get(args.start_date)
time_end = arrow.get(args.end_date)
else:
time_end = datetime.datetime.now()
time_start = time_end - datetime.timedelta(days=args.days)
if time_end < time_start:
raise Exception('time_start must be before time_end')
tconnect = TConnectApi(TCONNECT_EMAIL, TCONNECT_PASSWORD)
nightscout = NightscoutApi(NS_URL, NS_SECRET)
if args.check_login:
return check_login(tconnect, time_start, time_end)
if args.auto_update:
print("Starting auto-update between", time_start, "and", time_end, "(PRETEND)" if args.pretend else "")
process_auto_update(tconnect, nightscout, time_start, time_end, args.pretend)
else:
print("Processing data between", time_start, "and", time_end, "(PRETEND)" if args.pretend else "")
added = process_time_range(tconnect, nightscout, time_start, time_end, args.pretend)
print("Added", added, "items")
from tconnectsync import main
if __name__ == '__main__':
main()
+6
View File
@@ -0,0 +1,6 @@
[build-system]
requires = [
"setuptools>=42",
"wheel"
]
build-backend = "setuptools.build_meta"
+37
View File
@@ -0,0 +1,37 @@
[metadata]
name = tconnectsync
version = 0.6.2
author = James Woglom
author_email = j@wogloms.net
description = Syncs Tandem t:connect pump data to Nightscout for the t:slim X2
long_description = file: README.md
long_description_content_type = text/markdown
url = https://github.com/jwoglom/tconnectsync
project_urls =
Bug Tracker = https://github.com/jwoglom/tconnectsync/issues
classifiers =
Programming Language :: Python :: 3
License :: OSI Approved :: MIT License
Operating System :: OS Independent
[options]
package_dir =
= .
packages = find:
python_requires = >=3.6
install_requires =
requests
bs4
arrow
lxml
python-dotenv
[options.packages.find]
where = .
exclude =
tests*
scripts*
[options.entry_points]
console_scripts =
tconnectsync = tconnectsync:main
+3
View File
@@ -0,0 +1,3 @@
from setuptools import setup
setup()
+90
View File
@@ -0,0 +1,90 @@
import sys
import datetime
import arrow
import argparse
import logging
import pkg_resources
from .api import TConnectApi
from .process import process_time_range
from .autoupdate import process_auto_update
from .check import check_login
from .nightscout import NightscoutApi
from .features import DEFAULT_FEATURES, ALL_FEATURES
try:
from .secret import (
TCONNECT_EMAIL,
TCONNECT_PASSWORD,
NS_URL,
NS_SECRET
)
except Exception:
print('Unable to read secret.py')
sys.exit(1)
try:
__version__ = pkg_resources.require("tconnectsync")[0].version
except Exception:
__version__ = "UNKNOWN"
def parse_args(*args, **kwargs):
parser = argparse.ArgumentParser(description="Syncs bolus, basal, and IOB data from Tandem Diabetes t:connect to Nightscout.")
parser.add_argument('--version', action='version', version='tconnectsync %s' % __version__)
parser.add_argument('--pretend', dest='pretend', action='store_const', const=True, default=False, help='Pretend mode: do not upload any data to Nightscout.')
parser.add_argument('-v', '--verbose', dest='verbose', action='store_const', const=True, default=False, help='Verbose mode: show extra logging details')
parser.add_argument('--start-date', dest='start_date', type=str, default=None, help='The oldest date to process data from. Must be specified with --end-date.')
parser.add_argument('--end-date', dest='end_date', type=str, default=None, help='The newest date to process data until (inclusive). Must be specified with --start-date.')
parser.add_argument('--days', dest='days', type=int, default=1, help='The number of days of t:connect data to read in. Cannot be used with --from-date and --until-date.')
parser.add_argument('--auto-update', dest='auto_update', action='store_const', const=True, default=False, help='If set, continuously checks for updates from t:connect and syncs with Nightscout.')
parser.add_argument('--check-login', dest='check_login', action='store_const', const=True, default=False, help='If set, checks that the provided t:connect credentials can be used to log in.')
parser.add_argument('--features', dest='features', nargs='+', default=DEFAULT_FEATURES, choices=ALL_FEATURES, help='Specifies what data should be synchronized between tconnect and Nightscout.')
return parser.parse_args(*args, **kwargs)
def main(*args, **kwargs):
args = parse_args(*args, **kwargs)
if args.verbose:
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
logging.root.debug("Set logging level to DEBUG")
else:
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
if args.auto_update and (args.start_date or args.end_date):
raise Exception('Auto-update cannot be used with start/end date')
if args.start_date and args.end_date:
time_start = arrow.get(args.start_date)
time_end = arrow.get(args.end_date)
else:
time_end = datetime.datetime.now()
time_start = time_end - datetime.timedelta(days=args.days)
if time_end < time_start:
raise Exception('time_start must be before time_end')
tconnect = TConnectApi(TCONNECT_EMAIL, TCONNECT_PASSWORD)
nightscout = NightscoutApi(NS_URL, NS_SECRET)
if args.check_login:
return check_login(tconnect, time_start, time_end)
logging.info("Enabled features: " + ", ".join(args.features))
if args.auto_update:
print("Starting auto-update between", time_start, "and", time_end, "(PRETEND)" if args.pretend else "")
process_auto_update(tconnect, nightscout, time_start, time_end, args.pretend, features=args.features)
else:
print("Processing data between", time_start, "and", time_end, "(PRETEND)" if args.pretend else "")
added = process_time_range(tconnect, nightscout, time_start, time_end, args.pretend, features=args.features)
print("Added", added, "items")
+14 -2
View File
@@ -11,7 +11,7 @@ import logging
from bs4 import BeautifulSoup
from ..util import timeago
from .common import ApiException, ApiLoginException
from .common import ApiException, ApiLoginException, parse_date
logger = logging.getLogger(__name__)
@@ -69,7 +69,6 @@ class AndroidApi:
self.refreshToken = j["refreshToken"]
self.refreshTokenExpiresAt = j["refreshTokenExpiresAt"]
self.userId = j["user"]["id"]
self.patientObjectId = j["user"]["patientObjectId"]
logger.info("Logged in to AndroidApi successfully (expiration: %s, %s)" % (self.accessTokenExpiresAt, timeago(self.accessTokenExpiresAt)))
@@ -155,3 +154,16 @@ class AndroidApi:
"""
def user_profile(self):
return self.get('cloud/usersettings/api/UserProfile?userId=%s' % self.userId)
"""
Returns therapy events, used by the webui Therapy Timeline.
{'event': [
{'type': 'Basal', 'basalRate': ...},
{'type': 'Bolus', 'standard': ...},
{'type': 'CGM', 'egv': ...}
]}
"""
def therapy_events(self, start_date=None, end_date=None):
startDate = parse_date(start_date)
endDate = parse_date(end_date)
return self.get('tconnect/therapyevents/api/TherapyEvents/%s/%s/false?userId=%s' % (startDate, endDate, self.userId))
+7
View File
@@ -126,3 +126,10 @@ class ControlIQApi:
"startDate": startDate,
"endDate": endDate
})
"""
Returns active account features, including the date when ControlIQ was enabled.
[{"serialNumber": "11111111", "features": {"controlIQ": {"feature": 1, "dateTimeFirstDetected": "YYYY-MM-DD:THH:MM:SS", "unixTimestamp": 1111111111}}}]
"""
def pumpfeatures(self):
return self.get('pumpfeatures/users/%s' % self.userGuid, {})
+31 -2
View File
@@ -3,6 +3,7 @@ import datetime
import csv
import logging
import time
import json
from .common import parse_date, base_headers, ApiException
@@ -12,6 +13,7 @@ class WS2Api:
BASE_URL = 'https://tconnectws2.tandemdiabetes.com/'
MAX_RETRIES = 2
SLEEP_SECONDS_INCREMENT = 60
userGuid = None
@@ -35,7 +37,7 @@ class WS2Api:
if t.endswith(')'):
t = t[:-1]
return t
return json.loads(t)
def _split_empty_sections(self, text):
sections = [[]]
@@ -60,6 +62,15 @@ class WS2Api:
return data
"""
Returns information on therapy, displayed in the therapy timeline on the
t:connect website.
Contains BG reading (CGM), IOB, basal, and bolus data.
Basal data does NOT appear for the specified time range if using Control-IQ.
The ControlIQ API endpoints must be used for basal data instead.
However, all other fields are still accessed via this endpoint.
"""
def therapy_timeline_csv(self, start=None, end=None, tries=0):
startDate = parse_date(start)
endDate = parse_date(end)
@@ -70,7 +81,7 @@ class WS2Api:
# This seems to occur as some kind of soft rate-limit.
logger.warning("Received ApiException in therapy_timeline_csv: (retry count %d) %s" % (tries, e))
if e.status_code == 500:
sleep_seconds = (tries+1) * 60
sleep_seconds = (tries+1) * self.SLEEP_SECONDS_INCREMENT
logger.error("Retrying in %d seconds after HTTP 500 in therapy_timeline_csv (retry count %d): %s" % (sleep_seconds, tries, e))
time.sleep(sleep_seconds)
if tries < self.MAX_RETRIES:
@@ -103,6 +114,24 @@ class WS2Api:
"basalData": self._csv_to_dict(basalData),
"bolusData": self._csv_to_dict(bolusData)
}
"""
Returns information on basal suspension. The filterbasal option only returns site/cartridge changes.
SuspendReason values are:
- "site-cart"
- "basal-profile"
- "manual"
- "previous"
- "alarm"
{"BasalSuspension":[{"EventDateTime":"/Date(EPOCH_MS-0000)/", "SuspendReason": "reason"}]}
"""
def basalsuspension(self, start=None, end=None, filterbasal=False):
startDate = parse_date(start)
endDate = parse_date(end)
arg = "filterbasal/1" if filterbasal else ""
return self.get_jsonp('basalsuspension/%s/%s/%s/%s' % (self.userGuid, startDate, endDate, arg))
"""
Returns info on BasalIQ in JSONP format.
+13 -10
View File
@@ -1,7 +1,9 @@
import time
import logging
import sys
from .process import process_time_range
from .features import DEFAULT_FEATURES
from .secret import (
PUMP_SERIAL_NUMBER,
AUTOUPDATE_DEFAULT_SLEEP_SECONDS,
@@ -17,7 +19,7 @@ logger = logging.getLogger(__name__)
Performs the auto-update functionality. Runs indefinitely in a loop
until stopped (ctrl+c).
"""
def process_auto_update(tconnect, nightscout, time_start, time_end, pretend):
def process_auto_update(tconnect, nightscout, time_start, time_end, pretend, features=DEFAULT_FEATURES):
# Read from android api, find exact interval to cut down on API calls
# Refresh API token. If failure, die, have wrapper script re-run.
@@ -34,7 +36,7 @@ def process_auto_update(tconnect, nightscout, time_start, time_end, pretend):
if pretend:
logger.info('Would update now if not in pretend mode')
else:
added = process_time_range(tconnect, nightscout, time_start, time_end, pretend)
added = process_time_range(tconnect, nightscout, time_start, time_end, pretend, features=features)
logger.info('Added %d items from process_time_range' % added)
if added == 0:
if last_event_index:
@@ -54,18 +56,19 @@ def process_auto_update(tconnect, nightscout, time_start, time_end, pretend):
logger.info('No new reported t:connect data. (last event index: %s)' % last_event['maxPumpEventIndex'])
now = time.time()
if (now - last_event_time) >= 60 * AUTOUPDATE_FAILURE_MINUTES:
logger.error("No new data event indexes have been detected for over %d minutes. " % AUTOUPDATE_FAILURE_MINUTES +
"The t:connect app might no longer be functioning.")
if last_event_time and (now - last_event_time) >= 60 * AUTOUPDATE_FAILURE_MINUTES:
logger.error(AutoupdateFailureException("No new data event indexes have been detected for over %d minutes. " % AUTOUPDATE_FAILURE_MINUTES +
"The t:connect app might no longer be functioning."))
if AUTOUPDATE_RESTART_ON_FAILURE:
raise AutoupdateFailureException
elif (now - last_process_time_range) >= 60 * AUTOUPDATE_FAILURE_MINUTES:
logger.error("No new data has been found via the API for over %d minutes. " % AUTOUPDATE_FAILURE_MINUTES +
"tconnectsync might not be functioning properly.")
sys.exit(1)
elif last_process_time_range and (now - last_process_time_range) >= 60 * AUTOUPDATE_FAILURE_MINUTES:
logger.error(AutoupdateFailureException("No new data has been found via the API for over %d minutes. " % AUTOUPDATE_FAILURE_MINUTES +
"tconnectsync might not be functioning properly."))
if AUTOUPDATE_RESTART_ON_FAILURE:
raise AutoupdateFailureException
sys.exit(1)
if len(time_diffs) > 2:
+30
View File
@@ -0,0 +1,30 @@
from .secret import ENABLE_TESTING_MODES
"""Supported synchronization features."""
BASAL = "BASAL"
BOLUS = "BOLUS"
IOB = "IOB"
BOLUS_BG = "BOLUS_BG"
CGM = "CGM"
PUMP_EVENTS = "PUMP_EVENTS"
DEFAULT_FEATURES = [
BASAL,
BOLUS,
IOB
]
ALL_FEATURES = [
BASAL,
BOLUS,
IOB,
PUMP_EVENTS
]
# These modes are not yet ready for wide use.
if ENABLE_TESTING_MODES:
ALL_FEATURES += [
BOLUS_BG,
CGM
]
+13 -1
View File
@@ -53,7 +53,19 @@ class NightscoutApi:
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if latest.status_code != 200:
raise ApiException(latest.status_code, "Nightscout treatments response: %s" % latest.text)
raise ApiException(latest.status_code, "Nightscout last_uploaded_entry response: %s" % latest.text)
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
def last_uploaded_bg_entry(self):
latest = requests.get(urljoin(self.url, 'api/v1/entries.json?count=1&find[device]=' + urllib.parse.quote(ENTERED_BY) + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if latest.status_code != 200:
raise ApiException(latest.status_code, "Nightscout last_uploaded_bg_entry response: %s" % latest.text)
j = latest.json()
if j and len(j) > 0:
+73 -4
View File
@@ -1,8 +1,18 @@
import arrow
ENTERED_BY = "Pump (tconnectsync)"
BASAL_EVENTTYPE = "Temp Basal"
BOLUS_EVENTTYPE = "Combo Bolus"
SITECHANGE_EVENTTYPE = "Site Change"
BASALSUSPENSION_EVENTTYPE = "Basal Suspension"
ACTIVITY_EVENTTYPE = "Activity"
EXERCISE_EVENTTYPE = "Exercise"
SLEEP_EVENTTYPE = "Sleep"
IOB_ACTIVITYTYPE = "tconnect_iob"
"""
Conversion methods for parsing data into Nightscout objects.
"""
@@ -21,9 +31,14 @@ class NightscoutEntry:
"enteredBy": ENTERED_BY
}
# Note that Nightscout is not consistent and uses "Sensor"/"Finger"
# for treatment objects, unlike "sgv"/"mbg" for entries
SENSOR = "Sensor"
FINGER = "Finger"
@staticmethod
def bolus(bolus, carbs, created_at, notes=""):
return {
def bolus(bolus, carbs, created_at, notes="", bg="", bg_type=""):
data = {
"eventType": BOLUS_EVENTTYPE,
"created_at": created_at,
"carbs": int(carbs),
@@ -31,12 +46,66 @@ class NightscoutEntry:
"notes": notes,
"enteredBy": ENTERED_BY,
}
if bg:
if bg_type not in (NightscoutEntry.SENSOR, NightscoutEntry.FINGER):
raise InvalidBolusTypeException
data.update({
"glucose": str(bg),
"glucoseType": bg_type
})
return data
@staticmethod
def iob(iob, created_at):
return {
"activityType": IOB_ACTIVITYTYPE,
"iob": iob,
"iob": float(iob),
"created_at": created_at,
"enteredBy": ENTERED_BY
}
}
@staticmethod
def entry(sgv, created_at):
return {
"type": "sgv",
"sgv": int(sgv),
"date": int(1000 * arrow.get(created_at).timestamp()),
"dateString": arrow.get(created_at).strftime('%Y-%m-%dT%H:%M:%S%z'),
"device": ENTERED_BY,
# delta, direction are undefined
}
@staticmethod
def sitechange(created_at, reason=""):
return {
"eventType": SITECHANGE_EVENTTYPE,
"reason": reason,
"notes": reason,
"created_at": created_at,
"enteredBy": ENTERED_BY
}
@staticmethod
def basalsuspension(created_at, reason=""):
return {
"eventType": BASALSUSPENSION_EVENTTYPE,
"reason": reason,
"notes": reason,
"created_at": created_at,
"enteredBy": ENTERED_BY
}
@staticmethod
def activity(created_at, duration, reason="", event_type=ACTIVITY_EVENTTYPE):
return {
"eventType": event_type,
"reason": reason,
"notes": reason,
"duration": float(duration),
"created_at": created_at,
"enteredBy": ENTERED_BY
}
class InvalidBolusTypeException(RuntimeError):
pass
+98 -7
View File
@@ -1,3 +1,4 @@
from os import stat
import sys
import arrow
@@ -13,7 +14,6 @@ a more digestable format, which is used internally.
"""
class TConnectEntry:
BASAL_EVENTS = { 0: "Suspension", 1: "Profile", 2: "TempRate", 3: "Algorithm" }
ACTIVITY_EVENTS = { 1: "Sleep", 2: "Exercise", 3: "AutoBolus", 4: "CarbOnly" }
@staticmethod
def _epoch_parse(x):
@@ -27,6 +27,10 @@ class TConnectEntry:
# is overwritten with America/New_York, resulting in 2020-09-01T06:00:00-04:00, the
# correct timestamp.
return arrow.get(x, tzinfo="America/Los_Angeles").replace(tzinfo=TIMEZONE_NAME)
@staticmethod
def _jsonp_epoch_parse(x):
return TConnectEntry._epoch_parse(int(x.replace('/Date(', '').replace('-0000)/', '')))
@staticmethod
def parse_ciq_basal_entry(data, delivery_type=""):
@@ -40,6 +44,16 @@ class TConnectEntry:
"duration_mins": duration_mins,
"basal_rate": basal_rate,
}
@staticmethod
def manual_suspension_to_basal_entry(parsedSuspension, seconds):
duration_mins = seconds / 60
return {
"time": parsedSuspension["time"],
"delivery_type": "%s suspension" % parsedSuspension["suspendReason"],
"duration_mins": duration_mins,
"basal_rate": 0.0
}
@staticmethod
def parse_suspension_entry(data):
@@ -85,19 +99,96 @@ class TConnectEntry:
@staticmethod
def parse_bolus_entry(data):
# All DateTime's are stored in the user's timezone.
complete = (data["ExtendedBolusIsComplete"] or data["BolusIsComplete"])
def is_complete(s):
return s and int(s) == 1
complete = is_complete(data["ExtendedBolusIsComplete"]) or is_complete(data["BolusIsComplete"])
extended_bolus = ("extended" in data["Description"].lower())
return {
"description": data["Description"],
"complete": "1" if complete else "",
"completion": data["CompletionStatusDesc"] if not extended_bolus else data["BolexCompletionStatusDesc"],
"request_time": TConnectEntry._datetime_parse(data["RequestDateTime"]).format() if complete and not extended_bolus else None,
"completion_time": TConnectEntry._datetime_parse(data["CompletionDateTime"]).format() if complete and not extended_bolus else None,
"request_time": TConnectEntry._datetime_parse(data["RequestDateTime"]).format() if not extended_bolus else None,
"completion_time": TConnectEntry._datetime_parse(data["CompletionDateTime"]).format() if not extended_bolus else None,
"insulin": data["InsulinDelivered"],
"requested_insulin": data["ActualTotalBolusRequested"],
"carbs": data["CarbSize"],
"bg": data["BG"], # Note: can be empty string for automatic Control-IQ boluses
"user_override": data["UserOverride"],
"extended_bolus": "1" if extended_bolus else "",
"bolex_completion_time": TConnectEntry._datetime_parse(data["BolexCompletionDateTime"]).format() if complete and extended_bolus else None,
"bolex_start_time": TConnectEntry._datetime_parse(data["BolexStartDateTime"]).format() if complete and extended_bolus else None,
}
# Note: completion time can be empty if the extended bolus is in progress
"bolex_completion_time": TConnectEntry._datetime_parse(data["BolexCompletionDateTime"]).format() if data["BolexCompletionDateTime"] and complete and extended_bolus else None,
"bolex_start_time": TConnectEntry._datetime_parse(data["BolexStartDateTime"]).format() if data["BolexStartDateTime"] and complete and extended_bolus else None,
}
@staticmethod
def parse_reading_entry(data):
return {
"time": TConnectEntry._datetime_parse(data["EventDateTime"]).format(),
"bg": data["Readings (CGM / BGM)"],
"type": data["Description"]
}
ACTIVITY_EVENTS = { 1: "Sleep", 2: "Exercise", 3: "AutoBolus", 4: "CarbOnly" }
@staticmethod
def parse_ciq_activity_event(data):
if data["eventType"] not in TConnectEntry.ACTIVITY_EVENTS.keys():
raise UnknownCIQActivityEventException(data)
time = TConnectEntry._epoch_parse(data["x"])
return {
"time": time.format(),
"duration_mins": data["duration"] / 60,
"event_type": TConnectEntry.ACTIVITY_EVENTS[data["eventType"]]
}
BASALSUSPENSION_EVENTS = {
# site-cart corresponds to a Site or Cartridge change,
# specifically a Tubing Filled: Norm AND a Cannula Filled: Norm alert.
# (This means that a typical changing of a cartridge and then a site
# will result in two consecutive events of this type.)
"site-cart": "Site/Cartridge Change",
# alarm corresponds to one of the following:
# - an Empty Cartridge alarm
# - a Pump shutdown
"alarm": "Empty Cartridge/Pump Shutdown",
# manual corresponds to a Pumping Suspended by User event
"manual": "User Suspended"
}
BASALSUSPENSION_SKIPPED_EVENTS = {
# basal-profile events are not very useful; with ControlIQ enabled,
# Tandem does not show them in the tconnect UI.
"basal-profile",
# If an event continues to occur after the date switches over to the next
# day, then the pump generates a "previous" event. This isn't useful to
# us, so we skip them.
"previous",
}
@staticmethod
def parse_basalsuspension_event(data):
if data["SuspendReason"] in TConnectEntry.BASALSUSPENSION_SKIPPED_EVENTS:
return None
if data["SuspendReason"] not in TConnectEntry.BASALSUSPENSION_EVENTS.keys():
raise UnknownBasalSuspensionEventException(data)
time = TConnectEntry._jsonp_epoch_parse(data["EventDateTime"])
return {
"time": time.format(),
"event_type": TConnectEntry.BASALSUSPENSION_EVENTS[data["SuspendReason"]]
}
class UnknownCIQActivityEventException(Exception):
def __init__(self, data):
super().__init__("Unknown CIQ activity event type: %s" % data)
class UnknownBasalSuspensionEventException(Exception):
def __init__(self, data):
super().__init__("Unknown basal suspension event type: %s" % data)
+48 -13
View File
@@ -18,7 +18,17 @@ from .sync.iob import (
process_iob_events,
ns_write_iob_events
)
from .sync.cgm import (
process_cgm_events,
ns_write_cgm_events
)
from .sync.pump_events import (
process_ciq_activity_events,
process_basalsuspension_events,
ns_write_pump_events
)
from .parser.tconnect import TConnectEntry
from .features import BASAL, BOLUS, IOB, BOLUS_BG, CGM, DEFAULT_FEATURES, PUMP_EVENTS
logger = logging.getLogger(__name__)
@@ -27,7 +37,7 @@ Given a TConnectApi object and start/end range, performs a single
cycle of synchronizing data within the time range.
If pretend is true, then doesn't actually write data to Nightscout.
"""
def process_time_range(tconnect, nightscout, time_start, time_end, pretend):
def process_time_range(tconnect, nightscout, time_start, time_end, pretend, features=DEFAULT_FEATURES):
logger.info("Downloading t:connect ControlIQ data")
try:
ciqTherapyTimelineData = tconnect.controliq.therapy_timeline(time_start, time_end)
@@ -59,20 +69,45 @@ def process_time_range(tconnect, nightscout, time_start, time_end, pretend):
added = 0
basalEvents = process_ciq_basal_events(ciqTherapyTimelineData)
if csvBasalData:
logger.debug("CSV basal data found: processing it")
add_csv_basal_events(basalEvents, csvBasalData)
else:
logger.debug("No CSV basal data found")
cgmData = None
if CGM in features or BOLUS_BG in features:
logger.debug("Processing CGM events")
cgmData = process_cgm_events(readingData)
if CGM in features:
logger.debug("Writing CGM events")
added += ns_write_cgm_events(nightscout, cgmData, pretend)
added += ns_write_basal_events(nightscout, basalEvents, pretend=pretend)
if BASAL in features:
basalEvents = process_ciq_basal_events(ciqTherapyTimelineData)
if csvBasalData:
logger.debug("CSV basal data found: processing it")
add_csv_basal_events(basalEvents, csvBasalData)
else:
logger.debug("No CSV basal data found")
bolusEvents = process_bolus_events(bolusData)
added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend)
added += ns_write_basal_events(nightscout, basalEvents, pretend=pretend)
if PUMP_EVENTS in features:
pumpEvents = process_ciq_activity_events(ciqTherapyTimelineData)
logger.debug("CIQ activity events: %s" % pumpEvents)
iobEvents = process_iob_events(iobData)
added += ns_write_iob_events(nightscout, iobEvents, pretend=pretend)
ws2BasalSuspension = tconnect.ws2.basalsuspension(time_start, time_end)
logger.info("Wrote %d events to Nightscout this process cycle")
bsPumpEvents = process_basalsuspension_events(ws2BasalSuspension)
logger.debug("basalsuspension events: %s" % bsPumpEvents)
pumpEvents += bsPumpEvents
added += ns_write_pump_events(nightscout, pumpEvents, pretend=pretend)
if BOLUS in features:
bolusEvents = process_bolus_events(bolusData)
added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend, include_bg=(BOLUS_BG in features))
if IOB in features:
iobEvents = process_iob_events(iobData)
added += ns_write_iob_events(nightscout, iobEvents, pretend=pretend)
logger.info("Wrote %d events to Nightscout this process cycle" % added)
return added
+17 -10
View File
@@ -1,10 +1,20 @@
import os, sys
from dotenv import load_dotenv
import os, sys, pathlib
from dotenv import dotenv_values
load_dotenv()
cwd_path = os.path.join(os.getcwd(), '.env')
global_path = os.path.join(pathlib.Path.home(), '.config/tconnectsync/.env')
def get(*args):
return os.environ.get(*args)
values = {}
if os.path.exists(cwd_path):
values = dotenv_values(cwd_path)
elif os.path.exists(global_path):
values = dotenv_values(global_path)
else:
values = dotenv_values()
def get(val, default=None):
return os.environ.get(val, values.get(val, default))
def get_number(name, default):
val = get(name, default)
@@ -36,11 +46,8 @@ AUTOUPDATE_USE_FIXED_SLEEP = get_bool('AUTOUPDATE_USE_FIXED_SLEEP', 'false')
AUTOUPDATE_FAILURE_MINUTES = get_number('AUTOUPDATE_FAILURE_MINUTES', '180') # 3 hours
AUTOUPDATE_RESTART_ON_FAILURE = get_bool('AUTOUPDATE_RESTART_ON_FAILURE', 'false')
_config = ['TCONNECT_EMAIL', 'TCONNECT_PASSWORD', 'PUMP_SERIAL_NUMBER',
'NS_URL', 'NS_SECRET', 'TIMEZONE_NAME',
'AUTOUPDATE_DEFAULT_SLEEP_SECONDS', 'AUTOUPDATE_MAX_SLEEP_SECONDS',
'AUTOUPDATE_USE_FIXED_SLEEP', 'AUTOUPDATE_FAILURE_MINUTES',
'AUTOUPDATE_RESTART_ON_FAILURE']
ENABLE_TESTING_MODES = get_bool('ENABLE_TESTING_MODES', 'false')
SKIP_NS_LAST_UPLOADED_CHECK = get_bool('SKIP_NS_LAST_UPLOADED_CHECK', 'false')
if __name__ == '__main__':
for k in locals():
+50 -2
View File
@@ -6,6 +6,7 @@ from ..parser.nightscout import (
NightscoutEntry
)
from ..parser.tconnect import TConnectEntry
from ..secret import SKIP_NS_LAST_UPLOADED_CHECK
logger = logging.getLogger(__name__)
@@ -32,11 +33,53 @@ def process_ciq_basal_events(data):
for b in data["basal"]["profileDeliveryEvents"]:
basalEvents.append(TConnectEntry.parse_ciq_basal_entry(b, delivery_type="profileDelivery"))
basalEvents.sort(key=lambda x: arrow.get(x["time"]))
# Suspensions with suspendReason 'control-iq' will match a basal event found above.
for i in basalEvents:
if i["time"] in suspensionEvents:
i["suspendReason"] = suspensionEvents[i["time"]]["suspendReason"]
i["delivery_type"] += " (" + suspensionEvents[i["time"]]["suspendReason"] + " suspension)"
del suspensionEvents[i["time"]]
# Suspensions with suspendReason 'manual' do not have an associated basal event,
# and require extra processing.
basalEvents.sort(key=lambda x: arrow.get(x["time"]))
unprocessedSuspensions = list(suspensionEvents.values())
unprocessedSuspensions.sort(key=lambda x: arrow.get(x["time"]))
# For the remaining suspensions which did not match with an existing basal event,
# add a new event manually. This means we need to calculate the duration of the
# suspension.
newEvents = []
for i in range(len(basalEvents)):
if len(unprocessedSuspensions) == 0:
break
existingTime = arrow.get(basalEvents[i]["time"])
unprocessedTime = arrow.get(unprocessedSuspensions[0]["time"])
# If we've found an event which occurs after the suspension, then the
# difference in their timestamps is the duration of the suspension.
if i > 0 and existingTime > unprocessedTime:
suspension = unprocessedSuspensions.pop(0)
# TConnect's internal duration object tracks the duration in seconds
seconds = (existingTime - unprocessedTime).seconds
newEvent = TConnectEntry.manual_suspension_to_basal_entry(suspension, seconds)
logger.debug("Creating basal event for unprocessed suspension: %s" % newEvent)
newEvents.append(newEvent)
# Any remaining suspensions which have not been processed have not ended,
# which means we do not know their duration; so we will skip them (for now)
# Add any new events and re-sort
if newEvents:
basalEvents += newEvents
basalEvents.sort(key=lambda x: arrow.get(x["time"]))
return basalEvents
@@ -69,6 +112,11 @@ def ns_write_basal_events(nightscout, basalEvents, pretend=False):
last_upload_time = arrow.get(last_upload["created_at"])
logger.info("Last Nightscout basal upload: %s" % last_upload_time)
if SKIP_NS_LAST_UPLOADED_CHECK:
logger.warning("Overriding last upload check")
last_upload = None
last_upload_time = None
add_count = 0
for event in basalEvents:
if last_upload_time and arrow.get(event["time"]) < last_upload_time:
+51 -11
View File
@@ -1,18 +1,21 @@
import arrow
import logging
from tconnectsync.sync.cgm import find_event_at
from ..parser.nightscout import (
BOLUS_EVENTTYPE,
NightscoutEntry
)
from ..parser.tconnect import TConnectEntry
from ..secret import SKIP_NS_LAST_UPLOADED_CHECK
logger = logging.getLogger(__name__)
"""
Given bolus data input from the therapy timeline CSV, converts it into a digestable format.
"""
def process_bolus_events(bolusdata):
def process_bolus_events(bolusdata, cgmEvents=None):
bolusEvents = []
for b in bolusdata:
@@ -20,20 +23,41 @@ def process_bolus_events(bolusdata):
if parsed["completion"] != "Completed":
if parsed["insulin"] and float(parsed["insulin"]) > 0:
# Count non-completed bolus if any insulin was delivered (vs. the amount of insulin requested)
parsed["description"] += " (%s)" % parsed["completion"]
parsed["description"] += " (%s: requested %s units)" % (parsed["completion"], parsed["requested_insulin"])
else:
logger.warning("Skipping non-completed bolus data (was a bolus in progress?): %s parsed: %s" % (b, parsed))
continue
if parsed["bg"] and cgmEvents:
requested_at = parsed["request_time"] if not parsed["extended_bolus"] else parsed["bolex_start_time"]
parsed["bg_type"] = guess_bolus_bg_type(parsed["bg"], requested_at, cgmEvents)
bolusEvents.append(parsed)
bolusEvents.sort(key=lambda event: arrow.get(event["completion_time"] if not event["extended_bolus"] else event["bolex_start_time"]))
bolusEvents.sort(key=lambda event: arrow.get(event["request_time"] if not event["extended_bolus"] else event["bolex_start_time"]))
return bolusEvents
"""
Determine whether the given BG specified in the bolus is identical to the
most recent CGM reading at that time. If it is, return SENSOR.
Otherwise, return FINGER.
"""
def guess_bolus_bg_type(bg, created_at, cgmEvents):
if not cgmEvents:
return NightscoutEntry.FINGER
event = find_event_at(cgmEvents, created_at)
if event and str(event["bg"]) == str(bg):
return NightscoutEntry.SENSOR
return NightscoutEntry.FINGER
"""
Given processed bolus data, adds bolus events to Nightscout.
"""
def ns_write_bolus_events(nightscout, bolusEvents, pretend=False):
def ns_write_bolus_events(nightscout, bolusEvents, pretend=False, include_bg=False, reading_events=None):
logger.debug("ns_write_bolus_events: querying for last uploaded entry")
last_upload = nightscout.last_uploaded_entry(BOLUS_EVENTTYPE)
last_upload_time = None
@@ -41,19 +65,35 @@ def ns_write_bolus_events(nightscout, bolusEvents, pretend=False):
last_upload_time = arrow.get(last_upload["created_at"])
logger.info("Last Nightscout bolus upload: %s" % last_upload_time)
if SKIP_NS_LAST_UPLOADED_CHECK:
logger.warning("Overriding last upload check")
last_upload = None
last_upload_time = None
add_count = 0
for event in bolusEvents:
if last_upload_time and arrow.get(event["completion_time"]) <= last_upload_time:
created_at = event["completion_time"] if not event["extended_bolus"] else event["bolex_start_time"]
if last_upload_time and arrow.get(created_at) <= last_upload_time:
if pretend:
logger.info("Skipping basal event before last upload time: %s" % event)
continue
entry = NightscoutEntry.bolus(
bolus=event["insulin"],
carbs=event["carbs"],
created_at=event["completion_time"] if not event["extended_bolus"] else event["bolex_start_time"],
notes="{}{}{}".format(event["description"], " (Override)" if event["user_override"] == "1" else "", " (Extended)" if event["extended_bolus"] == "1" else "")
)
if include_bg and event["bg"]:
entry = NightscoutEntry.bolus(
bolus=event["insulin"],
carbs=event["carbs"],
created_at=created_at,
notes="{}{}{}".format(event["description"], " (Override)" if event["user_override"] == "1" else "", " (Extended)" if event["extended_bolus"] == "1" else ""),
bg=event["bg"],
bg_type=event["bg_type"]
)
else:
entry = NightscoutEntry.bolus(
bolus=event["insulin"],
carbs=event["carbs"],
created_at=created_at,
notes="{}{}{}".format(event["description"], " (Override)" if event["user_override"] == "1" else "", " (Extended)" if event["extended_bolus"] == "1" else "")
)
add_count += 1
+69
View File
@@ -0,0 +1,69 @@
import json
import arrow
import logging
from ..parser.tconnect import TConnectEntry
from ..parser.nightscout import NightscoutEntry
logger = logging.getLogger(__name__)
def process_cgm_events(readingData):
data = []
for r in readingData:
data.append(TConnectEntry.parse_reading_entry(r))
return data
"""
Given reading data and a time, finds the BG reading event which would have
been the current one at that time. e.g., it looks before the given time,
not after.
This is a heuristic for checking whether the BG component of a bolus was
manually entered or inferred based on the pump's CGM.
"""
def find_event_at(cgmEvents, find_time):
find_t = arrow.get(find_time)
events = list(map(lambda x: (arrow.get(x["time"]), x), cgmEvents))
events.sort()
closestReading = None
for t, r in events:
if t > find_t:
break
closestReading = r
return closestReading
"""
Given processed CGM data, adds reading entries to Nightscout.
"""
def ns_write_cgm_events(nightscout, cgmEvents, pretend=False):
logger.debug("ns_write_cgm_events: querying for last uploaded entry")
last_upload = nightscout.last_uploaded_bg_entry()
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["dateString"])
logger.info("Last Nightscout CGM upload: %s" % last_upload_time)
add_count = 0
for event in cgmEvents:
created_at = event["time"]
if last_upload_time and arrow.get(created_at) <= last_upload_time:
if pretend:
logger.info("Skipping CGM event before last upload time: %s" % event)
continue
entry = NightscoutEntry.entry(
sgv=event["bg"],
created_at=created_at
)
add_count += 1
logger.info(" Processing cgm reading: %s entry: %s" % (event, entry))
if not pretend:
nightscout.upload_entry(entry, entity='entries')
return add_count
+206
View File
@@ -0,0 +1,206 @@
import arrow
import logging
from ..parser.nightscout import (
SITECHANGE_EVENTTYPE,
BASALSUSPENSION_EVENTTYPE,
EXERCISE_EVENTTYPE,
SLEEP_EVENTTYPE,
ACTIVITY_EVENTTYPE,
NightscoutEntry
)
from ..parser.tconnect import TConnectEntry
from ..secret import SKIP_NS_LAST_UPLOADED_CHECK
logger = logging.getLogger(__name__)
"""
Given a list of "activity events" from the CIQ therapy timeline endpoint,
process it into our internal events format.
These events contain a duration.
"""
def process_ciq_activity_events(data):
events = []
for event in data["events"]:
events.append(TConnectEntry.parse_ciq_activity_event(event))
return events
"""
Given a list of "basal suspension events" from the basalsuspension WS2 endpoint,
process it into our internal events format.
These events do NOT contain a duration.
"""
def process_basalsuspension_events(data):
events = []
for event in data['BasalSuspension']:
parsed = TConnectEntry.parse_basalsuspension_event(event)
if parsed:
events.append(parsed)
return events
"""
Given processed pump event data (of various types), write them to Nightscout
"""
def ns_write_pump_events(nightscout, pumpEvents, pretend=False):
count = 0
siteChangeEvents = []
emptyCartEvents = []
userSuspendedEvents = []
exerciseEvents = []
sleepEvents = []
activityEvents = []
for event in pumpEvents:
if event["event_type"] == TConnectEntry.BASALSUSPENSION_EVENTS["site-cart"]:
siteChangeEvents.append(event)
elif event["event_type"] in TConnectEntry.BASALSUSPENSION_EVENTS["alarm"]:
emptyCartEvents.append(event)
elif event["event_type"] in TConnectEntry.BASALSUSPENSION_EVENTS["manual"]:
userSuspendedEvents.append(event)
elif event["event_type"] == "Exercise":
exerciseEvents.append(event)
elif event["event_type"] == "Sleep":
sleepEvents.append(event)
elif event["event_type"] in TConnectEntry.ACTIVITY_EVENTS.values():
activityEvents.append(event)
logger.debug("siteChangeEvents: %s" % siteChangeEvents)
logger.debug("emptyCartEvents: %s" % emptyCartEvents)
logger.debug("userSuspendedEvents: %s" % userSuspendedEvents)
logger.debug("exerciseEvents: %s" % exerciseEvents)
logger.debug("sleepEvents: %s" % sleepEvents)
logger.debug("activityEvents: %s" % activityEvents)
count += ns_write_pump_sitechange_events(nightscout, siteChangeEvents, pretend=pretend)
count += ns_write_empty_cart_events(nightscout, emptyCartEvents, pretend=pretend)
count += ns_write_user_suspended_events(nightscout, userSuspendedEvents, pretend=pretend)
count += ns_write_exercise_activity_events(nightscout, exerciseEvents, pretend=pretend)
count += ns_write_sleep_activity_events(nightscout, sleepEvents, pretend=pretend)
count += ns_write_activity_events(nightscout, activityEvents, pretend=pretend)
return count
def ns_write_pump_sitechange_events(nightscout, siteChangeEvents, pretend=False):
return _ns_write_pump_events(
nightscout,
siteChangeEvents,
lambda event: NightscoutEntry.sitechange(
created_at=event["time"],
reason=event["event_type"]
),
SITECHANGE_EVENTTYPE,
pretend=pretend)
def ns_write_empty_cart_events(nightscout, emptyCartEvents, pretend=False):
return _ns_write_pump_events(
nightscout,
emptyCartEvents,
lambda event: NightscoutEntry.basalsuspension(
created_at=event["time"],
reason=event["event_type"]
),
BASALSUSPENSION_EVENTTYPE,
pretend=pretend)
def ns_write_user_suspended_events(nightscout, userSuspendedEvents, pretend=False):
return _ns_write_pump_events(
nightscout,
userSuspendedEvents,
lambda event: NightscoutEntry.basalsuspension(
created_at=event["time"],
reason=event["event_type"]
),
BASALSUSPENSION_EVENTTYPE,
pretend=pretend)
def ns_write_exercise_activity_events(nightscout, exerciseEvents, pretend=False):
return _ns_write_pump_events(
nightscout,
exerciseEvents,
lambda event: NightscoutEntry.activity(
created_at=event["time"],
reason=event["event_type"],
duration=event["duration_mins"],
event_type=EXERCISE_EVENTTYPE
),
EXERCISE_EVENTTYPE,
pretend=pretend)
def ns_write_sleep_activity_events(nightscout, sleepEvents, pretend=False):
return _ns_write_pump_events(
nightscout,
sleepEvents,
lambda event: NightscoutEntry.activity(
created_at=event["time"],
reason=event["event_type"],
duration=event["duration_mins"],
event_type=SLEEP_EVENTTYPE
),
SLEEP_EVENTTYPE,
pretend=pretend)
def ns_write_activity_events(nightscout, activityEvents, pretend=False):
return _ns_write_pump_events(
nightscout,
activityEvents,
lambda event: NightscoutEntry.activity(
created_at=event["time"],
reason=event["event_type"],
duration=event["duration_mins"]
),
ACTIVITY_EVENTTYPE,
pretend=pretend)
def _ns_write_pump_events(nightscout, events, buildNsEventFunc, eventType, pretend=False):
if len(events) == 0:
logger.debug("No %s events to process" % eventType)
return 0
logger.debug("ns_write_pump_events: querying for last %s" % eventType)
last_upload = nightscout.last_uploaded_entry(eventType)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
logger.info("Last Nightscout %s: %s" % (eventType, last_upload_time))
if SKIP_NS_LAST_UPLOADED_CHECK:
logger.warning("Overriding last upload check")
last_upload = None
last_upload_time = None
add_count = 0
for event in events:
created_at = event["time"]
if last_upload_time and arrow.get(created_at) <= last_upload_time:
skip = True
if "duration_mins" in event.keys() and "duration" in last_upload.keys():
if created_at == last_upload["created_at"] and float(event["duration_mins"]) > float(last_upload["duration"]):
logger.info("Latest %s event needs updating: duration has increased from %s to %s: %s" % (eventType, last_upload["duration"], event["duration_mins"], event))
logger.info("Deleting previous %s: %s" % (eventType, last_upload))
nightscout.delete_entry('treatments/%s' % last_upload["_id"])
skip = False
if skip:
if pretend:
logger.info("Skipping %s event before last upload time: %s" % (eventType, event))
continue
entry = buildNsEventFunc(event)
add_count += 1
logger.info(" Processing %s: %s entry: %s" % (eventType, event, entry))
if not pretend:
nightscout.upload_entry(entry)
return add_count
+10
View File
@@ -0,0 +1,10 @@
# Hack to enable debug-level logging when running tests
# with `python3 -m unittest discover`
import sys
if 'unittest' in sys.modules:
import logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
+1
View File
@@ -17,6 +17,7 @@ class ControlIQApi(tconnectsync.api.controliq.ControlIQApi):
class WS2Api(tconnectsync.api.ws2.WS2Api):
def __init__(self):
self.BASE_URL = 'invalid://'
self.SLEEP_SECONDS_INCREMENT = 0.01
def get(self, endpoint, query):
raise NotImplementedError
+3 -3
View File
@@ -8,14 +8,14 @@ class NightscoutApi(tconnectsync.nightscout.NightscoutApi):
self.secret = 'invalid'
self.uploaded_entries = collections.defaultdict(list)
self.deleted_entries = collections.defaultdict(list)
self.deleted_entries = []
self.put_entries = collections.defaultdict(list)
def upload_entry(self, ns_format, entity='treatments'):
self.uploaded_entries[entity].append(ns_format)
def delete_entry(self, ns_format, entity):
self.deleted_entries[entity].append(ns_format)
def delete_entry(self, ns_path):
self.deleted_entries.append(ns_path)
def put_entry(self, ns_format, entity):
self.put_entries[entity].append(ns_format)
+106 -1
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.parser.nightscout import NightscoutEntry
from tconnectsync.parser.nightscout import NightscoutEntry, InvalidBolusTypeException
class TestNightscoutEntry(unittest.TestCase):
def test_basal(self):
@@ -73,6 +73,69 @@ class TestNightscoutEntry(unittest.TestCase):
}
)
def test_bolus_with_bg(self):
self.assertEqual(
NightscoutEntry.bolus(
bolus=7.5,
carbs=45,
created_at="2021-03-16 00:25:21-04:00",
bg="123",
bg_type=NightscoutEntry.SENSOR),
{
"eventType": "Combo Bolus",
"created_at": "2021-03-16 00:25:21-04:00",
"carbs": 45,
"insulin": 7.5,
"notes": "",
"enteredBy": "Pump (tconnectsync)",
"glucose": "123",
"glucoseType": "Sensor"
}
)
self.assertEqual(
NightscoutEntry.bolus(
bolus=0.5,
carbs=5,
created_at="2021-03-16 12:25:21-04:00",
bg="150",
bg_type=NightscoutEntry.FINGER),
{
"eventType": "Combo Bolus",
"created_at": "2021-03-16 12:25:21-04:00",
"carbs": 5,
"insulin": 0.5,
"notes": "",
"enteredBy": "Pump (tconnectsync)",
"glucose": "150",
"glucoseType": "Finger"
}
)
def test_bolus_with_bg_invalid_type(self):
self.assertRaises(InvalidBolusTypeException,
NightscoutEntry.bolus,
bolus=0.5,
carbs=5,
created_at="2021-03-16 12:25:21-04:00",
bg="150")
self.assertRaises(InvalidBolusTypeException,
NightscoutEntry.bolus,
bolus=0.5,
carbs=5,
created_at="2021-03-16 12:25:21-04:00",
bg="150",
bg_type="")
self.assertRaises(InvalidBolusTypeException,
NightscoutEntry.bolus,
bolus=0.5,
carbs=5,
created_at="2021-03-16 12:25:21-04:00",
bg="150",
bg_type="unknown")
def test_iob(self):
self.assertEqual(
NightscoutEntry.iob(
@@ -86,6 +149,48 @@ class TestNightscoutEntry(unittest.TestCase):
}
)
def test_entry(self):
self.assertEqual(
NightscoutEntry.entry(
sgv=152,
created_at="2021-10-23 22:17:14-04:00"),
{
"type": "sgv",
"sgv": 152,
"date": 1635041834000,
"dateString": "2021-10-23T22:17:14-0400",
"device": "Pump (tconnectsync)",
}
)
def test_sitechange(self):
self.assertEqual(
NightscoutEntry.sitechange(
created_at="2021-12-05T00:16:35.058Z",
reason="reason"),
{
"eventType": "Site Change",
"reason": "reason",
"notes": "reason",
"created_at": "2021-12-05T00:16:35.058Z",
"enteredBy": "Pump (tconnectsync)"
}
)
def test_basalsuspension(self):
self.assertEqual(
NightscoutEntry.basalsuspension(
created_at="2021-12-05T00:16:35.058Z",
reason="reason"),
{
"eventType": "Basal Suspension",
"reason": "reason",
"notes": "reason",
"created_at": "2021-12-05T00:16:35.058Z",
"enteredBy": "Pump (tconnectsync)"
}
)
if __name__ == '__main__':
unittest.main()
+407 -1
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.parser.tconnect import TConnectEntry
from tconnectsync.parser.tconnect import TConnectEntry, UnknownBasalSuspensionEventException, UnknownCIQActivityEventException
class TestTConnectEntryBasal(unittest.TestCase):
def test_parse_ciq_basal_entry(self):
@@ -33,6 +33,103 @@ class TestTConnectEntryBasal(unittest.TestCase):
}
)
class TestTConnectEntrySuspension(unittest.TestCase):
def test_parse_suspension_entry(self):
self.assertEqual(
TConnectEntry.parse_suspension_entry({
"suspendReason": "control-iq",
"continuation": None,
"x": 1615879821
}),
{
"time": "2021-03-16 00:30:21-04:00",
"continuation": None,
"suspendReason": "control-iq"
}
)
self.assertEqual(
TConnectEntry.parse_suspension_entry({
"suspendReason": "control-iq",
"continuation": "previous",
"x": 1634022000
}),
{
"time": "2021-10-12 00:00:00-04:00",
"continuation": "previous",
"suspendReason": "control-iq"
}
)
class TestTConnectEntrySuspensionToBasal(unittest.TestCase):
def test_manual_suspension_to_basal_entry(self):
suspension = {
"time": "2021-03-16 00:30:21-04:00",
"continuation": None,
"suspendReason": "manual"
}
self.assertEqual(
TConnectEntry.manual_suspension_to_basal_entry(
suspension,
seconds=300
), {
"time": "2021-03-16 00:30:21-04:00",
"delivery_type": "manual suspension",
"duration_mins": 5.0,
"basal_rate": 0.0
}
)
class TestTConnectEntryCGM(unittest.TestCase):
def test_parse_cgm_entry(self):
self.assertEqual(
TConnectEntry.parse_cgm_entry({
"DeviceType": "t:slim X2 Insulin Pump",
"SerialNumber": "11111111",
"Description": "EGV",
"EventDateTime": "2021-10-12T00:01:12",
"Readings (CGM / BGM)": "131"
}),
{
"time": "2021-10-12 00:01:12-04:00",
"reading": "131",
"reading_type": "EGV"
}
)
class TestTConnectEntryIOB(unittest.TestCase):
entry1 = {
"Type": "IOB",
"EventID": "81",
"EventDateTime": "2021-10-12T00:00:30",
"IOB": "6.91"
}
def test_parse_iob_entry1(self):
self.assertEqual(
TConnectEntry.parse_iob_entry(self.entry1),
{
"time": "2021-10-12 00:00:30-04:00",
"iob": "6.91",
"event_id": "81"
}
)
entry2 = {
"Type": "IOB",
"EventID": "9",
"EventDateTime": "2021-10-12T00:10:30",
"IOB": "6.80"
}
def test_parse_iob_entry2(self):
self.assertEqual(
TConnectEntry.parse_iob_entry(self.entry2),
{
"time": "2021-10-12 00:10:30-04:00",
"iob": "6.80",
"event_id": "9"
}
)
class TestTConnectEntryBolus(unittest.TestCase):
entryStdCorrection = {
"Type": "Bolus",
@@ -87,7 +184,9 @@ class TestTConnectEntryBolus(unittest.TestCase):
"request_time": "2021-04-01 12:53:36-04:00",
"completion_time": "2021-04-01 12:58:26-04:00",
"insulin": "13.53",
"requested_insulin": "13.53",
"carbs": "75",
"bg": "141",
"user_override": "0",
"extended_bolus": "",
"bolex_completion_time": None,
@@ -147,7 +246,9 @@ class TestTConnectEntryBolus(unittest.TestCase):
"request_time": "2021-04-01 23:21:58-04:00",
"completion_time": "2021-04-01 23:23:17-04:00",
"insulin": "1.25",
"requested_insulin": "1.25",
"carbs": "0",
"bg": "159",
"user_override": "1",
"extended_bolus": "",
"bolex_completion_time": None,
@@ -207,13 +308,318 @@ class TestTConnectEntryBolus(unittest.TestCase):
"request_time": "2021-04-02 00:59:13-04:00",
"completion_time": "2021-04-02 01:00:47-04:00",
"insulin": "1.70",
"requested_insulin": "1.70",
"carbs": "0",
"bg": "",
"user_override": "0",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
entryStdIncompleteZero = {
"Type": "Bolus",
"Description": "Standard",
"BG": "144",
"IOB": "1.20",
"BolusRequestID": "9694.000",
"BolusCompletionID": "9694.000",
"CompletionDateTime": "2021-10-08T15:47:02",
"InsulinDelivered": "0.00",
"FoodDelivered": "0.00",
"CorrectionDelivered": "0.00",
"CompletionStatusID": "0",
"CompletionStatusDesc": "User Aborted",
"BolusIsComplete": "0",
"BolexCompletionID": "",
"BolexSize": "",
"BolexStartDateTime": "",
"BolexCompletionDateTime": "",
"BolexInsulinDelivered": "",
"BolexIOB": "",
"BolexCompletionStatusID": "",
"BolexCompletionStatusDesc": "",
"ExtendedBolusIsComplete": "",
"EventDateTime": "2021-10-08T15:46:56",
"RequestDateTime": "2021-10-08T15:46:56",
"BolusType": "Carb",
"BolusRequestOptions": "Standard",
"StandardPercent": "100.00",
"Duration": "0",
"CarbSize": "0",
"UserOverride": "1",
"TargetBG": "110",
"CorrectionFactor": "30.00",
"FoodBolusSize": "0.00",
"CorrectionBolusSize": "0.00",
"ActualTotalBolusRequested": "0.50",
"IsQuickBolus": "0",
"EventHistoryReportEventDesc": "0",
"EventHistoryReportDetails": "Food Bolus",
"NoteID": "CF 1:30 - Carb Ratio 1:6 - Target BG 110 | Override: Pump calculated Bolus = 0.0 units",
"IndexID": "0",
"Note": "1669328"
}
def test_parse_bolus_entry_std_incomplete_zero(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdIncompleteZero),
{
"description": "Standard",
"complete": "",
"completion": "User Aborted",
"request_time": "2021-10-08 15:46:56-04:00",
"completion_time": "2021-10-08 15:47:02-04:00",
"insulin": "0.00",
"requested_insulin": "0.50",
"carbs": "0",
"bg": "144",
"user_override": "1",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
entryStdIncompletePartial = {
"Type": "Bolus",
"Description": "Standard/Correction",
"BG": "189",
"IOB": "",
"BolusRequestID": "9261.000",
"BolusCompletionID": "9261.000",
"CompletionDateTime": "2021-09-06T12:24:47",
"InsulinDelivered": "1.82",
"FoodDelivered": "0.00",
"CorrectionDelivered": "1.82",
"CompletionStatusID": "1",
"CompletionStatusDesc": "Terminated by Alarm",
"BolusIsComplete": "0",
"BolexCompletionID": "",
"BolexSize": "",
"BolexStartDateTime": "",
"BolexCompletionDateTime": "",
"BolexInsulinDelivered": "",
"BolexIOB": "",
"BolexCompletionStatusID": "",
"BolexCompletionStatusDesc": "",
"ExtendedBolusIsComplete": "",
"EventDateTime": "2021-09-06T12:23:23",
"RequestDateTime": "2021-09-06T12:23:23",
"BolusType": "Carb",
"BolusRequestOptions": "Standard/Correction",
"StandardPercent": "100.00",
"Duration": "0",
"CarbSize": "0",
"UserOverride": "0",
"TargetBG": "110",
"CorrectionFactor": "30.00",
"FoodBolusSize": "0.00",
"CorrectionBolusSize": "2.63",
"ActualTotalBolusRequested": "2.63",
"IsQuickBolus": "0",
"EventHistoryReportEventDesc": "0",
"EventHistoryReportDetails": "Correction & Food Bolus",
"NoteID": "CF 1:30 - Carb Ratio 1:6 - Target BG 110",
"IndexID": "0",
"Note": "1589227"
}
def test_parse_bolus_entry_std_incomplete_partial(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdIncompletePartial),
{
"description": "Standard/Correction",
"complete": "",
"completion": "Terminated by Alarm",
"request_time": "2021-09-06 12:23:23-04:00",
"completion_time": "2021-09-06 12:24:47-04:00",
"insulin": "1.82",
"requested_insulin": "2.63",
"carbs": "0",
"bg": "189",
"user_override": "0",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
class TestTConnectEntryReading(unittest.TestCase):
entry1 = {
"DeviceType": "t:slim X2 Insulin Pump",
"SerialNumber": "90556643",
"Description": "EGV",
"EventDateTime": "2021-10-23T12:55:53",
"Readings (CGM / BGM)": "135"
}
def test_parse_reading_entry1(self):
self.assertEqual(
TConnectEntry.parse_reading_entry(self.entry1),
{
"time": "2021-10-23 12:55:53-04:00",
"bg": "135",
"type": "EGV"
}
)
entry2 = {
"DeviceType": "t:slim X2 Insulin Pump",
"SerialNumber": "90556643",
"Description": "EGV",
"EventDateTime": "2021-10-23T16:15:52",
"Readings (CGM / BGM)": "93"
}
def test_parse_reading_entry2(self):
self.assertEqual(
TConnectEntry.parse_reading_entry(self.entry2),
{
"time": "2021-10-23 16:15:52-04:00",
"bg": "93",
"type": "EGV"
}
)
entry3 = {
"DeviceType": "t:slim X2 Insulin Pump",
"SerialNumber": "90556643",
"Description": "EGV",
"EventDateTime": "2021-10-23T16:20:52",
"Readings (CGM / BGM)": "100"
}
def test_parse_reading_entry3(self):
self.assertEqual(
TConnectEntry.parse_reading_entry(self.entry3),
{
"time": "2021-10-23 16:20:52-04:00",
"bg": "100",
"type": "EGV"
}
)
entry4 = {
"DeviceType": "t:slim X2 Insulin Pump",
"SerialNumber": "90556643",
"Description": "EGV",
"EventDateTime": "2021-10-23T16:25:52",
"Readings (CGM / BGM)": "107"
}
def test_parse_reading_entry4(self):
self.assertEqual(
TConnectEntry.parse_reading_entry(self.entry4),
{
"time": "2021-10-23 16:25:52-04:00",
"bg": "107",
"type": "EGV"
}
)
class TestTConnectEntryCIQEvent(unittest.TestCase):
def test_parse_ciq_activity_event_sleep(self):
self.assertEqual(
TConnectEntry.parse_ciq_activity_event({
"continuation": None,
"duration": 30661,
"eventType": 1,
"timeZoneId": "America/Los_Angeles",
"x": 1638091836
}),
{
"time": "2021-11-28 01:30:36-05:00",
"duration_mins": (30661 / 60),
"event_type": "Sleep"
}
)
def test_parse_ciq_activity_event_exercise(self):
self.assertEqual(
TConnectEntry.parse_ciq_activity_event({
"duration": 1200,
"eventType": 2,
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619901912
}),
{
"time": "2021-05-01 13:45:12-04:00",
"duration_mins": 20,
"event_type": "Exercise"
}
)
def test_parse_ciq_activity_event_unknown_id(self):
self.assertRaises(
UnknownCIQActivityEventException,
TConnectEntry.parse_ciq_activity_event,
{
"duration": 1200,
"eventType": 5,
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619901912
}
)
class TestTConnectEntryBasalSuspensionEvent(unittest.TestCase):
def test_parse_basalsuspension_event_sitecart(self):
self.assertEqual(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1638663490000-0000)/',
'SuspendReason': 'site-cart'
}),
{
"time": "2021-12-04 16:18:10-05:00",
"event_type": "Site/Cartridge Change"
}
)
def test_parse_basalsuspension_event_alarm(self):
self.assertEqual(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1637863616000-0000)/',
'SuspendReason': 'alarm'
}),
{
"time": "2021-11-25 10:06:56-05:00",
"event_type": "Empty Cartridge/Pump Shutdown"
}
)
def test_parse_basalsuspension_event_manual(self):
self.assertEqual(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1638662852000-0000)/',
'SuspendReason': 'manual'
}),
{
"time": "2021-12-04 16:07:32-05:00",
"event_type": "User Suspended"
}
)
def test_parse_basalsuspension_event_basalprofile_skipped(self):
self.assertIsNone(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1638659343000-0000)/',
'SuspendReason': 'basal-profile',
})
)
def test_parse_basalsuspension_event_previous_skipped(self):
self.assertIsNone(
TConnectEntry.parse_basalsuspension_event({
'Continuation': 'continuation',
'EventDateTime': '/Date(1638604800000-0000)/',
'SuspendReason': 'previous',
})
)
def test_parse_basalsuspension_event_unknown_suspendreason(self):
self.assertRaises(
UnknownBasalSuspensionEventException,
TConnectEntry.parse_basalsuspension_event,
{
'EventDateTime': '/Date(1638604800000-0000)/',
'SuspendReason': 'unknown',
}
)
if __name__ == '__main__':
+71 -11
View File
@@ -1,10 +1,14 @@
#!/usr/bin/env python3
import unittest
import copy
from tconnectsync.sync.basal import process_ciq_basal_events
from tconnectsync.parser.tconnect import TConnectEntry
class TestBasalSync(unittest.TestCase):
maxDiff = None
base = {
"basal": {
"profileRates": [],
@@ -20,31 +24,31 @@ class TestBasalSync(unittest.TestCase):
@staticmethod
def get_example_ciq_basal_events():
data = TestBasalSync.base.copy()
data = copy.deepcopy(TestBasalSync.base)
data["basal"]["tempDeliveryEvents"] = [
{
"y": 0.8,
"duration": 1221,
"x": 1615878000
"x": 1615878000 # 12:00:00
}
]
data["basal"]["algorithmDeliveryEvents"] = [
{
"y": 0.797,
"duration": 300,
"x": 1615879521
"x": 1615879521 # 12:25:21
},
{
"y": 0,
"duration": 2693,
"x": 1615879821
"x": 1615879821 # 12:30:21
},
]
data["basal"]["profileDeliveryEvents"] = [
{
"y": 0.799,
"duration": 300,
"x": 1615879221
"x": 1615879221 # 12:20:21
}
]
@@ -52,7 +56,7 @@ class TestBasalSync(unittest.TestCase):
{
"suspendReason": "control-iq",
"continuation": None,
"x": 1615879821
"x": 1615879821 # 12:30:21
},
]
@@ -73,12 +77,68 @@ class TestBasalSync(unittest.TestCase):
self.assertEqual(basalEvents[2], TConnectEntry.parse_ciq_basal_entry(
data["basal"]["algorithmDeliveryEvents"][0], delivery_type="algorithmDelivery"))
self.assertEqual(basalEvents[3], {
"suspendReason": "control-iq",
**TConnectEntry.parse_ciq_basal_entry(
self.assertEqual(basalEvents[3], TConnectEntry.parse_ciq_basal_entry(
data["basal"]["algorithmDeliveryEvents"][1],
delivery_type="algorithmDelivery")
})
delivery_type="algorithmDelivery (control-iq suspension)")
)
@staticmethod
def get_example_ciq_basal_events_with_manual_suspension():
data = copy.deepcopy(TestBasalSync.base)
data["basal"]["tempDeliveryEvents"] = []
data["basal"]["algorithmDeliveryEvents"] = [
{
"y": 1.14,
"duration": 300,
"x": 1635187357 # 11:42:37
},
{
"y": 0.8,
"duration": 1198,
"x": 1635187657 # 11:47:37
},
{
"y": 0.8,
"duration": 599,
"x": 1635190967 # 12:42:47
}
]
data["basal"]["profileDeliveryEvents"] = []
data["suspensionDeliveryEvents"] = [
{
"suspendReason": "manual",
"continuation": None,
"x": 1635188855 # 12:07:35
},
{
"suspendReason": "manual",
"continuation": None,
"x": 1635191566 # 12:52:46
},
]
return data
def test_process_ciq_basal_events_with_manual_suspension(self):
data = TestBasalSync.get_example_ciq_basal_events_with_manual_suspension()
basalEvents = process_ciq_basal_events(data)
self.assertEqual(len(basalEvents), 4)
self.assertEqual(basalEvents[0], TConnectEntry.parse_ciq_basal_entry(
data["basal"]["algorithmDeliveryEvents"][0], delivery_type="algorithmDelivery"))
self.assertEqual(basalEvents[1], TConnectEntry.parse_ciq_basal_entry(
data["basal"]["algorithmDeliveryEvents"][1], delivery_type="algorithmDelivery"))
self.assertEqual(basalEvents[2], TConnectEntry.manual_suspension_to_basal_entry(
TConnectEntry.parse_suspension_entry(data["suspensionDeliveryEvents"][0]),
seconds=2112, # 2112 seconds between 12:07:35 and 12:42:47
))
self.assertEqual(basalEvents[3], TConnectEntry.parse_ciq_basal_entry(
data["basal"]["algorithmDeliveryEvents"][2], delivery_type="algorithmDelivery"))
if __name__ == '__main__':
+150 -8
View File
@@ -1,34 +1,176 @@
#!/usr/bin/env python3
import unittest
import random
from tconnectsync.sync.bolus import process_bolus_events
from tconnectsync.parser.tconnect import TConnectEntry
from tconnectsync.parser.nightscout import NightscoutEntry
from ..parser.test_tconnect import TestTConnectEntryBolus
from ..parser.test_tconnect import TestTConnectEntryBolus, TestTConnectEntryCGM, TestTConnectEntryReading
class TestBolusSync(unittest.TestCase):
@staticmethod
def get_example_csv_bolus_events():
return [
TestTConnectEntryBolus.entryStdCorrection,
TestTConnectEntryBolus.entryStd,
TestTConnectEntryBolus.entryStdAutomatic,
TestTConnectEntryBolus.entryStdIncompletePartial
]
def test_process_bolus_events_standard(self):
bolusData = [
TestTConnectEntryBolus.entryStdCorrection,
TestTConnectEntryBolus.entryStd,
TestTConnectEntryBolus.entryStdAutomatic
]
def test_process_bolus_events(self):
bolusData = TestBolusSync.get_example_csv_bolus_events()
bolusEvents = process_bolus_events(bolusData)
self.assertEqual(len(bolusEvents), 3)
self.assertEqual(len(bolusEvents), len(bolusData))
self.assertListEqual(bolusEvents, [
TConnectEntry.parse_bolus_entry(bolusData[0]),
TConnectEntry.parse_bolus_entry(bolusData[1]),
TConnectEntry.parse_bolus_entry(bolusData[2])
TConnectEntry.parse_bolus_entry(d) for d in bolusData
])
def test_process_bolus_events_cgmevents_not_matching(self):
bolusData = [
TestTConnectEntryBolus.entryStdCorrection,
TestTConnectEntryBolus.entryStd,
TestTConnectEntryBolus.entryStdAutomatic
]
cgmEvents = [
TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry1),
TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry2),
TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry3)
]
bolusEvents = process_bolus_events(bolusData, cgmEvents=cgmEvents)
self.assertEqual(len(bolusEvents), len(bolusData))
def set_bg_type(entry, type):
entry["bg_type"] = type
return entry
# Expect FINGER for bolus entries with a BG because there's no matching event with the same BG
expected = [
set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[0]), NightscoutEntry.FINGER),
set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[1]), NightscoutEntry.FINGER),
# No BG specified for the automatic bolus
TConnectEntry.parse_bolus_entry(bolusData[2])
]
self.assertListEqual(bolusEvents, expected)
def test_process_bolus_events_cgmevents_matches(self):
bolusData = [
TestTConnectEntryBolus.entryStdCorrection,
TestTConnectEntryBolus.entryStd,
TestTConnectEntryBolus.entryStdAutomatic
]
cgmEvents = [
{
"time": "2021-04-01 12:45:30-04:00",
"bg": "100",
"type": "EGV"
},
# Matches entryStdCorrection time but with wrong BG
{
"time": "2021-04-01 12:50:30-04:00",
"bg": "105",
"type": "EGV"
},
{
"time": "2021-04-01 13:00:30-04:00",
"bg": "110",
"type": "EGV"
},
{
"time": "2021-04-01 23:15:30-04:00",
"bg": "150",
"type": "EGV"
},
# Matches entryStd time with correct BG
{
"time": "2021-04-01 23:20:30-04:00",
"bg": "159",
"type": "EGV"
},
{
"time": "2021-04-01 23:25:30-04:00",
"bg": "160",
"type": "EGV"
},
]
bolusEvents = process_bolus_events(bolusData, cgmEvents=cgmEvents)
self.assertEqual(len(bolusEvents), len(bolusData))
def set_bg_type(entry, type):
entry["bg_type"] = type
return entry
expected = [
# Time found but BG doesn't match
set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[0]), NightscoutEntry.FINGER),
# Time found and BG matches
set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[1]), NightscoutEntry.SENSOR),
# No BG specified for the automatic bolus
TConnectEntry.parse_bolus_entry(bolusData[2])
]
self.assertListEqual(bolusEvents, expected)
def test_process_bolus_events_update_partial_description(self):
stdData = [
TestTConnectEntryBolus.entryStdCorrection,
TestTConnectEntryBolus.entryStd,
TestTConnectEntryBolus.entryStdAutomatic
]
partialData = [
TestTConnectEntryBolus.entryStdIncompletePartial
]
bolusData = stdData + partialData
bolusEvents = process_bolus_events(bolusData)
self.assertEqual(len(bolusEvents), len(bolusData))
partialEntries = [
TConnectEntry.parse_bolus_entry(e) for e in partialData
]
for e in partialEntries:
e["description"] += " (%s: requested %s units)" % (e["completion"], e["requested_insulin"])
self.assertListEqual(bolusEvents, [
TConnectEntry.parse_bolus_entry(d) for d in stdData
] + partialEntries)
def test_process_bolus_events_skip_zero(self):
stdData = [
TestTConnectEntryBolus.entryStdCorrection,
TestTConnectEntryBolus.entryStd,
TestTConnectEntryBolus.entryStdAutomatic
]
zeroData = [
TestTConnectEntryBolus.entryStdIncompleteZero
]
bolusData = stdData + zeroData
bolusEvents = process_bolus_events(bolusData)
self.assertEqual(len(bolusEvents), len(stdData))
self.assertListEqual(bolusEvents, [
TConnectEntry.parse_bolus_entry(d) for d in stdData
])
for d in zeroData:
self.assertNotIn(TConnectEntry.parse_bolus_entry(d), bolusEvents)
if __name__ == '__main__':
unittest.main()
+51
View File
@@ -0,0 +1,51 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.sync.cgm import find_event_at, process_cgm_events
from tconnectsync.parser.tconnect import TConnectEntry
from ..parser.test_tconnect import TestTConnectEntryReading
class TestProcessCGMEvents(unittest.TestCase):
def test_process_cgm_events(self):
rawReadings = [
TestTConnectEntryReading.entry1,
TestTConnectEntryReading.entry2,
TestTConnectEntryReading.entry3,
TestTConnectEntryReading.entry4
]
self.assertListEqual(
process_cgm_events(rawReadings),
[TConnectEntry.parse_reading_entry(r) for r in rawReadings]
)
class TestFindEventAt(unittest.TestCase):
readingData = [
TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry1),
TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry2),
TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry3),
TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry4)
]
def test_find_event_at_exact(self):
for r in self.readingData:
self.assertEqual(find_event_at(self.readingData, r["time"]), r)
def test_find_event_at_before_not_found(self):
self.assertEqual(find_event_at(self.readingData, "2021-10-22 10:30:00-04:00"), None)
def test_find_event_at_large_gap(self):
self.assertEqual(find_event_at(self.readingData, "2021-10-23 13:30:00-04:00"), self.readingData[0])
def test_find_event_at_between_close(self):
self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:17:52-04:00"), self.readingData[1])
self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:21:52-04:00"), self.readingData[2])
self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:25:59-04:00"), self.readingData[3])
def test_find_event_at_most_recent(self):
self.assertEqual(find_event_at(self.readingData, "2021-10-23 18:00:00-04:00"), self.readingData[3])
if __name__ == '__main__':
unittest.main()
+30
View File
@@ -0,0 +1,30 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.sync.iob import process_iob_events
from tconnectsync.parser.tconnect import TConnectEntry
from ..parser.test_tconnect import TestTConnectEntryIOB
class TestIOBSync(unittest.TestCase):
@staticmethod
def get_example_csv_iob_events():
return [
TestTConnectEntryIOB.entry1,
TestTConnectEntryIOB.entry2,
]
def test_process_iob_events(self):
iobData = TestIOBSync.get_example_csv_iob_events()
iobEvents = process_iob_events(iobData)
self.assertEqual(len(iobEvents), len(iobData))
self.assertListEqual(iobEvents, [
TConnectEntry.parse_iob_entry(d) for d in iobData
])
if __name__ == '__main__':
unittest.main()
+518 -15
View File
@@ -3,20 +3,23 @@
import unittest
import datetime
import pprint
import copy
from tconnectsync.process import process_time_range
from tconnectsync.parser.nightscout import NightscoutEntry
from tconnectsync.parser.nightscout import EXERCISE_EVENTTYPE, IOB_ACTIVITYTYPE, SLEEP_EVENTTYPE, NightscoutEntry
from tconnectsync.features import BASAL, BOLUS, IOB, PUMP_EVENTS
from .api.fake import TConnectApi
from .nightscout_fake import NightscoutApi
from .sync.test_basal import TestBasalSync
from .sync.test_bolus import TestBolusSync
from .sync.test_iob import TestIOBSync
class TestProcessTimeRange(unittest.TestCase):
maxDiff = None
def stub_therapy_timeline(self, time_start, time_end):
pass
return copy.deepcopy(TestBasalSync.base)
def stub_therapy_timeline_csv(self, time_start, time_end):
return {
@@ -25,6 +28,9 @@ class TestProcessTimeRange(unittest.TestCase):
"basalData": [],
"bolusData": []
}
def stub_ws2_basalsuspension(self, time_start, time_end):
return {"BasalSuspension": []}
def stub_last_uploaded_entry(self, event_type):
return None
@@ -36,6 +42,7 @@ class TestProcessTimeRange(unittest.TestCase):
def test_new_ciq_basal_data(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
@@ -53,7 +60,7 @@ class TestProcessTimeRange(unittest.TestCase):
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 4)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
@@ -61,16 +68,44 @@ class TestProcessTimeRange(unittest.TestCase):
NightscoutEntry.basal(0.8, 20.35, "2021-03-16 00:00:00-04:00", reason="tempDelivery"),
NightscoutEntry.basal(0.799, 5.0, "2021-03-16 00:20:21-04:00", reason="profileDelivery"),
NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"),
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery")
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery (control-iq suspension)")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertDictEqual(nightscout.deleted_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""No data in Nightscout. Nothing should be updated in Nightscout without the BASAL feature."""
def test_basal_data_not_updated_without_feature(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
def fake_therapy_timeline(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return TestBasalSync.get_example_ciq_basal_events()
tconnect.controliq.therapy_timeline = fake_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, IOB])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0)
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""Two basal entries in Nightscout. Two new basal entries in tconnect."""
def test_partial_ciq_basal_data(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
@@ -95,16 +130,16 @@ class TestProcessTimeRange(unittest.TestCase):
nightscout.last_uploaded_entry = fake_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"),
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery")
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery (control-iq suspension)")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertDictEqual(nightscout.deleted_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""
@@ -113,6 +148,7 @@ class TestProcessTimeRange(unittest.TestCase):
def test_with_updated_duration_ciq_basal_data(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
@@ -138,13 +174,13 @@ class TestProcessTimeRange(unittest.TestCase):
nightscout.last_uploaded_entry = fake_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2)
self.assertDictEqual(nightscout.uploaded_entries, {
"treatments": [
NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"),
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery")
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery (control-iq suspension)")
]})
self.assertEqual(len(nightscout.put_entries["treatments"]), 1)
self.assertDictEqual(dict(nightscout.put_entries), {
@@ -155,21 +191,23 @@ class TestProcessTimeRange(unittest.TestCase):
}
]
})
self.assertDictEqual(nightscout.deleted_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""No data in Nightscout. Uploads all bolus data from tconnect."""
def test_new_ciq_bolus_data(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
bolusData = TestBolusSync.get_example_csv_bolus_events()
def fake_therapy_timeline_csv(time_start, time_end):
return {
**self.stub_therapy_timeline_csv(time_start, time_end),
"bolusData": TestBolusSync.get_example_csv_bolus_events(),
"bolusData": bolusData,
}
tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv
@@ -179,19 +217,484 @@ class TestProcessTimeRange(unittest.TestCase):
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL])
pprint.pprint(nightscout.uploaded_entries)
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 3)
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), len(bolusData))
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.bolus(13.53, 75, "2021-04-01 12:58:26-04:00", notes="Standard/Correction"),
NightscoutEntry.bolus(1.25, 0, "2021-04-01 23:23:17-04:00", notes="Standard (Override)"),
NightscoutEntry.bolus(1.7, 0, "2021-04-02 01:00:47-04:00", notes="Automatic Bolus/Correction"),
NightscoutEntry.bolus(1.82, 0, "2021-09-06 12:24:47-04:00", notes="Standard/Correction (Terminated by Alarm: requested 2.63 units)"),
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertDictEqual(nightscout.deleted_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""No data in Nightscout. Nothing should be updated in Nightscout without the BOLUS feature."""
def test_bolus_data_not_updated_without_feature(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
bolusData = TestBolusSync.get_example_csv_bolus_events()
def fake_therapy_timeline_csv(time_start, time_end):
return {
**self.stub_therapy_timeline_csv(time_start, time_end),
"bolusData": bolusData,
}
tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BASAL, IOB])
pprint.pprint(nightscout.uploaded_entries)
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0)
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""No data in Nightscout. Uploads new iob reading from tconnect."""
def test_new_ciq_iob_data(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
iobData = TestIOBSync.get_example_csv_iob_events()
def fake_therapy_timeline_csv(time_start, time_end):
return {
**self.stub_therapy_timeline_csv(time_start, time_end),
"iobData": iobData,
}
tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL, IOB])
pprint.pprint(nightscout.uploaded_entries)
self.assertEqual(len(nightscout.uploaded_entries["activity"]), 1)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"activity": [
# the most recent IOB entry is added
NightscoutEntry.iob(6.80, "2021-10-12 00:10:30-04:00")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""No data in Nightscout. Nothing should be updated in Nightscout without the IOB feature."""
def test_iob_data_not_updated_without_feature(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
iobData = TestIOBSync.get_example_csv_iob_events()
def fake_therapy_timeline_csv(time_start, time_end):
return {
**self.stub_therapy_timeline_csv(time_start, time_end),
"iobData": iobData,
}
tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BASAL, BOLUS])
self.assertEqual(len(nightscout.uploaded_entries["activity"]), 0)
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""Existing IOB in Nightscout. Uploads new iob reading and deletes old IOB."""
def test_updates_ciq_iob_data(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
iobData = TestIOBSync.get_example_csv_iob_events()
iobData[0]["created_at"] = start
iobData[0]["_id"] = "sentinel_existing_iob_id"
def fake_therapy_timeline_csv(time_start, time_end):
return {
**self.stub_therapy_timeline_csv(time_start, time_end),
"iobData": iobData,
}
tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
def fake_last_uploaded_activity(activityType):
if activityType == IOB_ACTIVITYTYPE:
return iobData[0]
return self.stub_last_uploaded_activity(activityType)
nightscout.last_uploaded_activity = fake_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[IOB])
pprint.pprint(nightscout.uploaded_entries)
self.assertEqual(len(nightscout.uploaded_entries["activity"]), 1)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"activity": [
# the most recent IOB entry is added
NightscoutEntry.iob(6.80, "2021-10-12 00:10:30-04:00")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [
"activity/sentinel_existing_iob_id"
])
"""No pump activity events in Nightscout. New CIQ activity events."""
def test_new_ciq_activity_events(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 5, 1, 0, 0)
end = datetime.datetime(2021, 5, 3, 0, 0)
def fake_therapy_timeline(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return {
**TestBasalSync.base,
"events": [{
"duration": 1200,
"eventType": 2, # Exercise
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619901912 # 2021-05-01 13:45:12-04:00
}, {
"duration": 30661,
"eventType": 1, # Sleep
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619992000 # 2021-05-02 14:46:40-04:00
}]
}
tconnect.controliq.therapy_timeline = fake_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
tconnect.ws2.basalsuspension = self.stub_ws2_basalsuspension
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.activity(created_at="2021-05-01 13:45:12-04:00", duration=20, reason="Exercise", event_type=EXERCISE_EVENTTYPE),
NightscoutEntry.activity(created_at="2021-05-02 14:46:40-04:00", duration=30661/60, reason="Sleep", event_type=SLEEP_EVENTTYPE),
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""No pump activity events in Nightscout. New CIQ activity events, but feature is disabled."""
def test_no_ciq_activity_events_without_feature(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 5, 1, 0, 0)
end = datetime.datetime(2021, 5, 3, 0, 0)
def fake_therapy_timeline(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return {
**TestBasalSync.base,
"events": [{
"duration": 1200,
"eventType": 2, # Exercise
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619901912 # 2021-05-01 13:45:12-04:00
}, {
"duration": 30661,
"eventType": 1, # Sleep
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619992000 # 2021-05-02 14:46:40-04:00
}]
}
tconnect.controliq.therapy_timeline = fake_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
tconnect.ws2.basalsuspension = self.stub_ws2_basalsuspension
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL, IOB])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0)
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""
Existing Sleep event in Nightscout with shorter duration than current, as well as a past Exercise event.
Ensures that the old sleep event is deleted and a new one is created with the correct duration."""
def test_existing_ciq_activity_events(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 5, 1, 0, 0)
end = datetime.datetime(2021, 5, 3, 0, 0)
def fake_therapy_timeline(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return {
**TestBasalSync.base,
"events": [{
"duration": 1200,
"eventType": 2, # Exercise
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619901912 # 2021-05-01 13:45:12-04:00
}, {
"duration": 4200, # Currently 60 mins (3600), changing to 70 mins (4200)
"eventType": 1, # Sleep
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619992000 # 2021-05-02 14:46:40-04:00
}]
}
tconnect.controliq.therapy_timeline = fake_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
tconnect.ws2.basalsuspension = self.stub_ws2_basalsuspension
nightscout = NightscoutApi()
def fake_last_uploaded_entry(event_type):
if event_type == "Sleep":
return {
"created_at": "2021-05-02 14:46:40-04:00",
"duration": 60,
"_id": "old_sleep"
}
elif event_type == "Exercise":
return {
"created_at": "2021-05-01 13:45:12-04:00",
"duration": 20,
"_id": "exercise"
}
return self.stub_last_uploaded_entry()
nightscout.last_uploaded_entry = fake_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 1)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
# Already exists:
# NightscoutEntry.activity(created_at="2021-05-01 13:45:12-04:00", duration=20, reason="Exercise", event_type=EXERCISE_EVENTTYPE),
# Updated event duration:
NightscoutEntry.activity(created_at="2021-05-02 14:46:40-04:00", duration=70, reason="Sleep", event_type=SLEEP_EVENTTYPE),
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertEqual(len(nightscout.deleted_entries), 1)
self.assertListEqual(nightscout.deleted_entries, [
"treatments/old_sleep"
])
"""No pump activity events in Nightscout. New WS2 activity events."""
def test_new_ws2_activity_events(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 5, 1, 0, 0)
end = datetime.datetime(2021, 5, 3, 0, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
def fake_basalsuspension(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return {"BasalSuspension": [
{
'EventDateTime': '/Date(1638663490000-0000)/',
'SuspendReason': 'site-cart'
},
{
'EventDateTime': '/Date(1637863616000-0000)/',
'SuspendReason': 'alarm'
},
{
'EventDateTime': '/Date(1638662852000-0000)/',
'SuspendReason': 'manual'
}
]}
tconnect.ws2.basalsuspension = fake_basalsuspension
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 3)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.sitechange(created_at="2021-12-04 16:18:10-05:00", reason="Site/Cartridge Change"),
NightscoutEntry.basalsuspension(created_at="2021-11-25 10:06:56-05:00", reason="Empty Cartridge/Pump Shutdown"),
NightscoutEntry.basalsuspension(created_at="2021-12-04 16:07:32-05:00", reason="User Suspended")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""Existing pump activity events in Nightscout. New WS2 activity events. Only adds new events."""
def test_existing_ws2_activity_events(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 5, 1, 0, 0)
end = datetime.datetime(2021, 5, 3, 0, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
def fake_basalsuspension(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return {"BasalSuspension": [
{
'EventDateTime': '/Date(1638663490000-0000)/',
'SuspendReason': 'site-cart'
},
{
'EventDateTime': '/Date(1637863616000-0000)/',
'SuspendReason': 'alarm'
},
{
'EventDateTime': '/Date(1638662852000-0000)/',
'SuspendReason': 'manual'
},
# This event is new:
{
'EventDateTime': '/Date(1638672852000-0000)/',
'SuspendReason': 'manual'
}
]}
tconnect.ws2.basalsuspension = fake_basalsuspension
nightscout = NightscoutApi()
def fake_last_uploaded_entry(event_type):
if event_type == "Site Change":
return {
"created_at": "2021-12-04 16:18:10-05:00"
}
elif event_type == "Basal Suspension":
return {
"created_at": "2021-12-04 16:07:32-05:00"
}
return self.stub_last_uploaded_entry()
nightscout.last_uploaded_entry = fake_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 1)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.basalsuspension(created_at="2021-12-04 18:54:12-05:00", reason="User Suspended")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
"""No pump activity events in nightscout. New WS2 activity events, but only of skipped types. None should be added."""
def test_skipped_ws2_activity_events(self):
tconnect = TConnectApi()
# datetimes are unused by the API fake
start = datetime.datetime(2021, 5, 1, 0, 0)
end = datetime.datetime(2021, 5, 3, 0, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
def fake_basalsuspension(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return {"BasalSuspension": [
{
'EventDateTime': '/Date(1638659343000-0000)/',
'SuspendReason': 'basal-profile',
},
{
'Continuation': 'continuation',
'EventDateTime': '/Date(1638604800000-0000)/',
'SuspendReason': 'previous',
},
{
'EventDateTime': '/Date(1638659343000-0000)/',
'SuspendReason': 'basal-profile',
}
]}
tconnect.ws2.basalsuspension = fake_basalsuspension
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS])
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0)
self.assertDictEqual(nightscout.put_entries, {})
self.assertListEqual(nightscout.deleted_entries, [])
if __name__ == '__main__':
+99
View File
@@ -0,0 +1,99 @@
#!/usr/bin/env python3
import unittest
import unittest.mock
import tempfile
import importlib
import contextlib
import pathlib
import os
@contextlib.contextmanager
def chdir(dir):
orig_cwd = os.getcwd()
os.chdir(dir)
try:
yield
finally:
os.chdir(orig_cwd)
class TestSecretDotEnv(unittest.TestCase):
maxDiff = None
def write_test_dotenv_file(self, path, type):
with open(os.path.join(path, ".env"), "w") as f:
f.write("""
TCONNECT_EMAIL=test_%s_email@email.com
NS_URL=http://test_%s_url
""" % (type, type))
f.close()
def import_secret(self):
return importlib.reload(importlib.import_module("tconnectsync.secret"))
def test_dotenv_in_current_working_directory(self):
with tempfile.TemporaryDirectory(prefix='dotenv_cwd') as dir, chdir(dir):
self.write_test_dotenv_file(dir, "dotenv_cwd")
secret = self.import_secret()
self.assertEqual(secret.TCONNECT_EMAIL, "test_dotenv_cwd_email@email.com")
self.assertEqual(secret.NS_URL, "http://test_dotenv_cwd_url")
def test_dotenv_in_homedir_config_folder(self):
with tempfile.TemporaryDirectory(prefix='dotenv_homedir_config') as dir, chdir(dir):
config_dir = os.path.join(dir, '.config/tconnectsync')
os.makedirs(config_dir)
self.write_test_dotenv_file(config_dir, "dotenv_homedir_config")
with unittest.mock.patch.object(pathlib.Path, "home") as mock_home:
mock_home.return_value = dir
secret = self.import_secret()
self.assertEqual(secret.TCONNECT_EMAIL, "test_dotenv_homedir_config_email@email.com")
self.assertEqual(secret.NS_URL, "http://test_dotenv_homedir_config_url")
def test_no_dotenv_file_reads_from_environment(self):
with tempfile.TemporaryDirectory(prefix='dotenv_environ') as dir, chdir(dir):
environ = {
"TCONNECT_EMAIL": "test_environ_email@email.com",
"NS_URL": "http://test_environ_url"
}
with unittest.mock.patch.dict(os.environ, environ):
secret = self.import_secret()
self.assertEqual(secret.TCONNECT_EMAIL, environ["TCONNECT_EMAIL"])
self.assertEqual(secret.NS_URL, environ["NS_URL"])
def test_environment_merges_with_dotenv_file(self):
with tempfile.TemporaryDirectory(prefix='env_plus_environ') as dir, chdir(dir):
self.write_test_dotenv_file(dir, "dotenv_cwd_defaults")
environ = {
"TCONNECT_EMAIL": "test_environ_override_email@email.com"
}
with unittest.mock.patch.dict(os.environ, environ):
secret = self.import_secret()
self.assertEqual(secret.TCONNECT_EMAIL, "test_environ_override_email@email.com")
self.assertEqual(secret.NS_URL, "http://test_dotenv_cwd_defaults_url")
def test_dotenv_in_current_working_directory_overrides_homedir_config(self):
with tempfile.TemporaryDirectory(prefix='dotenv_cwd') as dir, chdir(dir):
self.write_test_dotenv_file(dir, "dotenv_cwd")
config_dir = os.path.join(dir, '.config/tconnectsync')
os.makedirs(config_dir)
self.write_test_dotenv_file(config_dir, "dotenv_homedir_config")
with unittest.mock.patch.object(pathlib.Path, "home") as mock_home:
mock_home.return_value = dir
secret = self.import_secret()
self.assertEqual(secret.TCONNECT_EMAIL, "test_dotenv_cwd_email@email.com")
self.assertEqual(secret.NS_URL, "http://test_dotenv_cwd_url")
if __name__ == '__main__':
unittest.main()