Coverage for ovos_core/transformers.py: 72%

1from typing import Optional, List 

2from ovos_config import Configuration 

3from ovos_plugin_manager.intent_transformers import find_intent_transformer_plugins 

4from ovos_plugin_manager.metadata_transformers import find_metadata_transformer_plugins 

5from ovos_plugin_manager.text_transformers import find_utterance_transformer_plugins 

6 

7from ovos_plugin_manager.templates.pipeline import IntentHandlerMatch 

8from ovos_utils.json_helper import merge_dict 

9from ovos_utils.log import LOG 

10 

11 

12class UtteranceTransformersService: 

13 

14 def __init__(self, bus, config=None): 

15 self.config_core = config or Configuration() 

16 self.loaded_plugins = {} 

17 self.has_loaded = False 

18 self.bus = bus 

19 self.config = self.config_core.get("utterance_transformers") or {} 

20 self.load_plugins() 

21 

22 @staticmethod 

23 def find_plugins(): 

24 return find_utterance_transformer_plugins().items() 

25 

26 def load_plugins(self): 

27 for plug_name, plug in self.find_plugins(): 

28 if plug_name in self.config: 

29 # if disabled skip it 

30 if not self.config[plug_name].get("active", True): 

31 continue 

32 try: 

33 self.loaded_plugins[plug_name] = plug() 

34 LOG.info(f"loaded utterance transformer plugin: {plug_name}") 

35 except Exception as e: 

36 LOG.error(e) 

37 LOG.exception(f"Failed to load utterance transformer plugin: {plug_name}") 

38 

39 @property 

40 def plugins(self): 

41 """ 

42 Return loaded transformers in priority order, such that modules with a 

43 higher `priority` rank are called first and changes from lower ranked 

44 transformers are applied last 

45 

46 A plugin of `priority` 1 will override any existing context keys and 

47 will be the last to modify utterances` 

48 """ 

49 return sorted(self.loaded_plugins.values(), 

50 key=lambda k: k.priority, reverse=True) 

51 

52 def shutdown(self): 

53 for module in self.plugins: 

54 try: 

55 module.shutdown() 

56 except: 

57 pass 

58 

59 def transform(self, utterances: List[str], context: Optional[dict] = None): 

60 context = context or {} 

61 

62 for module in self.plugins: 

63 try: 

64 utterances, data = module.transform(utterances, context) 

65 _safe = {k:v for k,v in data.items() if k != "session"} # no leaking TTS/STT creds in logs  

66 LOG.debug(f"{module.name}: {_safe}") 

67 context = merge_dict(context, data) 

68 except Exception as e: 

69 LOG.warning(f"{module.name} transform exception: {e}") 

70 return utterances, context 

71 

72 

73class MetadataTransformersService: 

74 

75 def __init__(self, bus, config=None): 

76 self.config_core = config or Configuration() 

77 self.loaded_plugins = {} 

78 self.has_loaded = False 

79 self.bus = bus 

80 self.config = self.config_core.get("metadata_transformers") or {} 

81 self.load_plugins() 

82 

83 @staticmethod 

84 def find_plugins(): 

85 return find_metadata_transformer_plugins().items() 

86 

87 def load_plugins(self): 

88 for plug_name, plug in self.find_plugins(): 

89 if plug_name in self.config: 

90 # if disabled skip it 

91 if not self.config[plug_name].get("active", True): 

92 continue 

93 try: 

94 self.loaded_plugins[plug_name] = plug() 

95 LOG.info(f"loaded metadata transformer plugin: {plug_name}") 

96 except Exception as e: 

97 LOG.error(e) 

98 LOG.exception(f"Failed to load metadata transformer plugin: {plug_name}") 

99 

100 @property 

101 def plugins(self): 

102 """ 

103 Return loaded transformers in priority order, such that modules with a 

104 higher `priority` rank are called first and changes from lower ranked 

105 transformers are applied last. 

106 

107 A plugin of `priority` 1 will override any existing context keys 

108 """ 

109 return sorted(self.loaded_plugins.values(), 

110 key=lambda k: k.priority, reverse=True) 

111 

112 def shutdown(self): 

113 for module in self.plugins: 

114 try: 

115 module.shutdown() 

116 except: 

117 pass 

118 

119 def transform(self, context: Optional[dict] = None): 

120 """ 

121 Sequentially applies all loaded metadata transformer plugins to the provided context. 

122 

123 Each plugin's `transform` method is called in order of descending priority, and the resulting data is merged into the context. Sensitive session data is excluded from debug logs. Exceptions raised by plugins are logged as warnings and do not interrupt the transformation process. 

124 

125 Args: 

126 context: Optional dictionary containing metadata to be transformed. 

127 

128 Returns: 

129 The updated context dictionary after all transformations. 

130 """ 

131 context = context or {} 

132 

133 for module in self.plugins: 

134 try: 

135 data = module.transform(context) 

136 _safe = {k:v for k,v in data.items() if k != "session"} # no leaking TTS/STT creds in logs  

137 LOG.debug(f"{module.name}: {_safe}") 

138 context = merge_dict(context, data) 

139 except Exception as e: 

140 LOG.warning(f"{module.name} transform exception: {e}") 

141 return context 

142 

143 

144class IntentTransformersService: 

145 

146 def __init__(self, bus, config=None): 

147 """ 

148 Initializes the IntentTransformersService with the provided message bus and configuration. 

149 

150 Loads and prepares intent transformer plugins based on the configuration, making them ready for use. 

151 """ 

152 self.config_core = config or Configuration() 

153 self.loaded_plugins = {} 

154 self.has_loaded = False 

155 self.bus = bus 

156 self.config = self.config_core.get("intent_transformers") or {} 

157 self.load_plugins() 

158 

159 @staticmethod 

160 def find_plugins(): 

161 """ 

162 Discovers and returns available intent transformer plugins. 

163 

164 Returns: 

165 An iterable of (plugin_name, plugin_class) pairs for all discovered intent transformer plugins. 

166 """ 

167 return find_intent_transformer_plugins().items() 

168 

169 def load_plugins(self): 

170 """ 

171 Loads and initializes enabled intent transformer plugins based on the configuration. 

172 

173 Plugins marked as inactive in the configuration are skipped. Successfully loaded plugins are added to the internal registry, while failures are logged without interrupting the loading process. 

174 """ 

175 for plug_name, plug in self.find_plugins(): 

176 if plug_name in self.config: 

177 # if disabled skip it 

178 if not self.config[plug_name].get("active", True): 

179 continue 

180 try: 

181 self.loaded_plugins[plug_name] = plug() 

182 self.loaded_plugins[plug_name].bind(self.bus) 

183 LOG.info(f"loaded intent transformer plugin: {plug_name}") 

184 except Exception as e: 

185 LOG.error(e) 

186 LOG.exception(f"Failed to load intent transformer plugin: {plug_name}") 

187 

188 @property 

189 def plugins(self): 

190 """ 

191 Returns the loaded intent transformer plugins sorted by priority. 

192 """ 

193 return sorted(self.loaded_plugins.values(), 

194 key=lambda k: k.priority, reverse=True) 

195 

196 def shutdown(self): 

197 """ 

198 Shuts down all loaded plugins, suppressing any exceptions raised during shutdown. 

199 """ 

200 for module in self.plugins: 

201 try: 

202 module.shutdown() 

203 except: 

204 pass 

205 

206 def transform(self, intent: IntentHandlerMatch) -> IntentHandlerMatch: 

207 """ 

208 Sequentially applies all loaded intent transformer plugins to the given intent object. 

209 

210 Each plugin's `transform` method is called in order of priority. Exceptions raised by individual plugins are logged as warnings, and processing continues with the next plugin. The final, transformed intent object is returned. 

211 

212 Args: 

213 intent: The intent match object to be transformed. 

214 

215 Returns: 

216 The transformed intent match object after all plugins have been applied. 

217 """ 

218 for module in self.plugins: 

219 try: 

220 intent = module.transform(intent) 

221 LOG.debug(f"{module.name}: {intent}") 

222 except Exception as e: 

223 LOG.warning(f"{module.name} transform exception: {e}") 

224 return intent