Source code for param_study.analyze


import os
import json
import yaml

import logging

from pickle import UnpicklingError

import numpy as np

from param_study import database
from utils.helpers import Saver



[docs]class Analyzer():
    def __init__(self, parameter_study_name: str = 'default') -> None:
        self.parameter_study_name = parameter_study_name
        self.analysis = {}
        session = database.Session()
        self.test_cases = database.TestCase.find_finished_for_study(session=session, parameter_study_name=self.parameter_study_name)


[docs]    def read_infos_from_saver(self, saver: Saver):
        informations = {}
        if len(saver.t) == 0:
            return {'successful': False}
        t = np.array(saver.t) - saver.t[0]
        informations['timesteps_n'] = len(saver.t)
        informations['timesteps_taking_too_long_n'] = np.count_nonzero(np.diff(t) > 0.1)
        informations['last_time_step'] = (t[-1]).tolist()
        informations['landmarks_n'] = len(saver.landmarks[-1].tolist())
        informations['landmarks_lap1_n'] = len(saver.landmarks[np.argmax(t>19.5)].tolist())
        informations['landmarks_lap2_n'] = len(saver.landmarks[np.argmax(t>34)].tolist())
        informations['successful'] = self.check_if_test_case_successful(informations=informations)
        return informations



[docs]    def check_if_test_case_successful(self, informations: dict) -> bool:
        return informations['last_time_step'] > 38



[docs]    def analyze_test_case(self, test_case: database.TestCase):
        logging.info(f'Analyzing test_case "{test_case.index}" for parameter study "{self.parameter_study_name}"')
        try:
            saver = Saver.read_pickle_parameter_study(test_case=str(test_case.index), parameter_study_name=self.parameter_study_name)
        except EOFError:
            logging.error(f'EOFError for test case "{test_case.index}" for parameter study "{self.parameter_study_name}"')
            return
        except UnpicklingError:
            logging.error(f'UnpicklingError for test case "{test_case.index}" for parameter study "{self.parameter_study_name}"')
            return
        else:
            if saver is None:
                return
        informations = self.read_infos_from_saver(saver)
        with open(f'{os.path.abspath(os.path.dirname(__file__))}/../../param_study/{self.parameter_study_name}/configs/test_case_{test_case.index}.yaml', 'r') as stream:
            config = yaml.load(stream)
        informations['config'] = config
        self.analysis[test_case.index] = informations



[docs]    def save_analysis(self):
        logging.info('Saving analysis')
        with open(f'{os.path.abspath(os.path.dirname(__file__))}/../../param_study/{self.parameter_study_name}/analysis.json', 'w') as output:
            json.dump(self.analysis, output, indent=4, sort_keys=True)



[docs]    def run(self):
        logging.info('Running analysis')
        for test_case_i, test_case in enumerate(self.test_cases):
            self.analyze_test_case(test_case=test_case)
            # if test_case_i > 5:
            #     break
        self.save_analysis()



if __name__ == '__main__':
    analyzer = Analyzer(parameter_study_name='test')
    analyzer.run()