Coverage for ovos_core/skill_installer.py: 57%
204 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-17 13:44 +0000
1import enum
2import shutil
3import sys
4from importlib import reload
5from os.path import exists
6from subprocess import Popen, PIPE
7from typing import Optional
9import requests
10from combo_lock import NamedLock
11from ovos_bus_client import Message
12from ovos_config.config import Configuration
13from ovos_utils.log import LOG
15import ovos_plugin_manager
18class InstallError(str, enum.Enum):
19 DISABLED = "pip disabled in mycroft.conf"
20 PIP_ERROR = "error in pip subprocess"
21 BAD_URL = "skill url validation failed"
22 NO_PKGS = "no packages to install"
25class SkillsStore:
26 # default constraints to use if none are given
27 DEFAULT_CONSTRAINTS = 'https://raw.githubusercontent.com/OpenVoiceOS/ovos-releases/refs/heads/main/constraints-stable.txt'
28 PIP_LOCK = NamedLock("ovos_pip.lock")
29 UV = shutil.which("uv") # use 'uv pip' if available, speeds things up a lot and is the default in raspOVOS
31 def __init__(self, bus, config=None):
32 self.config = config or Configuration().get("skills", {}).get("installer", {})
33 self.bus = bus
34 self.bus.on("ovos.skills.install", self.handle_install_skill)
35 self.bus.on("ovos.skills.uninstall", self.handle_uninstall_skill)
36 self.bus.on("ovos.pip.install", self.handle_install_python)
37 self.bus.on("ovos.pip.uninstall", self.handle_uninstall_python)
39 def shutdown(self):
40 self.bus.remove("ovos.skills.install", self.handle_install_skill)
41 self.bus.remove("ovos.skills.uninstall", self.handle_uninstall_skill)
42 self.bus.remove("ovos.pip.install", self.handle_install_python)
43 self.bus.remove("ovos.pip.uninstall", self.handle_uninstall_python)
45 def play_error_sound(self):
46 snd = self.config.get("sounds", {}).get("pip_error", "snd/error.mp3")
47 self.bus.emit(Message("mycroft.audio.play_sound", {"uri": snd}))
49 def play_success_sound(self):
50 snd = self.config.get("sounds", {}).get("pip_success", "snd/acknowledge.mp3")
51 self.bus.emit(Message("mycroft.audio.play_sound", {"uri": snd}))
53 @staticmethod
54 def validate_constrainsts(constraints: str):
55 if constraints.startswith('http'):
56 LOG.debug(f"Constraints url: {constraints}")
57 try:
58 response = requests.head(constraints)
59 if response.status_code != 200:
60 LOG.error(f'Remote constraints file not accessible: {response.status_code}')
61 return False
62 return True
63 except Exception as e:
64 LOG.error(f'Error accessing remote constraints: {str(e)}')
65 return False
67 # Use constraints to limit the installed versions
68 if not exists(constraints):
69 LOG.error('Couldn\'t find the constraints file')
70 return False
72 return True
74 def pip_install(self, packages: list,
75 constraints: Optional[str] = None,
76 print_logs: bool = True):
77 if not len(packages):
78 LOG.error("no package list provided to install")
79 self.play_error_sound()
80 return False
82 # can be set in mycroft.conf to change to testing/alpha channels
83 constraints = constraints or self.config.get("constraints", SkillsStore.DEFAULT_CONSTRAINTS)
85 if not self.validate_constrainsts(constraints):
86 self.play_error_sound()
87 return False
89 if self.UV is not None:
90 pip_args = [self.UV, 'pip', 'install']
91 else:
92 pip_args = [sys.executable, '-m', 'pip', 'install']
93 if constraints:
94 pip_args += ['-c', constraints]
95 if self.config.get("break_system_packages", False):
96 pip_args += ["--break-system-packages"]
97 if self.config.get("allow_alphas", False):
98 pip_args += ["--pre"]
100 with SkillsStore.PIP_LOCK:
101 """
102 Iterate over the individual Python packages and
103 install them one by one to enforce the order specified
104 in the manifest.
105 """
106 for dependent_python_package in packages:
107 LOG.info("(pip) Installing " + dependent_python_package)
108 pip_command = pip_args + [dependent_python_package]
109 LOG.debug(" ".join(pip_command))
110 if print_logs:
111 proc = Popen(pip_command)
112 else:
113 proc = Popen(pip_command, stdout=PIPE, stderr=PIPE)
114 pip_code = proc.wait()
115 if pip_code != 0:
116 stderr = proc.stderr
117 if stderr:
118 stderr = stderr.read().decode()
119 self.play_error_sound()
120 raise RuntimeError(stderr)
122 reload(ovos_plugin_manager) # force core to pick new entry points
123 self.play_success_sound()
124 return True
126 def pip_uninstall(self, packages: list,
127 constraints: Optional[str] = None,
128 print_logs: bool = True):
129 if not len(packages):
130 LOG.error("no package list provided to uninstall")
131 self.play_error_sound()
132 return False
134 # can be set in mycroft.conf to change to testing/alpha channels
135 constraints = constraints or self.config.get("constraints", SkillsStore.DEFAULT_CONSTRAINTS)
137 if not self.validate_constrainsts(constraints):
138 self.play_error_sound()
139 return False
141 # get protected packages that can't be uninstalled
142 # by default cant uninstall any official ovos package via this bus api
143 if constraints.startswith("http"):
144 cpkgs = requests.get(constraints).text.split("\n")
145 elif exists(constraints):
146 with open(constraints) as f:
147 cpkgs = f.read().split("\n")
148 else:
149 cpkgs = ["ovos-core", "ovos-utils", "ovos-plugin-manager",
150 "ovos-config", "ovos-bus-client", "ovos-workshop"]
152 # remove version pinning and normalize _ to - (pip accepts both)
153 cpkgs = [p.split("~")[0].split("<")[0].split(">")[0].split("=")[0].replace("_", "-")
154 for p in cpkgs]
156 if any(p in cpkgs for p in packages):
157 LOG.error(f'tried to uninstall a protected package: {cpkgs}')
158 self.play_error_sound()
159 return False
161 if self.UV is not None:
162 pip_args = [self.UV, 'pip', 'uninstall']
163 else:
164 pip_args = [sys.executable, '-m', 'pip', 'uninstall', '-y']
165 if self.config.get("break_system_packages", False):
166 pip_args += ["--break-system-packages"]
168 with SkillsStore.PIP_LOCK:
169 """
170 Iterate over the individual Python packages and
171 install them one by one to enforce the order specified
172 in the manifest.
173 """
174 for dependent_python_package in packages:
175 LOG.info("(pip) Uninstalling " + dependent_python_package)
176 pip_command = pip_args + [dependent_python_package]
177 LOG.debug(" ".join(pip_command))
178 if print_logs:
179 proc = Popen(pip_command)
180 else:
181 proc = Popen(pip_command, stdout=PIPE, stderr=PIPE)
182 pip_code = proc.wait()
183 if pip_code != 0:
184 stderr = proc.stderr.read().decode()
185 self.play_error_sound()
186 raise RuntimeError(stderr)
188 reload(ovos_plugin_manager) # force core to pick new entry points
189 self.play_success_sound()
190 return True
192 @staticmethod
193 def validate_skill(url):
194 if not url.startswith("https://github.com/"):
195 return False
196 # TODO - check if setup.py
197 # TODO - check if not using MycroftSkill class
198 # TODO - check if not mycroft CommonPlay
199 return True
201 def handle_install_skill(self, message: Message):
202 if not self.config.get("allow_pip"):
203 LOG.error(InstallError.DISABLED.value)
204 self.play_error_sound()
205 self.bus.emit(message.reply("ovos.skills.install.failed",
206 {"error": InstallError.DISABLED.value}))
207 return
209 url = message.data["url"]
210 if self.validate_skill(url):
211 success = self.pip_install([f"git+{url}"])
212 if success:
213 self.bus.emit(message.reply("ovos.skills.install.complete"))
214 else:
215 self.bus.emit(message.reply("ovos.skills.install.failed",
216 {"error": InstallError.PIP_ERROR.value}))
217 else:
218 LOG.error("invalid skill url, does not appear to be a github skill")
219 self.play_error_sound()
220 self.bus.emit(message.reply("ovos.skills.install.failed",
221 {"error": InstallError.BAD_URL.value}))
223 def handle_uninstall_skill(self, message: Message):
224 if not self.config.get("allow_pip"):
225 LOG.error(InstallError.DISABLED.value)
226 self.play_error_sound()
227 self.bus.emit(message.reply("ovos.skills.uninstall.failed",
228 {"error": InstallError.DISABLED.value}))
229 return
230 # TODO
231 LOG.error("pip uninstall not yet implemented")
232 self.play_error_sound()
233 self.bus.emit(message.reply("ovos.skills.uninstall.failed",
234 {"error": "not implemented"}))
236 def handle_install_python(self, message: Message):
237 if not self.config.get("allow_pip"):
238 LOG.error(InstallError.DISABLED.value)
239 self.play_error_sound()
240 self.bus.emit(message.reply("ovos.pip.install.failed",
241 {"error": InstallError.DISABLED.value}))
242 return
243 pkgs = message.data.get("packages")
244 if pkgs:
245 if self.pip_install(pkgs):
246 self.bus.emit(message.reply("ovos.pip.install.complete"))
247 else:
248 self.bus.emit(message.reply("ovos.pip.install.failed",
249 {"error": InstallError.PIP_ERROR.value}))
250 else:
251 self.bus.emit(message.reply("ovos.pip.install.failed",
252 {"error": InstallError.NO_PKGS.value}))
254 def handle_uninstall_python(self, message: Message):
255 if not self.config.get("allow_pip"):
256 LOG.error(InstallError.DISABLED.value)
257 self.play_error_sound()
258 self.bus.emit(message.reply("ovos.pip.uninstall.failed",
259 {"error": InstallError.DISABLED.value}))
260 return
261 pkgs = message.data.get("packages")
262 if pkgs:
263 if self.pip_uninstall(pkgs):
264 self.bus.emit(message.reply("ovos.pip.uninstall.complete"))
265 else:
266 self.bus.emit(message.reply("ovos.pip.uninstall.failed",
267 {"error": InstallError.PIP_ERROR.value}))
268 else:
269 self.bus.emit(message.reply("ovos.pip.uninstall.failed",
270 {"error": InstallError.NO_PKGS.value}))
273def launch_standalone():
274 # TODO - add docker detection and warn user
275 from ovos_bus_client import MessageBusClient
276 from ovos_utils import wait_for_exit_signal
277 from ovos_utils.log import init_service_logger
279 LOG.info("Launching SkillsStore in standalone mode")
280 init_service_logger("skill-installer")
282 bus = MessageBusClient()
283 bus.run_in_thread()
284 bus.connected_event.wait()
286 store = SkillsStore(bus)
288 wait_for_exit_signal()
290 store.shutdown()
292 LOG.info('SkillsStore shutdown complete!')
295if __name__ == "__main__":
296 launch_standalone()