1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
|
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional, Dict, Any
import json
import os
import shutil
import zipfile
import tempfile
from PyQt5.QtCore import pyqtSlot, QObject, pyqtSignal, QUrl
from UM.Application import Application
from UM.Logger import Logger
from UM.Resources import Resources
from UM.Version import Version
class CuraPackageManager(QObject):
Version = 1
def __init__(self, parent = None):
super().__init__(parent)
self._application = Application.getInstance()
self._container_registry = self._application.getContainerRegistry()
self._plugin_registry = self._application.getPluginRegistry()
#JSON files that keep track of all installed packages.
self._user_package_management_file_path = None
self._bundled_package_management_file_path = None
for search_path in Resources.getSearchPaths():
candidate_bundled_path = os.path.join(search_path, "bundled_packages.json")
if os.path.exists(candidate_bundled_path):
self._bundled_package_management_file_path = candidate_bundled_path
for search_path in (Resources.getDataStoragePath(), Resources.getConfigStoragePath()):
candidate_user_path = os.path.join(search_path, "packages.json")
if os.path.exists(candidate_user_path):
self._user_package_management_file_path = candidate_user_path
if self._user_package_management_file_path is None: #Doesn't exist yet.
self._user_package_management_file_path = os.path.join(Resources.getDataStoragePath(), "packages.json")
self._bundled_package_dict = {} # A dict of all bundled packages
self._installed_package_dict = {} # A dict of all installed packages
self._to_remove_package_set = set() # A set of packages that need to be removed at the next start
self._to_install_package_dict = {} # A dict of packages that need to be installed at the next start
installedPackagesChanged = pyqtSignal() # Emitted whenever the installed packages collection have been changed.
def initialize(self):
self._loadManagementData()
self._removeAllScheduledPackages()
self._installAllScheduledPackages()
# (for initialize) Loads the package management file if exists
def _loadManagementData(self) -> None:
# The bundles package management file should always be there
if not os.path.exists(self._bundled_package_management_file_path):
Logger.log("w", "Bundled package management file could not be found!")
return
# Load the bundled packages:
with open(self._bundled_package_management_file_path, "r", encoding = "utf-8") as f:
self._bundled_package_dict = json.load(f, encoding = "utf-8")
Logger.log("i", "Loaded bundled packages data from %s", self._bundled_package_management_file_path)
# Load the user package management file
if not os.path.exists(self._user_package_management_file_path):
Logger.log("i", "User package management file %s doesn't exist, do nothing", self._user_package_management_file_path)
return
# Need to use the file lock here to prevent concurrent I/O from other processes/threads
container_registry = self._application.getContainerRegistry()
with container_registry.lockFile():
# Load the user packages:
with open(self._user_package_management_file_path, "r", encoding="utf-8") as f:
management_dict = json.load(f, encoding="utf-8")
self._installed_package_dict = management_dict.get("installed", {})
self._to_remove_package_set = set(management_dict.get("to_remove", []))
self._to_install_package_dict = management_dict.get("to_install", {})
Logger.log("i", "Loaded user packages management file from %s", self._user_package_management_file_path)
def _saveManagementData(self) -> None:
# Need to use the file lock here to prevent concurrent I/O from other processes/threads
container_registry = self._application.getContainerRegistry()
with container_registry.lockFile():
with open(self._user_package_management_file_path, "w", encoding = "utf-8") as f:
data_dict = {"version": CuraPackageManager.Version,
"installed": self._installed_package_dict,
"to_remove": list(self._to_remove_package_set),
"to_install": self._to_install_package_dict}
json.dump(data_dict, f, sort_keys = True, indent = 4)
Logger.log("i", "Package management file %s was saved", self._user_package_management_file_path)
# (for initialize) Removes all packages that have been scheduled to be removed.
def _removeAllScheduledPackages(self) -> None:
for package_id in self._to_remove_package_set:
self._purgePackage(package_id)
del self._installed_package_dict[package_id]
self._to_remove_package_set.clear()
self._saveManagementData()
# (for initialize) Installs all packages that have been scheduled to be installed.
def _installAllScheduledPackages(self) -> None:
while self._to_install_package_dict:
package_id, package_info = list(self._to_install_package_dict.items())[0]
self._installPackage(package_info)
del self._to_install_package_dict[package_id]
self._saveManagementData()
def getBundledPackageInfo(self, package_id: str) -> Optional[dict]:
package_info = None
if package_id in self._bundled_package_dict:
package_info = self._bundled_package_dict[package_id]["package_info"]
return package_info
# Checks the given package is installed. If so, return a dictionary that contains the package's information.
def getInstalledPackageInfo(self, package_id: str) -> Optional[dict]:
if package_id in self._to_remove_package_set:
return None
if package_id in self._to_install_package_dict:
package_info = self._to_install_package_dict[package_id]["package_info"]
return package_info
if package_id in self._installed_package_dict:
package_info = self._installed_package_dict[package_id]["package_info"]
return package_info
if package_id in self._bundled_package_dict:
package_info = self._bundled_package_dict[package_id]["package_info"]
return package_info
return None
def getAllInstalledPackageIDs(self) -> set:
# Add bundled, installed, and to-install packages to the set of installed package IDs
all_installed_ids = set()
if self._bundled_package_dict.keys():
all_installed_ids = all_installed_ids.union(set(self._bundled_package_dict.keys()))
if self._installed_package_dict.keys():
all_installed_ids = all_installed_ids.union(set(self._installed_package_dict.keys()))
all_installed_ids = all_installed_ids.difference(self._to_remove_package_set)
# If it's going to be installed and to be removed, then the package is being updated and it should be listed.
if self._to_install_package_dict.keys():
all_installed_ids = all_installed_ids.union(set(self._to_install_package_dict.keys()))
return all_installed_ids
def getAllInstalledPackagesInfo(self) -> dict:
all_installed_ids = self.getAllInstalledPackageIDs()
# map of <package_type> -> <package_id> -> <package_info>
installed_packages_dict = {}
for package_id in all_installed_ids:
# Skip required plugins as they should not be tampered with
if package_id in Application.getInstance().getRequiredPlugins():
continue
package_info = None
# Add bundled plugins
if package_id in self._bundled_package_dict:
package_info = self._bundled_package_dict[package_id]["package_info"]
package_info["is_installed"] = True
# Add installed plugins
if package_id in self._installed_package_dict:
package_info = self._installed_package_dict[package_id]["package_info"]
package_info["is_installed"] = True
# Add to install plugins
if package_id in self._to_install_package_dict:
package_info = self._to_install_package_dict[package_id]["package_info"]
package_info["is_installed"] = False
if package_info is None:
continue
# We also need to get information from the plugin registry such as if a plugin is active
package_info["is_active"] = self._plugin_registry.isActivePlugin(package_id)
# If the package ID is in bundled, label it as such
package_info["is_bundled"] = package_info["package_id"] in self._bundled_package_dict.keys() and not self.isUserInstalledPackage(package_info["package_id"])
# If there is not a section in the dict for this type, add it
if package_info["package_type"] not in installed_packages_dict:
installed_packages_dict[package_info["package_type"]] = []
# Finally, add the data
installed_packages_dict[package_info["package_type"]].append(package_info)
return installed_packages_dict
# Checks if the given package is installed (at all).
def isPackageInstalled(self, package_id: str) -> bool:
return self.getInstalledPackageInfo(package_id) is not None
# This is called by drag-and-dropping curapackage files.
@pyqtSlot(QUrl)
def installPackageViaDragAndDrop(self, file_url: str) -> None:
filename = QUrl(file_url).toLocalFile()
return self.installPackage(filename)
# Schedules the given package file to be installed upon the next start.
@pyqtSlot(str)
def installPackage(self, filename: str) -> None:
has_changes = False
try:
# Get package information
package_info = self.getPackageInfo(filename)
if not package_info:
return
package_id = package_info["package_id"]
# Check if it is installed
installed_package_info = self.getInstalledPackageInfo(package_info["package_id"])
to_install_package = installed_package_info is None # Install if the package has not been installed
if installed_package_info is not None:
# Compare versions and only schedule the installation if the given package is newer
new_version = package_info["package_version"]
installed_version = installed_package_info["package_version"]
if Version(new_version) > Version(installed_version):
Logger.log("i", "Package [%s] version [%s] is newer than the installed version [%s], update it.",
package_id, new_version, installed_version)
to_install_package = True
if to_install_package:
# Need to use the lock file to prevent concurrent I/O issues.
with self._container_registry.lockFile():
Logger.log("i", "Package [%s] version [%s] is scheduled to be installed.",
package_id, package_info["package_version"])
# Copy the file to cache dir so we don't need to rely on the original file to be present
package_cache_dir = os.path.join(os.path.abspath(Resources.getCacheStoragePath()), "cura_packages")
if not os.path.exists(package_cache_dir):
os.makedirs(package_cache_dir, exist_ok=True)
target_file_path = os.path.join(package_cache_dir, package_id + ".curapackage")
shutil.copy2(filename, target_file_path)
self._to_install_package_dict[package_id] = {"package_info": package_info,
"filename": target_file_path}
has_changes = True
except:
Logger.logException("c", "Failed to install package file '%s'", filename)
finally:
self._saveManagementData()
if has_changes:
self.installedPackagesChanged.emit()
# Schedules the given package to be removed upon the next start.
# \param package_id id of the package
# \param force_add is used when updating. In that case you actually want to uninstall & install
@pyqtSlot(str)
def removePackage(self, package_id: str, force_add: bool = False) -> None:
# Check the delayed installation and removal lists first
if not self.isPackageInstalled(package_id):
Logger.log("i", "Attempt to remove package [%s] that is not installed, do nothing.", package_id)
return
# Extra safety check
if package_id not in self._installed_package_dict and package_id in self._bundled_package_dict:
Logger.log("i", "Not uninstalling [%s] because it is a bundled package.")
return
if package_id not in self._to_install_package_dict or force_add:
# Schedule for a delayed removal:
self._to_remove_package_set.add(package_id)
else:
if package_id in self._to_install_package_dict:
# Remove from the delayed installation list if present
del self._to_install_package_dict[package_id]
self._saveManagementData()
self.installedPackagesChanged.emit()
## Is the package an user installed package?
def isUserInstalledPackage(self, package_id: str):
return package_id in self._installed_package_dict
# Removes everything associated with the given package ID.
def _purgePackage(self, package_id: str) -> None:
# Iterate through all directories in the data storage directory and look for sub-directories that belong to
# the package we need to remove, that is the sub-dirs with the package_id as names, and remove all those dirs.
data_storage_dir = os.path.abspath(Resources.getDataStoragePath())
for root, dir_names, _ in os.walk(data_storage_dir):
for dir_name in dir_names:
package_dir = os.path.join(root, dir_name, package_id)
if os.path.exists(package_dir):
Logger.log("i", "Removing '%s' for package [%s]", package_dir, package_id)
shutil.rmtree(package_dir)
break
# Installs all files associated with the given package.
def _installPackage(self, installation_package_data: dict):
package_info = installation_package_data["package_info"]
filename = installation_package_data["filename"]
package_id = package_info["package_id"]
if not os.path.exists(filename):
Logger.log("w", "Package [%s] file '%s' is missing, cannot install this package", package_id, filename)
return
Logger.log("i", "Installing package [%s] from file [%s]", package_id, filename)
# remove it first and then install
self._purgePackage(package_id)
# Install the package
with zipfile.ZipFile(filename, "r") as archive:
temp_dir = tempfile.TemporaryDirectory()
archive.extractall(temp_dir.name)
from cura.CuraApplication import CuraApplication
installation_dirs_dict = {
"materials": Resources.getStoragePath(CuraApplication.ResourceTypes.MaterialInstanceContainer),
"qualities": Resources.getStoragePath(CuraApplication.ResourceTypes.QualityInstanceContainer),
"plugins": os.path.abspath(Resources.getStoragePath(Resources.Plugins)),
}
for sub_dir_name, installation_root_dir in installation_dirs_dict.items():
src_dir_path = os.path.join(temp_dir.name, "files", sub_dir_name)
dst_dir_path = os.path.join(installation_root_dir, package_id)
if not os.path.exists(src_dir_path):
continue
self.__installPackageFiles(package_id, src_dir_path, dst_dir_path)
# Remove the file
os.remove(filename)
# Move the info to the installed list of packages only when it succeeds
self._installed_package_dict[package_id] = self._to_install_package_dict[package_id]
def __installPackageFiles(self, package_id: str, src_dir: str, dst_dir: str) -> None:
Logger.log("i", "Moving package {package_id} from {src_dir} to {dst_dir}".format(package_id=package_id, src_dir=src_dir, dst_dir=dst_dir))
shutil.move(src_dir, dst_dir)
# Gets package information from the given file.
def getPackageInfo(self, filename: str) -> Dict[str, Any]:
with zipfile.ZipFile(filename) as archive:
try:
# All information is in package.json
with archive.open("package.json") as f:
package_info_dict = json.loads(f.read().decode("utf-8"))
return package_info_dict
except Exception as e:
Logger.logException("w", "Could not get package information from file '%s': %s" % (filename, e))
return {}
# Gets the license file content if present in the given package file.
# Returns None if there is no license file found.
def getPackageLicense(self, filename: str) -> Optional[str]:
license_string = None
with zipfile.ZipFile(filename) as archive:
# Go through all the files and use the first successful read as the result
for file_info in archive.infolist():
if file_info.filename.endswith("LICENSE"):
Logger.log("d", "Found potential license file '%s'", file_info.filename)
try:
with archive.open(file_info.filename, "r") as f:
data = f.read()
license_string = data.decode("utf-8")
break
except:
Logger.logException("e", "Failed to load potential license file '%s' as text file.",
file_info.filename)
license_string = None
return license_string
|