Source code for machine_common_sense

from . import import_override  # isort: skip

import concurrent.futures
import json
import logging
import logging.config
import signal
from contextlib import contextmanager
from typing import Dict, Union

import typeguard

from ._version import __version__
from .action import Action
from .config_manager import ConfigManager
from .controller import Controller
from .goal_metadata import GoalCategory, GoalMetadata
from .history_writer import HistoryWriter
from .logging_config import LoggingConfig
from .material import Material
from .object_metadata import ObjectMetadata
from .return_status import ReturnStatus
from .reward import Reward
from .scene_history import SceneHistory
from .serializer import SerializerMsgPack
from .step_metadata import StepMetadata
from .stringifier import Stringifier
from .subscriber import add_subscribers
from .unity_executable_provider import UnityExecutableProvider

logger = logging.getLogger(__name__)
# Set default logging handler to avoid "No handler found" warnings
logger.addHandler(logging.NullHandler())


[docs]def get_controller(unity_exec: str, config: ConfigManager): """Function to get the controller, pulled into its own function so we can time it. """ controller = Controller(unity_exec, config) return controller
[docs]def get_controller_with_timeout(unity_exec: str, config: ConfigManager): """Wrapper function that sets a timeout for the controller creation. If getting the controller times out, None is returned. """ with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: future = executor.submit(get_controller, unity_exec, config) try: controller = future.result(timeout=config.get_controller_timeout()) return controller except concurrent.futures.TimeoutError as Msg: logger.error("Timeout error in creating controller", exc_info=Msg) # TODO: Add checks to continue waiting for the controller creation. return None
[docs]@typeguard.typechecked def init_logging(log_config: Dict = None, log_config_file: str = "log.config.user.py"): """ Initializes logging system. If no parameters are provided, a default configuration will be applied. See python logging documentation for details. https://docs.python.org/3/library/logging.config.html#logging-config-dictschema Parameters ---------- log_config : dict, optional A dictionary the contains the logging configuration. If None, a default configuration will be used log_config_file: str, optional Path to an override configuration file. The file will contain a python dictionary for the logging configuration. This file is typically not used, but allows a user to change the logging configuration without code changes. Default, log.config.user.py """ LoggingConfig.init_logging( log_config=log_config, log_config_file=log_config_file)
[docs]@typeguard.typechecked def create_controller(config_file_or_dict: Union[Dict, str] = None, unity_app_file_path: str = None, unity_cache_version: str = None): """ Creates and returns a new MCS Controller object. Parameters ---------- config_file_or_dict: str or dict, required Can be a path to configuration file to read in or a dictionary of various properties, such as metadata level and whether or not to save history files (default None) * Note the **order of precedence for config options**, in case more than one is given: 1. **MCS_CONFIG_FILE_PATH** environment variable (meant for internal TA2 use during evaluation) 2. If no environment variable given, use **config_file_or_dict** parameter. The value can be a string file path or a dictionary. 3. Raises FileNotFoundError if no config found. unity_app_file_path : str, optional The file path to your MCS Unity application. If Not provided, the internal cache and downloader will attempt to locate and use the current version. (default None) unity_cache_version : str, optional If no file path is provided for the MCS Unity application, the version provided will be found via cache and internal downloader. If not provided, the version matching the MCS code will be used. (default None) Returns ------- Controller The MCS Controller object. """ try: unity_exec = unity_app_file_path if (unity_exec is None): unity_provider = UnityExecutableProvider() unity_exec = unity_provider.get_executable( unity_cache_version).as_posix() config = ConfigManager(config_file_or_dict) controller = get_controller_with_timeout(unity_exec, config) if not controller: raise Exception('MCS/Python Controller failed to initialize') add_subscribers(controller, config) return controller except Exception as Msg: logger.error("Exception in create_controller()", exc_info=Msg) return None
[docs]@typeguard.typechecked def change_config(controller: Controller, config_file_or_dict: Union[Dict, str] = None): """ Creates and returns a new MCS Controller object. Should only be called After a run and before a scene is changed. Parameters ---------- controller : Controller The currently used controller that the config should be changed on. config_file_or_dict: str or dict, optional Can be a path to configuration file to read in or a dictionary of various properties, such as metadata level and whether or not to save history files (default None) """ config = ConfigManager(config_file_or_dict) controller._set_config(config) controller.remove_all_event_handlers() add_subscribers(controller, config)
[docs]@typeguard.typechecked def load_scene_json_file(scene_json_file_path: str) -> Dict: """ Loads the given JSON scene config file and returns its data. Parameters ---------- config_json_file_path : str The file path to your MCS JSON scene configuration file. Returns ------- dict The MCS scene configuration data from the given JSON file. Raises ------ FileNotFoundError ValueError """ with open(scene_json_file_path, encoding='utf-8-sig') \ as config_json_file_object: return json.load(config_json_file_object)