agregator.py
def __check_correction(self):
        """
        If the historic is big enough, we can try to find the correction value if a adjacency room is set by the user
        We use genetic algorithms to find two settings: thermal conductivity and the temperature of the adjacency room
        :return: A tuple. First element is the correction. Second element is a boolean which precise if a better
        correction as the current one has been found.
        """

        debug("| | + Check correction {}".format(len(self._historic_for_correction)))
        if len(self._historic_for_correction) == 0:
            return self._current_correction, False

        #
        # Check if the error is due to difference between real and forecasted temperature
        #
        if self.__is_delta_temperature_error():
            debug("| | | + Temperature error")
            return self._current_correction, False

        ###############################################################################################################
        #
        # Load data & prepare genetic components
        #

        subject_efficiency = getattr(self.subject, "_efficiency_factor")
        subject_thermal_capacity = units.value(self.thermal_process._thermal_capacity)
        subject_thermal_outside_coupling = units.value(self._outside_coupling.thermal_conductivity)

        # Create individual with his attributes
        creator.create("FitnessMinError", base.Fitness, weights=(-1.0,))
        creator.create("Individual", list, fitness=creator.FitnessMinError)

        toolbox = base.Toolbox()

        # Attribute generator
        toolbox.register("thermal_conductivity", random.randint, -1000., 1000.)
        toolbox.register("temperature", random.randint, -50, 50)

        # Structure initializer
        toolbox.register("individual", tools.initCycle, creator.Individual,
                         (toolbox.thermal_conductivity, toolbox.temperature), n=1)
        toolbox.register("population", tools.initRepeat, list, toolbox.individual)

        #
        # Load data & prepare genetic components
        #
        ################################################################################################################

        ################################################################################################################
        #
        # Fitness evaluation & mutation operator
        #
        def evaluate(individual):

            weight = 0.0

            individual_thermal_conductivity = individual[0]
            individual_constant_temperature = individual[1]

            eval_temperature = [self._historic_for_correction[0][1]]

            for eval_time, \
                eval_init_time,\
                eval_opt_temp,\
                eval_found_temp,\
                eval_temperature_ext_forecast,\
                evaL_temperature_ext,\
                eval_power \
                    in self._historic_for_correction:

                eval_last_temperature = eval_temperature[-1]

                eval_outside_leverage = self.__external_leverage(
                    thermal_capacity=subject_thermal_capacity,
                    thermal_conductivity=subject_thermal_outside_coupling,
                    external_temperature=evaL_temperature_ext + self._current_temperature_correction,
                    internal_temperature=eval_last_temperature)

                eval_inner_production = self.__internal_production(
                    internal_temperature=eval_init_time,
                    power_on=eval_power,
                    subject_efficiency=subject_efficiency,
                    subject_energy=units.value(self.subject.power),
                    thermal_capacity=subject_thermal_capacity)

                if self._current_correction is None:
                    correction_leverage = self.__external_leverage(
                        thermal_capacity=subject_thermal_capacity,
                        thermal_conductivity=individual_thermal_conductivity,
                        external_temperature=individual_constant_temperature,
                        internal_temperature=eval_last_temperature)
                else:
                    correction_leverage = self.__external_leverage(
                        thermal_capacity=subject_thermal_capacity,
                        thermal_conductivity=individual_thermal_conductivity
                        + self._current_correction.thermal_conductivity,
                        external_temperature=individual_constant_temperature+self._current_correction.temperature,
                        internal_temperature=eval_last_temperature)

                eval_final_temperature = eval_outside_leverage + eval_inner_production + correction_leverage

                eval_temperature.append(eval_final_temperature)

                weight += abs(eval_found_temp - eval_final_temperature) * 10000

            return weight,

        def mut_set(individual):

            individual[0] += random.uniform(-10, 10)  # thermal_coupling
            individual[1] += random.uniform(-5, 5)  # temperature

            return individual,

        #
        # Fitness evaluation & mutation operator
        #
        ################################################################################################################

        ################################################################################################################
        #
        # Initialize & compute
        #

        toolbox.register("evaluate", evaluate)
        # toolbox.register("mate", cxSet)
        toolbox.register("mate", tools.cxUniform, indpb=0.5)
        # toolbox.register("mate", tools.cxOnePoint)
        toolbox.register("mutate", mut_set)
        toolbox.register("select", tools.selTournament, tournsize=3)
        # toolbox.register("select", tools.selNSGA2)

        nb_generation = 60  # 60 or 200
        nb_population = 300  # 300 or 500
        crossing_rate = 0.3  # 0.5
        mutation_rate = 0.4  # 0.6

        pop = toolbox.population(n=nb_population)
        hof = tools.ParetoFront()
        stats = tools.Statistics(lambda ind: ind.fitness.values)
        stats.register("avg", numpy.mean, axis=0)
        stats.register("std", numpy.std, axis=0)
        stats.register("min", numpy.min, axis=0)
        stats.register("max", numpy.max, axis=0)

        pop, _ = algorithms.eaSimple(
            pop,
            toolbox,
            stats=stats,
            cxpb=crossing_rate,
            mutpb=mutation_rate,
            ngen=nb_generation,
            halloffame=hof,
            verbose=False
        )
        best = tools.selBest(pop, 1)[0]
        score = best.fitness.values[0]

        #
        # Initialize & compute
        #
        ################################################################################################################

        ################################################################################################################
        #
        # Compute the actual score and compare it with the new
        # If the new is better, we swap the current correction
        #
        temperature = [self._historic_for_correction[0][1]]
        old_score = 0.
        score_basic = 0.
        for time, init_temp,\
            opt_temp,\
            found_temp, \
            temperature_ext_forecast,\
            temperature_ext,\
            power \
                in self._historic_for_correction:

            old_score += abs(opt_temp - found_temp) * 10000

            last_temperature = temperature[-1]
            outside_leverage = self.__external_leverage(
                thermal_capacity=subject_thermal_capacity,
                thermal_conductivity=subject_thermal_outside_coupling,
                external_temperature=temperature_ext,
                internal_temperature=last_temperature)

            inner_production = self.__internal_production(
                internal_temperature=init_temp,
                power_on=power,
                subject_efficiency=subject_efficiency,
                subject_energy=units.value(self.subject.power),
                thermal_capacity=subject_thermal_capacity)

            final_temperature = outside_leverage + inner_production
            temperature.append(final_temperature)
            score_basic += abs(found_temp - final_temperature) * 10000

        if score_basic < old_score and score_basic < score:
            debug("| | | + reinit")
            return None, True

        if score > old_score or score == float('nan'):
            debug("| | | + keep current correction")
            return self._current_correction, False

        if self._current_correction is None:
            c = self.Correction(best[0], best[1])
            debug("| | | + swap with correction: {} and {} fitness value [{} {}]".format(best,
                                                                                         best.fitness.values[0],
                                                                                         c.thermal_conductivity,
                                                                                         c.temperature))
            return c, True

        c = self.Correction(
            best[0] + self._current_correction.thermal_conductivity,
            best[1] + self._current_correction.temperature
        )

        return c, True

        #
        # Compute the actual score and compare it with the new
        #
        ################################################################################################################

        If the historic is big enough, we can try to find the correction value if a adjacency room is set by the user
        We use genetic algorithms to find two settings: thermal conductivity and the temperature of the adjacency room
        :return: A tuple. First element is the correction. Second element is a boolean which precise if a better
        correction as the current one has been found.
        
If the historic is big enough, we can try to find the correction value if a adjacency room is set by the user
We use genetic algorithms to find two settings: thermal conductivity and the temperature of the adjacency room
If the historic is big enough we can try to find the correction value if a adjacency room is set by the user We use genetic algorithms to find two settings : thermal conductivity and the temperature of the adjacency room
def __check_correction(self):
        debug("| | + Check correction {}".format(len(self._historic_for_correction)))
        if len(self._historic_for_correction) == 0:
            return self._current_correction, False
         Check if the error is due to difference between real and forecasted temperature
        if self.__is_delta_temperature_error():
            debug("| | | + Temperature error")
            return self._current_correction, False
         Load data & prepare genetic components
        subject_efficiency = getattr(self.subject, "_efficiency_factor")
        subject_thermal_capacity = units.value(self.thermal_process._thermal_capacity)
        subject_thermal_outside_coupling = units.value(self._outside_coupling.thermal_conductivity)
         Create individual with his attributes
        creator.create("FitnessMinError", base.Fitness, weights=(-1.0,))
        creator.create("Individual", list, fitness=creator.FitnessMinError)
        toolbox = base.Toolbox()
         Attribute generator
        toolbox.register("thermal_conductivity", random.randint, -1000., 1000.)
        toolbox.register("temperature", random.randint, -50, 50)
         Structure initializer
        toolbox.register("individual", tools.initCycle, creator.Individual,
                         (toolbox.thermal_conductivity, toolbox.temperature), n=1)
        toolbox.register("population", tools.initRepeat, list, toolbox.individual)
         Load data & prepare genetic components
         Fitness evaluation & mutation operator
        def evaluate(individual):
            weight = 0.0
            individual_thermal_conductivity = individual[0]
            individual_constant_temperature = individual[1]
            eval_temperature = [self._historic_for_correction[0][1]]
            for eval_time, \
                eval_init_time,\
                eval_opt_temp,\
                eval_found_temp,\
                eval_temperature_ext_forecast,\
                evaL_temperature_ext,\
                eval_power \
                    in self._historic_for_correction:
                eval_last_temperature = eval_temperature[-1]
                eval_outside_leverage = self.__external_leverage(
                    thermal_capacity=subject_thermal_capacity,
                    thermal_conductivity=subject_thermal_outside_coupling,
                    external_temperature=evaL_temperature_ext + self._current_temperature_correction,
                    internal_temperature=eval_last_temperature)
                eval_inner_production = self.__internal_production(
                    internal_temperature=eval_init_time,
                    power_on=eval_power,
                    subject_efficiency=subject_efficiency,
                    subject_energy=units.value(self.subject.power),
                    thermal_capacity=subject_thermal_capacity)
                if self._current_correction is None:
                    correction_leverage = self.__external_leverage(
                        thermal_capacity=subject_thermal_capacity,
                        thermal_conductivity=individual_thermal_conductivity,
                        external_temperature=individual_constant_temperature,
                        internal_temperature=eval_last_temperature)
                else:
                    correction_leverage = self.__external_leverage(
                        thermal_capacity=subject_thermal_capacity,
                        thermal_conductivity=individual_thermal_conductivity
                        + self._current_correction.thermal_conductivity,
                        external_temperature=individual_constant_temperature+self._current_correction.temperature,
                        internal_temperature=eval_last_temperature)
                eval_final_temperature = eval_outside_leverage + eval_inner_production + correction_leverage
                eval_temperature.append(eval_final_temperature)
                weight += abs(eval_found_temp - eval_final_temperature) * 10000
            return weight,
        def mut_set(individual):
            individual[0] += random.uniform(-10, 10)   thermal_coupling
            individual[1] += random.uniform(-5, 5)   temperature
            return individual,
         Fitness evaluation & mutation operator
         Initialize & compute
        toolbox.register("evaluate", evaluate)
         toolbox.register("mate", cxSet)
        toolbox.register("mate", tools.cxUniform, indpb=0.5)
         toolbox.register("mate", tools.cxOnePoint)
        toolbox.register("mutate", mut_set)
        toolbox.register("select", tools.selTournament, tournsize=3)
         toolbox.register("select", tools.selNSGA2)
        nb_generation = 60   60 or 200
        nb_population = 300   300 or 500
        crossing_rate = 0.3   0.5
        mutation_rate = 0.4   0.6
        pop = toolbox.population(n=nb_population)
        hof = tools.ParetoFront()
        stats = tools.Statistics(lambda ind: ind.fitness.values)
        stats.register("avg", numpy.mean, axis=0)
        stats.register("std", numpy.std, axis=0)
        stats.register("min", numpy.min, axis=0)
        stats.register("max", numpy.max, axis=0)
        pop, _ = algorithms.eaSimple(
            pop,
            toolbox,
            stats=stats,
            cxpb=crossing_rate,
            mutpb=mutation_rate,
            ngen=nb_generation,
            halloffame=hof,
            verbose=False
        )
        best = tools.selBest(pop, 1)[0]
        score = best.fitness.values[0]
         Initialize & compute
         Compute the actual score and compare it with the new
         If the new is better, we swap the current correction
        temperature = [self._historic_for_correction[0][1]]
        old_score = 0.
        score_basic = 0.
        for time, init_temp,\
            opt_temp,\
            found_temp, \
            temperature_ext_forecast,\
            temperature_ext,\
            power \
                in self._historic_for_correction:
            old_score += abs(opt_temp - found_temp) * 10000
            last_temperature = temperature[-1]
            outside_leverage = self.__external_leverage(
                thermal_capacity=subject_thermal_capacity,
                thermal_conductivity=subject_thermal_outside_coupling,
                external_temperature=temperature_ext,
                internal_temperature=last_temperature)
            inner_production = self.__internal_production(
                internal_temperature=init_temp,
                power_on=power,
                subject_efficiency=subject_efficiency,
                subject_energy=units.value(self.subject.power),
                thermal_capacity=subject_thermal_capacity)
            final_temperature = outside_leverage + inner_production
            temperature.append(final_temperature)
            score_basic += abs(found_temp - final_temperature) * 10000
        if score_basic < old_score and score_basic < score:
            debug("| | | + reinit")
            return None, True
        if score > old_score or score == float('nan'):
            debug("| | | + keep current correction")
            return self._current_correction, False
        if self._current_correction is None:
            c = self.Correction(best[0], best[1])
            debug("| | | + swap with correction: {} and {} fitness value [{} {}]".format(best,
                                                                                         best.fitness.values[0],
                                                                                         c.thermal_conductivity,
                                                                                         c.temperature))
            return c, True
        c = self.Correction(
            best[0] + self._current_correction.thermal_conductivity,
            best[1] + self._current_correction.temperature
        )
        return c, True
         Compute the actual score and compare it with the new
def __check_correction ( self ) : debug ( "| | + Check correction {}" . format ( len ( self . _historic_for_correction ) ) ) if len ( self . _historic_for_correction ) == 0 : return self . _current_correction , False if self . __is_delta_temperature_error ( ) : debug ( "| | | + Temperature error" ) return self . _current_correction , False subject_efficiency = getattr ( self . subject , "_efficiency_factor" ) subject_thermal_capacity = units . value ( self . thermal_process . _thermal_capacity ) subject_thermal_outside_coupling = units . value ( self . _outside_coupling . thermal_conductivity ) creator . create ( "FitnessMinError" , base . Fitness , weights = ( - 1.0 , ) ) creator . create ( "Individual" , list , fitness = creator . FitnessMinError ) toolbox = base . Toolbox ( ) toolbox . register ( "thermal_conductivity" , random . randint , - 1000. , 1000. ) toolbox . register ( "temperature" , random . randint , - 50 , 50 ) toolbox . register ( "individual" , tools . initCycle , creator . Individual , ( toolbox . thermal_conductivity , toolbox . temperature ) , n = 1 ) toolbox . register ( "population" , tools . initRepeat , list , toolbox . individual ) def evaluate ( individual ) : weight = 0.0 individual_thermal_conductivity = individual [ 0 ] individual_constant_temperature = individual [ 1 ] eval_temperature = [ self . _historic_for_correction [ 0 ] [ 1 ] ] for eval_time , eval_init_time , eval_opt_temp , eval_found_temp , eval_temperature_ext_forecast , evaL_temperature_ext , eval_power in self . _historic_for_correction : eval_last_temperature = eval_temperature [ - 1 ] eval_outside_leverage = self . __external_leverage ( thermal_capacity = subject_thermal_capacity , thermal_conductivity = subject_thermal_outside_coupling , external_temperature = evaL_temperature_ext + self . _current_temperature_correction , internal_temperature = eval_last_temperature ) eval_inner_production = self . __internal_production ( internal_temperature = eval_init_time , power_on = eval_power , subject_efficiency = subject_efficiency , subject_energy = units . value ( self . subject . power ) , thermal_capacity = subject_thermal_capacity ) if self . _current_correction is None : correction_leverage = self . __external_leverage ( thermal_capacity = subject_thermal_capacity , thermal_conductivity = individual_thermal_conductivity , external_temperature = individual_constant_temperature , internal_temperature = eval_last_temperature ) else : correction_leverage = self . __external_leverage ( thermal_capacity = subject_thermal_capacity , thermal_conductivity = individual_thermal_conductivity + self . _current_correction . thermal_conductivity , external_temperature = individual_constant_temperature + self . _current_correction . temperature , internal_temperature = eval_last_temperature ) eval_final_temperature = eval_outside_leverage + eval_inner_production + correction_leverage eval_temperature . append ( eval_final_temperature ) weight += abs ( eval_found_temp - eval_final_temperature ) * 10000 return weight , def mut_set ( individual ) : individual [ 0 ] += random . uniform ( - 10 , 10 ) individual [ 1 ] += random . uniform ( - 5 , 5 ) return individual , toolbox . register ( "evaluate" , evaluate ) toolbox . register ( "mate" , tools . cxUniform , indpb = 0.5 ) toolbox . register ( "mutate" , mut_set ) toolbox . register ( "select" , tools . selTournament , tournsize = 3 ) nb_generation = 60 nb_population = 300 crossing_rate = 0.3 mutation_rate = 0.4 pop = toolbox . population ( n = nb_population ) hof = tools . ParetoFront ( ) stats = tools . Statistics ( lambda ind : ind . fitness . values ) stats . register ( "avg" , numpy . mean , axis = 0 ) stats . register ( "std" , numpy . std , axis = 0 ) stats . register ( "min" , numpy . min , axis = 0 ) stats . register ( "max" , numpy . max , axis = 0 ) pop , _ = algorithms . eaSimple ( pop , toolbox , stats = stats , cxpb = crossing_rate , mutpb = mutation_rate , ngen = nb_generation , halloffame = hof , verbose = False ) best = tools . selBest ( pop , 1 ) [ 0 ] score = best . fitness . values [ 0 ] temperature = [ self . _historic_for_correction [ 0 ] [ 1 ] ] old_score = 0. score_basic = 0. for time , init_temp , opt_temp , found_temp , temperature_ext_forecast , temperature_ext , power in self . _historic_for_correction : old_score += abs ( opt_temp - found_temp ) * 10000 last_temperature = temperature [ - 1 ] outside_leverage = self . __external_leverage ( thermal_capacity = subject_thermal_capacity , thermal_conductivity = subject_thermal_outside_coupling , external_temperature = temperature_ext , internal_temperature = last_temperature ) inner_production = self . __internal_production ( internal_temperature = init_temp , power_on = power , subject_efficiency = subject_efficiency , subject_energy = units . value ( self . subject . power ) , thermal_capacity = subject_thermal_capacity ) final_temperature = outside_leverage + inner_production temperature . append ( final_temperature ) score_basic += abs ( found_temp - final_temperature ) * 10000 if score_basic < old_score and score_basic < score : debug ( "| | | + reinit" ) return None , True if score > old_score or score == float ( 'nan' ) : debug ( "| | | + keep current correction" ) return self . _current_correction , False if self . _current_correction is None : c = self . Correction ( best [ 0 ] , best [ 1 ] ) debug ( "| | | + swap with correction: {} and {} fitness value [{} {}]" . format ( best , best . fitness . values [ 0 ] , c . thermal_conductivity , c . temperature ) ) return c , True c = self . Correction ( best [ 0 ] + self . _current_correction . thermal_conductivity , best [ 1 ] + self . _current_correction . temperature ) return c , True
If the historic is big enough, we can try to find the correction value if a adjacency room is set by the user
We use genetic algorithms to find two settings: thermal conductivity and the temperature of the adjacency room
If the historic is big enough we can try to find the correction value if a adjacency room is set by the user We use genetic algorithms to find two settings : thermal conductivity and the temperature of the adjacency room
"""
        If the historic is big enough, we can try to find the correction value if a adjacency room is set by the user
        We use genetic algorithms to find two settings: thermal conductivity and the temperature of the adjacency room
        :return: A tuple. First element is the correction. Second element is a boolean which precise if a better
        correction as the current one has been found.
        """ # # Check if the error is due to difference between real and forecasted temperature # ############################################################################################################### # # Load data & prepare genetic components # # Create individual with his attributes # Attribute generator # Structure initializer # # Load data & prepare genetic components # ################################################################################################################ ################################################################################################################ # # Fitness evaluation & mutation operator # # thermal_coupling # temperature # # Fitness evaluation & mutation operator # ################################################################################################################ ################################################################################################################ # # Initialize & compute # # toolbox.register("mate", cxSet) # toolbox.register("mate", tools.cxOnePoint) # toolbox.register("select", tools.selNSGA2) # 60 or 200 # 300 or 500 # 0.5 # 0.6 # # Initialize & compute # ################################################################################################################ ################################################################################################################ # # Compute the actual score and compare it with the new # If the new is better, we swap the current correction # # # Compute the actual score and compare it with the new # ################################################################################################################
A tuple. First element is the correction. Second element is a boolean which precise if a better
correction as the current one has been found.
A tuple . First element is the correction . Second element is a boolean which precise if a better correction as the current one has been found .
self