<s> """<STR_LIT>""" <EOL> from openaps . uses . use import Use <EOL> from openaps . uses . registry import Registry <EOL> from openaps . configurable import Configurable <EOL> from openaps . glucose . convert import Convert as GlucoseConvert <EOL> import decocare <EOL> import argparse <EOL> import json <EOL> from decocare import stick , session , link , commands , history <EOL> from datetime import datetime <EOL> from dateutil import relativedelta <EOL> from dateutil . parser import parse <EOL> def configure_use_app ( app , parser ) : <EOL> pass <EOL> def configure_add_app ( app , parser ) : <EOL> parser . add_argument ( '<STR_LIT>' ) <EOL> def configure_app ( app , parser ) : <EOL> if app . parent . name == '<STR_LIT>' : <EOL> """<STR_LIT>""" <EOL> def configure_parser ( parser ) : <EOL> pass <EOL> def main ( args , app ) : <EOL> """<STR_LIT>""" <EOL> use = Registry ( ) <EOL> @ use ( ) <EOL> class scan ( Use ) : <EOL> """<STR_LIT>""" <EOL> def configure_app ( self , app , parser ) : <EOL> pass <EOL> def scanner ( self ) : <EOL> from decocare . scan import scan <EOL> return scan ( ) <EOL> def main ( self , args , app ) : <EOL> return self . scanner ( ) <EOL> import logging <EOL> import logging . handlers <EOL> class MedtronicTask ( scan ) : <EOL> MAX_SESSION_DURATION = <NUM_LIT:10> <EOL> requires_session = True <EOL> save_session = True <EOL> record_stats = True <EOL> def before_main ( self , args , app ) : <EOL> self . setup_medtronic ( ) <EOL> if self . requires_session : <EOL> self . check_session ( app ) <EOL> else : <EOL> self . pump . setModel ( number = self . device . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def after_main ( self , args , app ) : <EOL> if self . save_session : <EOL> self . device . store ( app . config ) <EOL> app . config . save ( ) <EOL> if self . uart : <EOL> self . uart . close ( ) <EOL> def get_session_info ( self ) : <EOL> expires = self . device . get ( '<STR_LIT>' , None ) <EOL> if expires is not None : <EOL> expires = parse ( expires ) <EOL> now = datetime . now ( ) <EOL> out = dict ( device = self . device . name <EOL> , vendor = __name__ <EOL> , used = now <EOL> ) <EOL> if expires is None or expires < now or ( expires - now ) . total_seconds ( ) > ( <NUM_LIT> * self . MAX_SESSION_DURATION ) : <EOL> fields = self . create_session ( ) <EOL> out . update ( ** self . update_session_info ( fields ) ) <EOL> else : <EOL> out [ '<STR_LIT>' ] = expires <EOL> out [ '<STR_LIT>' ] = self . device . get ( '<STR_LIT>' , None ) <EOL> return out <EOL> def update_session_info ( self , fields ) : <EOL> out = { } <EOL> uses_extra = self . device . get ( '<STR_LIT>' , None ) <EOL> config = self . device <EOL> if uses_extra : <EOL> config = self . device . extra <EOL> config . add_option ( '<STR_LIT>' , fields [ '<STR_LIT>' ] . isoformat ( ) ) <EOL> config . add_option ( '<STR_LIT>' , fields [ '<STR_LIT>' ] ) <EOL> out [ '<STR_LIT>' ] = fields [ '<STR_LIT>' ] <EOL> out [ '<STR_LIT>' ] = fields [ '<STR_LIT>' ] <EOL> return out <EOL> def create_session ( self ) : <EOL> minutes = int ( self . device . get ( '<STR_LIT>' , <NUM_LIT:3> ) ) <EOL> now = datetime . now ( ) <EOL> self . pump . power_control ( minutes = minutes ) <EOL> model = self . get_model ( ) <EOL> offset = relativedelta . relativedelta ( minutes = minutes ) + relativedelta . relativedelta ( minutes = - <NUM_LIT:1> ) <EOL> out = dict ( device = self . device . name <EOL> , model = model <EOL> , vendor = __name__ <EOL> , created_at = now <EOL> , started = now <EOL> , expires = now + offset <EOL> ) <EOL> return out <EOL> def check_session ( self , app ) : <EOL> self . session = self . get_session_info ( ) <EOL> model = self . device . get ( '<STR_LIT>' , None ) <EOL> if model is None : <EOL> model = self . get_model ( ) <EOL> self . pump . setModel ( number = self . device . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> uses_extra = self . device . get ( '<STR_LIT>' , None ) <EOL> config = self . device <EOL> if uses_extra : <EOL> config = self . device . extra <EOL> config . add_option ( '<STR_LIT>' , self . device . get ( '<STR_LIT>' , model ) ) <EOL> def get_model ( self ) : <EOL> model = self . pump . read_model ( ) . getData ( ) <EOL> return model <EOL> def setup_medtronic ( self ) : <EOL> log = logging . getLogger ( decocare . __name__ ) <EOL> level = getattr ( logging , self . device . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> address = self . device . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> log . setLevel ( level ) <EOL> for previous in log . handlers [ : ] : <EOL> log . removeHandler ( previous ) <EOL> log . addHandler ( logging . handlers . SysLogHandler ( address = address ) ) <EOL> self . uart = stick . Stick ( link . Link ( self . scanner ( ) ) ) <EOL> self . uart . open ( ) <EOL> serial = self . device . get ( '<STR_LIT>' ) <EOL> self . pump = session . Pump ( self . uart , serial ) <EOL> stats = self . uart . interface_stats ( ) <EOL> def main ( self , args , app ) : <EOL> return self . scanner ( ) <EOL> class Session ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> requires_session = False <EOL> def configure_parser ( self , parser ) : <EOL> parser . add_argument ( '<STR_LIT>' , type = int , default = '<STR_LIT>' ) <EOL> def setup_application ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> def main ( self , args , app ) : <EOL> info = self . create_session ( ) <EOL> info . update ( ** self . update_session_info ( info ) ) <EOL> return info <EOL> @ use ( ) <EOL> class model ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def configure_app ( self , app , parser ) : <EOL> pass <EOL> def main ( self , args , app ) : <EOL> model = self . pump . read_model ( ) . getData ( ) <EOL> return model <EOL> @ use ( ) <EOL> class read_status ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def main ( self , args , app ) : <EOL> return self . pump . model . read_status ( ) <EOL> @ use ( ) <EOL> class status ( read_status ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class reservoir ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def main ( self , args , app ) : <EOL> return self . pump . model . read_reservoir ( ) <EOL> @ use ( ) <EOL> class settings ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def main ( self , args , app ) : <EOL> return self . pump . model . read_settings ( ) <EOL> @ use ( ) <EOL> class mytest ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> requires_session = False <EOL> def main ( self , args , app ) : <EOL> return self . pump . model . my_read_settings ( ) <EOL> @ use ( ) <EOL> class read_clock ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def main ( self , args , app ) : <EOL> return self . pump . model . read_clock ( ) <EOL> class SameNameCommand ( MedtronicTask ) : <EOL> def main ( self , args , app ) : <EOL> name = self . __class__ . __name__ . split ( '<STR_LIT:.>' ) . pop ( ) <EOL> return getattr ( self . pump . model , name ) ( ** self . get_params ( args ) ) <EOL> class SelectedNameCommand ( MedtronicTask ) : <EOL> def main ( self , args , app ) : <EOL> name = self . selected <EOL> return getattr ( self . pump . model , name ) ( ** self . get_params ( args ) ) <EOL> @ use ( ) <EOL> class read_temp_basal ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_settings ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_carb_ratios ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_basal_profile_std ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_basal_profile_A ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_basal_profile_B ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_selected_basal_profile ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_current_glucose_pages ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_current_history_pages ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class suspend_pump ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> def main ( self , args , app ) : <EOL> result = super ( suspend_pump , self ) . main ( args , app ) <EOL> result . update ( enacted_at = datetime . now ( ) ) <EOL> return result <EOL> @ use ( ) <EOL> class resume_pump ( suspend_pump ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_battery_status ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_bg_targets ( SelectedNameCommand ) : <EOL> """<STR_LIT>""" <EOL> selected = '<STR_LIT>' <EOL> @ use ( ) <EOL> class read_insulin_sensitivies ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_insulin_sensitivities ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_glucose_data ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> def configure_app ( self , app , parser ) : <EOL> parser . add_argument ( '<STR_LIT>' , type = int , default = <NUM_LIT:0> ) <EOL> def get_params ( self , args ) : <EOL> return dict ( page = int ( args . page ) ) <EOL> class InputProgramRequired ( MedtronicTask ) : <EOL> def upload_program ( self , program ) : <EOL> raise NotImplementedError ( ) <EOL> def get_params ( self , args ) : <EOL> return dict ( input = args . input ) <EOL> def configure_app ( self , app , parser ) : <EOL> parser . add_argument ( '<STR_LIT:input>' , default = '<STR_LIT:->' ) <EOL> def get_program ( self , args ) : <EOL> params = self . get_params ( args ) <EOL> program = json . load ( argparse . FileType ( '<STR_LIT:r>' ) ( params . get ( '<STR_LIT:input>' ) ) ) <EOL> return program <EOL> def main ( self , args , app ) : <EOL> program = self . get_program ( args ) <EOL> results = self . upload_program ( program ) <EOL> program . update ( timestamp = datetime . now ( ) , ** results ) <EOL> return program <EOL> def parse_clock ( candidate ) : <EOL> if candidate . lower ( ) in [ '<STR_LIT>' ] : <EOL> return datetime . now ( ) <EOL> return parse ( candidate ) <EOL> @ use ( ) <EOL> class set_clock ( InputProgramRequired ) : <EOL> """<STR_LIT>""" <EOL> def get_params ( self , args ) : <EOL> return dict ( input = args . input , to = args . to ) <EOL> def configure_app ( self , app , parser ) : <EOL> parser . add_argument ( '<STR_LIT:input>' , nargs = '<STR_LIT:?>' , default = None ) <EOL> parser . add_argument ( '<STR_LIT>' , default = None ) <EOL> def upload_program ( self , program ) : <EOL> if not program . get ( '<STR_LIT>' , None ) : <EOL> print "<STR_LIT>" <EOL> raise Exception ( "<STR_LIT>" . format ( program . get ( '<STR_LIT>' ) ) ) <EOL> return self . pump . model . set_clock ( ** program ) <EOL> def get_program ( self , args ) : <EOL> params = self . get_params ( args ) <EOL> program = dict ( clock = None ) <EOL> if params . get ( '<STR_LIT:to>' , None ) is None and params . get ( '<STR_LIT:input>' ) : <EOL> program . update ( clock = parse ( json . load ( argparse . FileType ( '<STR_LIT:r>' ) ( params . get ( '<STR_LIT:input>' ) ) ) ) ) <EOL> else : <EOL> if params . get ( '<STR_LIT:to>' ) : <EOL> program . update ( clock = parse_clock ( params . get ( '<STR_LIT:to>' ) ) ) <EOL> return program <EOL> @ use ( ) <EOL> class set_temp_basal ( InputProgramRequired ) : <EOL> """<STR_LIT>""" <EOL> required_inputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> def upload_program ( self , program ) : <EOL> missing = [ ] <EOL> for req in self . required_inputs : <EOL> if not req in program : <EOL> missing . append ( req ) <EOL> if len ( missing ) > <NUM_LIT:0> : <EOL> return dict ( error = "<STR_LIT>" , missing = missing , input = dict ( ** program ) ) <EOL> return self . pump . model . set_temp_basal ( ** program ) <EOL> @ use ( ) <EOL> class bolus ( InputProgramRequired ) : <EOL> """<STR_LIT>""" <EOL> def upload_program ( self , program ) : <EOL> return self . pump . model . bolus ( ** program ) <EOL> @ use ( ) <EOL> class filter_glucose_date ( SameNameCommand ) : <EOL> """<STR_LIT>""" <EOL> def get_params ( self , args ) : <EOL> return dict ( begin = args . begin , end = args . end ) <EOL> def configure_app ( self , app , parser ) : <EOL> parser . add_argument ( '<STR_LIT>' , default = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT:end>' , default = '<STR_LIT>' ) <EOL> @ use ( ) <EOL> class filter_isig_date ( filter_glucose_date ) : <EOL> """<STR_LIT>""" <EOL> @ use ( ) <EOL> class read_history_data ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def get_params ( self , args ) : <EOL> return dict ( page = int ( args . page ) ) <EOL> def configure_app ( self , app , parser ) : <EOL> parser . add_argument ( '<STR_LIT>' , type = int , default = <NUM_LIT:0> ) <EOL> def main ( self , args , app ) : <EOL> history = self . pump . model . read_history_data ( ** self . get_params ( args ) ) <EOL> return history <EOL> @ use ( ) <EOL> class iter_glucose ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def get_params ( self , args ) : <EOL> return dict ( count = int ( args . count ) ) <EOL> def configure_app ( self , app , parser ) : <EOL> parser . add_argument ( '<STR_LIT:count>' , type = int , default = <NUM_LIT> ) <EOL> def range ( self ) : <EOL> return self . pump . model . iter_glucose_pages ( ) <EOL> maxCount = <NUM_LIT> <EOL> def main ( self , args , app ) : <EOL> self . maxCount = self . get_params ( args ) . get ( '<STR_LIT:count>' , self . maxCount ) <EOL> num = <NUM_LIT:0> <EOL> records = [ ] <EOL> for rec in self . range ( ) : <EOL> records . append ( rec ) <EOL> num = num + <NUM_LIT:1> <EOL> if num > self . maxCount : <EOL> break <EOL> return records <EOL> @ use ( ) <EOL> class iter_pump ( iter_glucose ) : <EOL> """<STR_LIT>""" <EOL> def range ( self ) : <EOL> return self . pump . model . iter_history_pages ( ) <EOL> @ use ( ) <EOL> class iter_glucose_hours ( MedtronicTask ) : <EOL> """<STR_LIT>""" <EOL> def get_params ( self , args ) : <EOL> params = dict ( hours = float ( args . hours ) ) <EOL> if '<STR_LIT>' in args and args . now is not None : <EOL> params . update ( now = args . now ) <EOL> return params <EOL> def configure_app ( self , app , parser ) : <EOL> parser . add_argument ( '<STR_LIT>' , type = float , help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , help = '<STR_LIT>' ) <EOL> def range ( self ) : <EOL> return self . pump . model . iter_glucose_pages ( ) <EOL> def get_record_timestamp ( self , record ) : <EOL> return parse ( record [ '<STR_LIT:date>' ] ) if '<STR_LIT:date>' in record else None <EOL> def main ( self , args , app ) : <EOL> params = self . get_params ( args ) <EOL> min_offset = relativedelta . relativedelta ( hours = params [ '<STR_LIT>' ] ) <EOL> min_timestamp = parse ( json . load ( argparse . FileType ( '<STR_LIT:r>' ) ( params [ '<STR_LIT>' ] ) ) ) - min_offset if '<STR_LIT>' in params else None <EOL> records = [ ] <EOL> for rec in self . range ( ) : <EOL> timestamp = self . get_record_timestamp ( rec ) <EOL> if timestamp is not None and min_timestamp is None : <EOL> min_timestamp = timestamp - min_offset <EOL> if timestamp is None or timestamp >= min_timestamp : <EOL> records . append ( rec ) <EOL> else : <EOL> break <EOL> return records <EOL> @ use ( ) <EOL> class iter_pump_hours ( iter_glucose_hours ) : <EOL> """<STR_LIT>""" <EOL> def range ( self ) : <EOL> return self . pump . model . iter_history_pages ( ) <EOL> def get_record_timestamp ( self , record ) : <EOL> return parse ( record [ '<STR_LIT>' ] ) if '<STR_LIT>' in record else None <EOL> def set_config ( args , device ) : <EOL> device . add_option ( '<STR_LIT>' , args . serial ) <EOL> def display_device ( device ) : <EOL> return '<STR_LIT>' <EOL> known_uses = [ <EOL> Session , <EOL> ] <EOL> def get_uses ( device , config ) : <EOL> all_uses = known_uses [ : ] + use . get_uses ( device , config ) <EOL> all_uses . sort ( key = lambda usage : getattr ( usage , '<STR_LIT>' , usage . __name__ ) ) <EOL> return all_uses </s>
