| 20 | |
| 21 | |
| 22 | def load_config() -> Dict[str, Any]: |
| 23 | _ = coordinator.translator.gettext |
| 24 | # Include custom channels |
| 25 | custom_channel_path = str(utils.get_custom_modules_path()) |
| 26 | if custom_channel_path not in sys.path: |
| 27 | sys.path.insert(0, custom_channel_path) |
| 28 | |
| 29 | conf_path = utils.get_config_path() |
| 30 | if not conf_path.exists(): |
| 31 | raise FileNotFoundError(_("Config File does not exist. ({})").format(conf_path)) |
| 32 | with conf_path.open() as f: |
| 33 | data: Dict[str, Any] = OPTIONAL_DEFAULTS.copy() |
| 34 | data.update(YAML().load(f)) |
| 35 | |
| 36 | # Verify configuration |
| 37 | |
| 38 | # - Master channel |
| 39 | master_channel_id = data.get("master_channel", None) |
| 40 | if not master_channel_id: |
| 41 | raise ValueError(_("Master Channel is not specified in the profile config.")) |
| 42 | elif not isinstance(master_channel_id, str): |
| 43 | raise ValueError(_("Master Channel ID is expected to be a string, but " |
| 44 | "\"{0}\" is of type {1}.").format(master_channel_id, type(master_channel_id))) |
| 45 | channel = utils.locate_module(data['master_channel'], 'master') |
| 46 | if not channel: |
| 47 | raise ValueError(_("\"{}\" is not found.").format(master_channel_id)) |
| 48 | if not issubclass(channel, MasterChannel): |
| 49 | raise ValueError(_("\"{0}\" is not a master channel, but a {1}.").format(master_channel_id, channel)) |
| 50 | |
| 51 | # - Slave channels |
| 52 | slave_channels_list = data.get("slave_channels", None) |
| 53 | if not slave_channels_list: |
| 54 | raise ValueError(_("Slave Channels are not specified in the profile config.")) |
| 55 | elif not isinstance(slave_channels_list, list): |
| 56 | raise ValueError(_("Slave Channel IDs are expected to be a list, but {} is found.") |
| 57 | .format(slave_channels_list)) |
| 58 | for i in slave_channels_list: |
| 59 | channel = utils.locate_module(i, 'slave') |
| 60 | if not channel: |
| 61 | raise ValueError(_("\"{}\" is not found.").format(i)) |
| 62 | if not issubclass(channel, SlaveChannel): |
| 63 | raise ValueError(_("\"{0}\" is not a slave channel, but a {1}.").format(i, channel)) |
| 64 | |
| 65 | # - Middlewares |
| 66 | middlewares_list = data.get("middlewares", None) |
| 67 | if middlewares_list is not None: |
| 68 | if not isinstance(middlewares_list, list): |
| 69 | raise ValueError(_("Middleware IDs must be a list, but a {} is found.") |
| 70 | .format(type(middlewares_list))) |
| 71 | for i in middlewares_list: |
| 72 | middleware = utils.locate_module(i, 'middleware') |
| 73 | if not middleware: |
| 74 | raise ValueError(_("\"{}\" is not found.").format(i)) |
| 75 | if not issubclass(middleware, Middleware): |
| 76 | raise ValueError(_("\"{0}\" is not a middleware, but a {1}.") |
| 77 | .format(i, middleware)) |
| 78 | else: |
| 79 | data['middlewares'] = list() |