<s> import rospkg <EOL> import rospy <EOL> import sys <EOL> import xml . etree . ElementTree as ET <EOL> import calendar <EOL> import time <EOL> import random <EOL> from datetime import datetime <EOL> from os . path import expanduser <EOL> from collections import OrderedDict <EOL> from random import randint <EOL> from codecs import open <EOL> from rapp_platform_ros_communications . srv import ( <EOL> ontologySubSuperClassesOfSrv , <EOL> ontologySubSuperClassesOfSrvRequest , <EOL> ontologySubSuperClassesOfSrvResponse , <EOL> testSelectorSrv , <EOL> testSelectorSrvResponse , <EOL> createOntologyAliasSrv , <EOL> createOntologyAliasSrvRequest , <EOL> createOntologyAliasSrvResponse , <EOL> userPerformanceCognitveTestsSrv , <EOL> userPerformanceCognitveTestsSrvRequest , <EOL> userPerformanceCognitveTestsSrvResponse , <EOL> cognitiveTestsOfTypeSrv , <EOL> cognitiveTestsOfTypeSrvRequest , <EOL> cognitiveTestsOfTypeSrvResponse , <EOL> fetchDataSrv , <EOL> fetchDataSrvRequest , <EOL> fetchDataSrvResponse <EOL> ) <EOL> from rapp_platform_ros_communications . msg import ( <EOL> StringArrayMsg <EOL> ) <EOL> class TestSelector : <EOL> lookBackTimeStamp = <NUM_LIT> <EOL> def chooserFunction ( self , req ) : <EOL> try : <EOL> res = testSelectorSrvResponse ( ) <EOL> currentTimestamp = int ( time . time ( ) ) <EOL> returnWithError , modifier1 , modifier2 = self . loadParamDifficultyModifiers ( res ) <EOL> if ( returnWithError ) : <EOL> return res <EOL> returnWithError , userOntologyAlias = self . getUserOntologyAlias ( req . username , res ) <EOL> if ( returnWithError ) : <EOL> return res <EOL> returnWithError , userLanguage = self . getUserLanguage ( req . username , res ) <EOL> if ( returnWithError ) : <EOL> return res <EOL> returnWithError , testTypesList = self . getTestTypesFromOntology ( res ) <EOL> if ( returnWithError ) : <EOL> return res <EOL> if ( req . testType == "<STR_LIT>" ) : <EOL> testType = self . determineTestTypeIfNotProvided ( res , testTypesList , userOntologyAlias ) <EOL> else : <EOL> if ( req . testType not in testTypesList ) : <EOL> res . trace . append ( "<STR_LIT>" ) <EOL> res . error = "<STR_LIT>" <EOL> res . success = False <EOL> return res <EOL> else : <EOL> testType = req . testType <EOL> chosenDif , noUserPerformanceRecordsExist , userPerfOrganizedByTimestamp = self . organizeUserPerformanceByTimestampAndDetermineTestDifficulty ( testType , userOntologyAlias , currentTimestamp , self . lookBackTimeStamp , res , modifier1 , modifier2 ) <EOL> returnWithError , testsOfTypeOrdered = self . getCognitiveTestsOfType ( testType , userLanguage , chosenDif , res ) <EOL> if ( returnWithError ) : <EOL> return res <EOL> finalTestFilePath , finalTestname = self . getLRUtestOfTypeAndXmlPath ( testsOfTypeOrdered , noUserPerformanceRecordsExist , userPerfOrganizedByTimestamp ) <EOL> tmpList = finalTestname . split ( '<STR_LIT:#>' ) <EOL> res . test = tmpList [ <NUM_LIT:1> ] <EOL> rospack = rospkg . RosPack ( ) <EOL> localPackagePath = rospack . get_path ( '<STR_LIT>' ) <EOL> finalTestFilePath = localPackagePath + finalTestFilePath <EOL> res . trace . append ( finalTestFilePath ) <EOL> self . retrieveDataFromTestXml ( finalTestFilePath , res , userLanguage ) <EOL> res . language = userLanguage <EOL> res . success = True <EOL> except IndexError : <EOL> res . trace . append ( "<STR_LIT>" ) <EOL> res . error = "<STR_LIT>" <EOL> res . success = False <EOL> except IOError : <EOL> res . success = False <EOL> res . trace . append ( "<STR_LIT>" ) <EOL> res . error = "<STR_LIT>" <EOL> return res <EOL> def getUserLanguage ( self , username , res ) : <EOL> serv_topic = rospy . get_param ( '<STR_LIT>' ) <EOL> knowrob_service = rospy . ServiceProxy ( serv_topic , fetchDataSrv ) <EOL> fetchDataSrvReq = fetchDataSrvRequest ( ) <EOL> fetchDataSrvReq . req_cols = [ "<STR_LIT>" ] <EOL> fetchDataSrvReq . where_data = [ StringArrayMsg ( s = [ "<STR_LIT:username>" , username ] ) ] <EOL> fetchDataSrvResponse = knowrob_service ( fetchDataSrvReq ) <EOL> if ( fetchDataSrvResponse . success . data != True ) : <EOL> res . trace . extend ( fetchDataSrvResponse . trace ) <EOL> res . error = fetchDataSrvResponse . trace [ <NUM_LIT:0> ] <EOL> res . success = False <EOL> return True , "<STR_LIT>" <EOL> return False , fetchDataSrvResponse . res_data [ <NUM_LIT:0> ] . s [ <NUM_LIT:0> ] <EOL> def getTestTypesFromOntology ( self , res ) : <EOL> serv_topic = rospy . get_param ( '<STR_LIT>' ) <EOL> knowrob_service = rospy . ServiceProxy ( serv_topic , ontologySubSuperClassesOfSrv ) <EOL> testTypesReq = ontologySubSuperClassesOfSrvRequest ( ) <EOL> testTypesReq . ontology_class = "<STR_LIT>" <EOL> testTypesResponse = knowrob_service ( testTypesReq ) <EOL> if ( testTypesResponse . success != True ) : <EOL> res . trace . extend ( testTypesResponse . trace ) <EOL> res . trace . append ( "<STR_LIT>" ) <EOL> res . error = testTypesResponse . error + "<STR_LIT>" <EOL> res . success = False <EOL> return True , "<STR_LIT>" <EOL> testTypesList = [ ] <EOL> for s in testTypesResponse . results : <EOL> tmpList = s . split ( '<STR_LIT:#>' ) <EOL> testTypesList . append ( tmpList [ <NUM_LIT:1> ] ) <EOL> return False , testTypesList <EOL> def determineTestTypeIfNotProvided ( self , res , testTypesList , userOntologyAlias ) : <EOL> serv_topic = rospy . get_param ( '<STR_LIT>' ) <EOL> res . trace . append ( "<STR_LIT>" ) <EOL> noRecordsList = [ ] <EOL> d1 = OrderedDict ( ) <EOL> for s in testTypesList : <EOL> userPerformanceReq = userPerformanceCognitveTestsSrvRequest ( ) <EOL> userPerformanceReq . test_type = s <EOL> userPerformanceReq . ontology_alias = userOntologyAlias <EOL> knowrob_service = rospy . ServiceProxy ( serv_topic , userPerformanceCognitveTestsSrv ) <EOL> userPerformanceResponse = knowrob_service ( userPerformanceReq ) <EOL> if ( userPerformanceResponse . success != True ) : <EOL> noRecordsList . append ( s ) <EOL> else : <EOL> tmpUserPerfOrganizedByTimestamp = self . organizeUserPerformance ( userPerformanceResponse ) <EOL> t1 = tmpUserPerfOrganizedByTimestamp . items ( ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> d1 [ t1 ] = [ s ] <EOL> if ( len ( noRecordsList ) > <NUM_LIT:0> ) : <EOL> testType = random . choice ( noRecordsList ) <EOL> res . trace . append ( "<STR_LIT>" + testType ) <EOL> else : <EOL> d1 = OrderedDict ( sorted ( d1 . items ( ) , key = lambda t : t [ <NUM_LIT:0> ] ) ) <EOL> testType = d1 . values ( ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> res . trace . append ( "<STR_LIT>" + testType ) <EOL> return testType <EOL> def organizeUserPerformanceByTimestampAndDetermineTestDifficulty ( self , testType , userOntologyAlias , currentTimestamp , lookBackTimeStamp , res , modifier1 , modifier2 ) : <EOL> noUserPerformanceRecordsExist = False ; <EOL> chosenDif = "<STR_LIT:1>" <EOL> userPerformanceReq = userPerformanceCognitveTestsSrvRequest ( ) <EOL> userPerformanceReq . test_type = testType <EOL> userPerformanceReq . ontology_alias = userOntologyAlias <EOL> serv_topic = rospy . get_param ( '<STR_LIT>' ) <EOL> knowrob_service = rospy . ServiceProxy ( serv_topic , userPerformanceCognitveTestsSrv ) <EOL> userPerformanceResponse = knowrob_service ( userPerformanceReq ) <EOL> userPerfOrganizedByTimestamp = [ ] <EOL> if ( userPerformanceResponse . success != True ) : <EOL> res . trace . extend ( userPerformanceResponse . trace ) <EOL> res . trace . append ( "<STR_LIT>" ) <EOL> noUserPerformanceRecordsExist = True <EOL> else : <EOL> userPerfOrganizedByTimestamp = self . organizeUserPerformance ( userPerformanceResponse ) <EOL> for k , v in userPerfOrganizedByTimestamp . items ( ) : <EOL> if ( currentTimestamp - k > lookBackTimeStamp ) : <EOL> del userPerfOrganizedByTimestamp [ k ] <EOL> else : <EOL> break <EOL> userScore = self . calculateUserScore ( userPerfOrganizedByTimestamp ) <EOL> res . trace . append ( "<STR_LIT>" + str ( userScore ) ) <EOL> if ( userScore == <NUM_LIT:0> ) : <EOL> chosenDif = "<STR_LIT:1>" <EOL> elif ( userScore < modifier1 ) : <EOL> chosenDif = "<STR_LIT:1>" <EOL> elif ( userScore < modifier2 ) : <EOL> chosenDif = "<STR_LIT:2>" <EOL> else : <EOL> chosenDif = "<STR_LIT:3>" <EOL> res . trace . append ( "<STR_LIT>" + chosenDif ) <EOL> return chosenDif , noUserPerformanceRecordsExist , userPerfOrganizedByTimestamp <EOL> def getCognitiveTestsOfType ( self , testType , userLanguage , chosenDif , res ) : <EOL> serv_topic = rospy . get_param ( '<STR_LIT>' ) <EOL> cognitiveTestsOfTypeSrvReq = cognitiveTestsOfTypeSrvRequest ( ) <EOL> cognitiveTestsOfTypeSrvReq . test_type = testType <EOL> cognitiveTestsOfTypeSrvReq . test_language = userLanguage <EOL> knowrob_service = rospy . ServiceProxy ( serv_topic , cognitiveTestsOfTypeSrv ) <EOL> cognitiveTestsOfTypeResponse = knowrob_service ( cognitiveTestsOfTypeSrvReq ) <EOL> if ( cognitiveTestsOfTypeResponse . success != True ) : <EOL> res . trace . extend ( cognitiveTestsOfTypeResponse . trace ) <EOL> res . error = cognitiveTestsOfTypeResponse . error <EOL> res . success = False <EOL> return True , "<STR_LIT>" <EOL> success , testsOfTypeOrdered = self . filterTestsbyDifficulty ( cognitiveTestsOfTypeResponse , chosenDif , res ) <EOL> if ( not success ) : <EOL> res . trace . append ( "<STR_LIT>" ) <EOL> res . error = "<STR_LIT>" <EOL> res . success = False <EOL> return True , "<STR_LIT>" <EOL> return False , testsOfTypeOrdered <EOL> def getLRUtestOfTypeAndXmlPath ( self , testsOfTypeOrdered , noUserPerformanceRecordsExist , userPerfOrganizedByTimestamp ) : <EOL> finalTestname = "<STR_LIT>" <EOL> finalTestFilePath = "<STR_LIT>" <EOL> if ( noUserPerformanceRecordsExist ) : <EOL> finalTestname = random . choice ( testsOfTypeOrdered . keys ( ) ) <EOL> finalTest = testsOfTypeOrdered [ finalTestname ] <EOL> finalTestFilePath = finalTest [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> else : <EOL> testsOfTypeOrderedCopy = testsOfTypeOrdered . copy ( ) <EOL> for k , v in userPerfOrganizedByTimestamp . items ( ) : <EOL> if ( v [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] in testsOfTypeOrderedCopy ) : <EOL> del testsOfTypeOrderedCopy [ v [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ] <EOL> if ( len ( testsOfTypeOrderedCopy ) > <NUM_LIT:0> ) : <EOL> finalTestname = random . choice ( testsOfTypeOrderedCopy . keys ( ) ) <EOL> finalTest = testsOfTypeOrderedCopy [ finalTestname ] <EOL> finalTestFilePath = finalTest [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> else : <EOL> finalTestname = userPerfOrganizedByTimestamp . values ( ) [ len ( userPerfOrganizedByTimestamp ) - <NUM_LIT:1> ] <EOL> finalTestname = finalTestname [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> finalTest = testsOfTypeOrdered [ finalTestname ] <EOL> finalTestFilePath = finalTest [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> return finalTestFilePath , finalTestname <EOL> def retrieveDataFromTestXml ( self , finalTestFilePath , res , userLanguage ) : <EOL> tree = ET . parse ( finalTestFilePath ) <EOL> root = tree . getroot ( ) <EOL> res . testType = root . find ( "<STR_LIT>" ) . text . encode ( '<STR_LIT>' ) <EOL> res . testSubType = root . find ( "<STR_LIT>" ) . text . encode ( '<STR_LIT>' ) <EOL> language = root . find ( '<STR_LIT>' ) <EOL> for question in language . find ( userLanguage ) : <EOL> res . questions . append ( question . find ( "<STR_LIT:body>" ) . text . encode ( '<STR_LIT>' ) ) <EOL> res . correctAnswers . append ( question . find ( "<STR_LIT>" ) . text . encode ( '<STR_LIT>' ) ) <EOL> line = StringArrayMsg ( ) <EOL> for answers in question . findall ( '<STR_LIT>' ) : <EOL> line . s . append ( answers . find ( "<STR_LIT:body>" ) . text . encode ( '<STR_LIT>' ) ) <EOL> res . answers . append ( line ) <EOL> def organizeUserPerformance ( self , userPerf ) : <EOL> d = OrderedDict ( ) <EOL> for i in range ( len ( userPerf . tests ) ) : <EOL> tlist = [ userPerf . tests [ i ] , userPerf . scores [ i ] , userPerf . difficulty [ i ] ] <EOL> d [ int ( userPerf . timestamps [ i ] ) ] = [ tlist ] <EOL> d = OrderedDict ( sorted ( d . items ( ) , key = lambda t : t [ <NUM_LIT:0> ] , reverse = True ) ) <EOL> return d <EOL> def calculateUserScore ( self , d ) : <EOL> score = <NUM_LIT:0> <EOL> if ( len ( d ) > <NUM_LIT:0> ) : <EOL> for k , v in d . items ( ) : <EOL> score = score + int ( v [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) * int ( v [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> score = score / len ( d ) <EOL> return score <EOL> else : <EOL> return <NUM_LIT:0> <EOL> def filterTestsbyDifficulty ( self , testsOfType , chosenDif , res ) : <EOL> success = False <EOL> d = dict ( ) <EOL> intDif = int ( chosenDif ) <EOL> if ( intDif == <NUM_LIT:0> ) : <EOL> return success , d <EOL> else : <EOL> for i in range ( len ( testsOfType . tests ) ) : <EOL> if ( testsOfType . difficulty [ i ] == chosenDif ) : <EOL> tlist = [ testsOfType . file_paths [ i ] , testsOfType . difficulty [ i ] , testsOfType . subtype [ i ] ] <EOL> d [ testsOfType . tests [ i ] ] = [ tlist ] <EOL> if ( not len ( d ) > <NUM_LIT:0> ) : <EOL> res . trace . append ( "<STR_LIT>" + chosenDif ) ; <EOL> chosenDif = str ( int ( chosenDif ) - <NUM_LIT:1> ) <EOL> success , d = self . filterTestsbyDifficulty ( testsOfType , chosenDif , res ) <EOL> else : <EOL> success = True <EOL> return success , d <EOL> def getUserOntologyAlias ( self , username , res ) : <EOL> serv_topic = rospy . get_param ( '<STR_LIT>' ) <EOL> knowrob_service = rospy . ServiceProxy ( serv_topic , createOntologyAliasSrv ) <EOL> createOntologyAliasReq = createOntologyAliasSrvRequest ( ) <EOL> createOntologyAliasReq . username = username <EOL> createOntologyAliasResponse = knowrob_service ( createOntologyAliasReq ) <EOL> if ( createOntologyAliasResponse . success != True ) : <EOL> res . trace . extend ( createOntologyAliasResponse . trace ) <EOL> res . error = createOntologyAliasResponse . error <EOL> res . success = False <EOL> returnWithError = True ; <EOL> return True , "<STR_LIT>" <EOL> return False , createOntologyAliasResponse . ontology_alias <EOL> def loadParamDifficultyModifiers ( self , res ) : <EOL> modifier1 = rospy . get_param ( '<STR_LIT>' ) <EOL> if ( not modifier1 ) : <EOL> rospy . logerror ( "<STR_LIT>" ) <EOL> res . trace . extend ( "<STR_LIT>" ) <EOL> res . error = "<STR_LIT>" <EOL> res . success = False <EOL> returnWithError = True ; <EOL> return True , "<STR_LIT>" , "<STR_LIT>" <EOL> modifier2 = rospy . get_param ( '<STR_LIT>' ) <EOL> if ( not modifier2 ) : <EOL> rospy . logerror ( "<STR_LIT>" ) <EOL> res . trace . extend ( "<STR_LIT>" ) <EOL> res . error = "<STR_LIT>" <EOL> res . success = False <EOL> returnWithError = True ; <EOL> return True , "<STR_LIT>" , "<STR_LIT>" <EOL> modifier1 = int ( modifier1 ) <EOL> modifier2 = int ( modifier2 ) <EOL> return False , modifier1 , modifier2 </s>
