Source code for spinn_front_end_common.interface.config_handler

# Copyright (c) 2017 The University of Manchester
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

from configparser import NoOptionError
import logging
import os
import shutil
import traceback
from spinn_utilities.log import FormatAdapter
from spinn_utilities.config_holder import (
    config_options, load_config, get_config_bool, get_config_int,
    get_config_str, get_config_str_list, set_config)
from spinn_machine import Machine
from spinn_front_end_common.interface.provenance import LogStoreDB
from import FecDataWriter
from spinn_front_end_common.utilities.exceptions import ConfigurationException

logger = FormatAdapter(logging.getLogger(__name__))

APP_DIRNAME = 'application_generated_data_files'
WARNING_LOGS_FILENAME = "warning_logs.txt"

# options names are all lower without _ inside config
_DEBUG_ENABLE_OPTS = frozenset([
    "clear_iobuf_during_run", "extract_iobuf"])
_REPORT_DISABLE_OPTS = frozenset([
    "clear_iobuf_during_run", "extract_iobuf"])

[docs]class ConfigHandler(object): """ Superclass of AbstractSpinnakerBase that handles function only dependent of the config and the order its methods are called. """ __slots__ = [ # The writer and therefore view of the global data "_data_writer" ] def __init__(self, data_writer_cls=None): """ :param FecDataWriter data_writer: The Global data writer object """ load_config() if data_writer_cls: self._data_writer = data_writer_cls.setup() else: self._data_writer = FecDataWriter.setup() logger.set_log_store(LogStoreDB()) # set up machine targeted data self._debug_configs() self._previous_handler() # Pass max_machine_cores to Machine so if effects everything! max_machine_core = get_config_int("Machine", "max_machine_core") if max_machine_core is not None: Machine.set_max_cores_per_chip(max_machine_core) def _debug_configs(self): """ Adjust and checks config based on mode and reports_enabled. :raises ConfigurationException: """ if get_config_str("Mode", "mode") == "Debug": for option in config_options("Reports"): # options names are all lower without _ inside config if option in _DEBUG_ENABLE_OPTS or option[:5] == "write": if not get_config_bool("Reports", option): set_config("Reports", option, "True")"As mode == \"Debug\", [Reports] {} " "has been set to True", option) elif not get_config_bool("Reports", "reportsEnabled"): for option in config_options("Reports"): # options names are all lower without _ inside config if option in _REPORT_DISABLE_OPTS or option[:5] == "write": if not get_config_bool("Reports", option): set_config("Reports", option, "False") "As reportsEnabled == \"False\", [Reports] {} " "has been set to False", option) if get_config_bool("Machine", "virtual_board"): # TODO handle in the execute methods if get_config_bool("Reports", "write_energy_report"): set_config("Reports", "write_energy_report", "False")"[Reports]write_energy_report has been set to " "False as using virtual boards") def _previous_handler(self): self._error_on_previous("loading_algorithms") self._error_on_previous("application_to_machine_graph_algorithms") self._error_on_previous("machine_graph_to_machine_algorithms") self._error_on_previous("machine_graph_to_virtual_machine_algorithms") def _error_on_previous(self, option): try: get_config_str_list("Mapping", option) except NoOptionError: # GOOD! return raise ConfigurationException( f"cfg setting {option} is no longer supported! " "See" "Algorithms.html.") def _adjust_config(self, runtime,): """ Adjust and checks config based on runtime :param runtime: :type runtime: int or bool :param frozenset(str) debug_enable_opts: :param frozenset(str) report_disable_opts: :raises ConfigurationException: """ if runtime is None: if get_config_bool("Reports", "write_energy_report"): set_config("Reports", "write_energy_report", "False")"[Reports]write_energy_report has been set to " "False as runtime is set to forever") def _remove_excess_folders( self, max_kept, starting_directory, remove_errored_folders): try: files_in_report_folder = os.listdir(starting_directory) # while there's more than the valid max, remove the oldest one if len(files_in_report_folder) > max_kept: # sort files into time frame files_in_report_folder.sort( key=lambda temp_file: os.path.getmtime( os.path.join(starting_directory, temp_file))) # remove only the number of files required, and only if they # have the finished flag file created num_files_to_remove = len(files_in_report_folder) - max_kept files_removed = 0 files_not_closed = 0 for current_oldest_file in files_in_report_folder: finished_flag = os.path.join(os.path.join( starting_directory, current_oldest_file), FINISHED_FILENAME) errored_flag = os.path.join(os.path.join( starting_directory, current_oldest_file), ERRORED_FILENAME) finished_flag_exists = os.path.exists(finished_flag) errored_flag_exists = os.path.exists(errored_flag) if finished_flag_exists and ( not errored_flag_exists or remove_errored_folders): shutil.rmtree(os.path.join( starting_directory, current_oldest_file), ignore_errors=True) files_removed += 1 else: files_not_closed += 1 if files_removed + files_not_closed >= num_files_to_remove: break if files_not_closed > max_kept // 4: logger.warning( "{} has {} old reports that have not been closed", starting_directory, files_not_closed) except IOError: # This might happen if there is an open file, or more than one # process in the same folder, but we shouldn't die because of it pass def _set_up_report_specifics(self): # clear and clean out folders considered not useful anymore report_dir_path = self._data_writer.get_report_dir_path() if os.listdir(report_dir_path): self._remove_excess_folders( get_config_int("Reports", "max_reports_kept"), report_dir_path, get_config_bool("Reports", "remove_errored_folders")) # store timestamp in latest/time_stamp for provenance reasons timestamp_dir_path = self._data_writer.get_timestamp_dir_path() time_of_run_file_name = os.path.join( timestamp_dir_path, TIMESTAMP_FILENAME) _, timestamp = os.path.split(timestamp_dir_path) with open(time_of_run_file_name, "w", encoding="utf-8") as f: f.writelines(timestamp) f.write("\n") f.write("Traceback of setup call:\n") traceback.print_stack(file=f) def __write_named_file(self, file_name): app_file_name = os.path.join( self._data_writer.get_timestamp_dir_path(), file_name) with open(app_file_name, "w", encoding="utf-8") as f: f.writelines("file_name")
[docs] def write_finished_file(self): """ Write a finished file that allows file removal to only remove folders that are finished. """ self.__write_named_file(FINISHED_FILENAME)
[docs] def write_errored_file(self): """ Writes a errored file that allows file removal to only remove folders that are errored if requested to do so """ self.__write_named_file(ERRORED_FILENAME)