forked from bewest/openaps-example
-
Notifications
You must be signed in to change notification settings - Fork 0
/
agp.py
145 lines (130 loc) · 4.3 KB
/
agp.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
"""
AGP - calculate agp values given some glucose text
"""
import dateutil.parser
class AGP (object):
"""
The actual calculator.
"""
def __init__ (self):
# init empty values
self.values = []
# and empty buckets
self.hour_buckets = {}
# initialize all buckets with empty list
for hour in range(0,24):
self.hour_buckets[hour] = []
def add_record (self, record):
""" Add record to global list and assign to bucket
"""
# clean/prep the record
(time, glucose, trend) = record.strip().split()
# get a proper datetime object
datetime = dateutil.parser.parse(time)
glucose = int(glucose)
# ignore special values < 40
if (glucose >= 39):
# print datetime.hour, glucose, trend
# append to internal list
self.values.append((datetime, glucose, trend))
# assign to bucket
bucket = self.hour_buckets.get(datetime.hour, [])
bucket.append(glucose)
# process data and return new agp stats
def __call__ (self, data):
stats = [ ]
# add all records
for record in data:
self.add_record(record)
# calculate for stats each hour of day
for hour in range(0,24):
stats.append((hour, calc_agp(self.hour_buckets[hour])))
return stats
def calc_agp (bucket):
vals_sorted = sorted(bucket)
# print vals_sorted, bucket
percentile_10 = []
median = []
percentile_25 = []
percentile_75 = []
percentile_90 = []
if len(vals_sorted) > 0:
percentile_10 = vals_sorted[int(len(vals_sorted)*.1)]
median = vals_sorted[int(len(vals_sorted)/2)]
percentile_25 = vals_sorted[int(len(vals_sorted)*.25)]
percentile_75 = vals_sorted[int(len(vals_sorted)*.75)]
percentile_90 = vals_sorted[int(len(vals_sorted)*.9)]
return (percentile_10, percentile_25, median, percentile_75, percentile_90)
##########################################
#
# openaps vendor example:
# The following shows what is needed to make the module available as a vendor
# plugin to openaps.
#
# Inherit from openaps.uses.use.Use class
from openaps.uses.use import Use
class agp (Use):
""" Calculate agp
"""
# get_params helps openaps save your configurations
def get_params (self, args):
"""
Create a dict data type from args namespace so that config serializer can
save these args for report generation.
"""
return dict(input=args.input)
# configure_app allows your plugin to specify command line parameters
def configure_app (self, app, parser):
"""
Set up any arguments needed by this use.
"""
# get file based argument called input.
parser.add_argument('input', default='glucose.txt')
# main logic for the app
def main (self, args, app):
"""
Main logic for calculating agp
"""
# print args
# get parameters
params = self.get_params(args)
# print params.get('input')
# create calculator
parser = AGP( )
with open(params.get('input'), 'r') as f:
# calculate agp for all input
return parser(f.readlines())
# set_config is needed by openaps for all vendors.
# set_config is used by `device add` commands so save any needed
# information.
# See the medtronic builtin module for an example of how to use this
# to save needed information to establish sessions (serial numbers,
# etc).
def set_config (args, device):
# no special config
return
# display_device allows our custom vendor implementation to include
# special information when displaying information about a device using
# our plugin as a vendor.
def display_device (device):
# no special information needed to run
return ''
# openaps calls get_uses to figure out how how to use a device using
# agp as a vendor. Return a list of classes which inherit from Use,
# or are compatible with it:
def get_uses (device, config):
# make agp an openaps use command
return [ agp ]
######################################################
# openaps definitions are complete
######################################################
# The remainder is for debugging and testing purposes.
# This allows running the module from commandline without openaps.
# this uses no openaps logic, and is useful for debugging/testing
# this only runs when run as:
# $ python agp.py
if __name__ == '__main__':
parser = AGP( )
with open("glucose.txt") as f:
for hour, vals in parser(f.readlines()):
print hour, vals