1# Copyright (c) 2019 Ultimaker B.V. 2# Uranium is released under the terms of the LGPLv3 or higher. 3 4import collections 5import gc 6import os 7import pickle #For serializing/deserializing Python classes to binary files 8import re #For finding containers with asterisks in the constraints and for detecting backup files. 9import time 10from typing import Any, cast, Dict, List, Optional, Set, Type, TYPE_CHECKING 11 12import UM.Dictionary 13import UM.FlameProfiler 14from UM.LockFile import LockFile 15from UM.Logger import Logger 16from UM.MimeTypeDatabase import MimeType, MimeTypeDatabase 17from UM.PluginRegistry import PluginRegistry #To register the container type plug-ins and container provider plug-ins. 18from UM.Resources import Resources 19from UM.Settings.EmptyInstanceContainer import EmptyInstanceContainer 20from UM.Settings.ContainerFormatError import ContainerFormatError 21from UM.Settings.ContainerProvider import ContainerProvider 22from UM.Settings.constant_instance_containers import empty_container 23from . import ContainerQuery 24from UM.Settings.ContainerStack import ContainerStack 25from UM.Settings.DefinitionContainer import DefinitionContainer 26from UM.Settings.InstanceContainer import InstanceContainer 27from UM.Settings.Interfaces import ContainerInterface, ContainerRegistryInterface, DefinitionContainerInterface 28from UM.Signal import Signal, signalemitter 29 30if TYPE_CHECKING: 31 from UM.PluginObject import PluginObject 32 from UM.Qt.QtApplication import QtApplication 33 34 35@signalemitter 36class ContainerRegistry(ContainerRegistryInterface): 37 """Central class to manage all setting providers. 38 39 This class aggregates all data from all container providers. If only the 40 metadata is used, it requests the metadata lazily from the providers. If 41 more than that is needed, the entire container is requested from the 42 appropriate providers. 43 """ 44 45 def __init__(self, application: "QtApplication") -> None: 46 if ContainerRegistry.__instance is not None: 47 raise RuntimeError("Try to create singleton '%s' more than once" % self.__class__.__name__) 48 ContainerRegistry.__instance = self 49 50 super().__init__() 51 52 self._application = application # type: QtApplication 53 54 self._emptyInstanceContainer = empty_container # type: InstanceContainer 55 56 # Sorted list of container providers (keep it sorted by sorting each time you add one!). 57 self._providers = [] # type: List[ContainerProvider] 58 PluginRegistry.addType("container_provider", self.addProvider) 59 60 self.metadata = {} # type: Dict[str, Dict[str, Any]] 61 self._containers = {} # type: Dict[str, ContainerInterface] 62 self._wrong_container_ids = set() # type: Set[str] # Set of already known wrong containers that must be skipped 63 self.source_provider = {} # type: Dict[str, Optional[ContainerProvider]] # Where each container comes from. 64 # Ensure that the empty container is added to the ID cache. 65 self.metadata["empty"] = self._emptyInstanceContainer.getMetaData() 66 self._containers["empty"] = self._emptyInstanceContainer 67 self.source_provider["empty"] = None 68 self._resource_types = {"definition": Resources.DefinitionContainers} # type: Dict[str, int] 69 70 #Since queries are based on metadata, we need to make sure to clear the cache when a container's metadata changes. 71 self.containerMetaDataChanged.connect(self._clearQueryCache) 72 73 self._explicit_read_only_container_ids = set() # type: Set[str] 74 75 containerAdded = Signal() 76 containerRemoved = Signal() 77 containerMetaDataChanged = Signal() 78 containerLoadComplete = Signal() 79 allMetadataLoaded = Signal() 80 81 def addResourceType(self, resource_type: int, container_type: str) -> None: 82 self._resource_types[container_type] = resource_type 83 84 def getResourceTypes(self) -> Dict[str, int]: 85 """Returns all resource types.""" 86 87 return self._resource_types 88 89 def getDefaultSaveProvider(self) -> "ContainerProvider": 90 if len(self._providers) == 1: 91 return self._providers[0] 92 raise NotImplementedError("Not implemented default save provider for multiple providers") 93 94 def addWrongContainerId(self, wrong_container_id: str) -> None: 95 """This method adds the current id to the list of wrong containers that are skipped when looking for a container""" 96 97 self._wrong_container_ids.add(wrong_container_id) 98 99 def addProvider(self, provider: ContainerProvider) -> None: 100 """Adds a container provider to search through containers in.""" 101 102 self._providers.append(provider) 103 # Re-sort every time. It's quadratic, but there shouldn't be that many providers anyway... 104 self._providers.sort(key = lambda provider: PluginRegistry.getInstance().getMetaData(provider.getPluginId())["container_provider"].get("priority", 0)) 105 106 def findDefinitionContainers(self, **kwargs: Any) -> List[DefinitionContainerInterface]: 107 """Find all DefinitionContainer objects matching certain criteria. 108 109 :param dict kwargs: A dictionary of keyword arguments containing 110 keys and values that need to match the metadata of the 111 DefinitionContainer. An asterisk in the values can be used to denote a 112 wildcard. 113 """ 114 115 return cast(List[DefinitionContainerInterface], self.findContainers(container_type = DefinitionContainer, **kwargs)) 116 117 def findDefinitionContainersMetadata(self, **kwargs: Any) -> List[Dict[str, Any]]: 118 """Get the metadata of all definition containers matching certain criteria. 119 120 :param kwargs: A dictionary of keyword arguments containing keys and 121 values that need to match the metadata. An asterisk in the values can be 122 used to denote a wildcard. 123 :return: A list of metadata dictionaries matching the search criteria, or 124 an empty list if nothing was found. 125 """ 126 127 return self.findContainersMetadata(container_type = DefinitionContainer, **kwargs) 128 129 def findInstanceContainers(self, **kwargs: Any) -> List[InstanceContainer]: 130 """Find all InstanceContainer objects matching certain criteria. 131 132 :param kwargs: A dictionary of keyword arguments containing 133 keys and values that need to match the metadata of the 134 InstanceContainer. An asterisk in the values can be used to denote a 135 wildcard. 136 """ 137 138 return cast(List[InstanceContainer], self.findContainers(container_type = InstanceContainer, **kwargs)) 139 140 def findInstanceContainersMetadata(self, **kwargs: Any) -> List[Dict[str, Any]]: 141 """Find the metadata of all instance containers matching certain criteria. 142 143 :param kwargs: A dictionary of keyword arguments containing keys and 144 values that need to match the metadata. An asterisk in the values can be 145 used to denote a wildcard. 146 :return: A list of metadata dictionaries matching the search criteria, or 147 an empty list if nothing was found. 148 """ 149 150 return self.findContainersMetadata(container_type = InstanceContainer, **kwargs) 151 152 def findContainerStacks(self, **kwargs: Any) -> List[ContainerStack]: 153 """Find all ContainerStack objects matching certain criteria. 154 155 :param kwargs: A dictionary of keyword arguments containing 156 keys and values that need to match the metadata of the ContainerStack. 157 An asterisk in the values can be used to denote a wildcard. 158 """ 159 160 return cast(List[ContainerStack], self.findContainers(container_type = ContainerStack, **kwargs)) 161 162 def findContainerStacksMetadata(self, **kwargs: Any) -> List[Dict[str, Any]]: 163 """Find the metadata of all container stacks matching certain criteria. 164 165 :param kwargs: A dictionary of keyword arguments containing keys and 166 values that need to match the metadata. An asterisk in the values can be 167 used to denote a wildcard. 168 :return: A list of metadata dictionaries matching the search criteria, or 169 an empty list if nothing was found. 170 """ 171 172 return self.findContainersMetadata(container_type = ContainerStack, **kwargs) 173 174 @UM.FlameProfiler.profile 175 def findContainers(self, *, ignore_case: bool = False, **kwargs: Any) -> List[ContainerInterface]: 176 """Find all container objects matching certain criteria. 177 178 :param container_type: If provided, return only objects that are 179 instances or subclasses of container_type. 180 :param kwargs: A dictionary of keyword arguments containing 181 keys and values that need to match the metadata of the container. An 182 asterisk can be used to denote a wildcard. 183 184 :return: A list of containers matching the search criteria, or an empty 185 list if nothing was found. 186 """ 187 188 # Find the metadata of the containers and grab the actual containers from there. 189 results_metadata = self.findContainersMetadata(ignore_case = ignore_case, **kwargs) 190 result = [] 191 for metadata in results_metadata: 192 if metadata["id"] in self._containers: # Already loaded, so just return that. 193 result.append(self._containers[metadata["id"]]) 194 else: # Metadata is loaded, but not the actual data. 195 if metadata["id"] in self._wrong_container_ids: 196 Logger.logException("e", "Error when loading container {container_id}: This is a weird container, probably some file is missing".format(container_id = metadata["id"])) 197 continue 198 provider = self.source_provider[metadata["id"]] 199 if not provider: 200 Logger.log("w", "The metadata of container {container_id} was added during runtime, but no accompanying container was added.".format(container_id = metadata["id"])) 201 continue 202 try: 203 new_container = provider.loadContainer(metadata["id"]) 204 except ContainerFormatError as e: 205 Logger.logException("e", "Error in the format of container {container_id}: {error_msg}".format(container_id = metadata["id"], error_msg = str(e))) 206 continue 207 except Exception as e: 208 Logger.logException("e", "Error when loading container {container_id}: {error_msg}".format(container_id = metadata["id"], error_msg = str(e))) 209 continue 210 self.addContainer(new_container) 211 self.containerLoadComplete.emit(new_container.getId()) 212 result.append(new_container) 213 return result 214 215 def findContainersMetadata(self, *, ignore_case: bool = False, **kwargs: Any) -> List[Dict[str, Any]]: 216 """Find the metadata of all container objects matching certain criteria. 217 218 :param container_type: If provided, return only objects that are 219 instances or subclasses of ``container_type``. 220 :param kwargs: A dictionary of keyword arguments containing keys and 221 values that need to match the metadata. An asterisk can be used to 222 denote a wildcard. 223 :return: A list of metadata dictionaries matching the search criteria, or 224 an empty list if nothing was found. 225 """ 226 227 candidates = None 228 if "id" in kwargs and kwargs["id"] is not None and "*" not in kwargs["id"] and not ignore_case: 229 if kwargs["id"] not in self.metadata: # If we're looking for an unknown ID, try to lazy-load that one. 230 if kwargs["id"] not in self.source_provider: 231 for candidate in self._providers: 232 if kwargs["id"] in candidate.getAllIds(): 233 self.source_provider[kwargs["id"]] = candidate 234 break 235 else: 236 return [] 237 provider = self.source_provider[kwargs["id"]] 238 if not provider: 239 Logger.log("w", "Metadata of container {container_id} is missing even though the container is added during run-time.") 240 return [] 241 metadata = provider.loadMetadata(kwargs["id"]) 242 if metadata is None or metadata.get("id", "") in self._wrong_container_ids or "id" not in metadata: 243 return [] 244 self.metadata[metadata["id"]] = metadata 245 self.source_provider[metadata["id"]] = provider 246 247 # Since IDs are the primary key and unique we can now simply request the candidate and check if it matches all requirements. 248 if kwargs["id"] not in self.metadata: 249 return [] # Still no result, so return an empty list. 250 if len(kwargs) == 1: 251 return [self.metadata[kwargs["id"]]] 252 candidates = [self.metadata[kwargs["id"]]] 253 del kwargs["id"] # No need to check for the ID again. 254 255 query = ContainerQuery.ContainerQuery(self, ignore_case = ignore_case, **kwargs) 256 query.execute(candidates = candidates) 257 258 return cast(List[Dict[str, Any]], query.getResult()) # As the execute of the query is done, result won't be none. 259 260 def findDirtyContainers(self, *, ignore_case: bool = False, **kwargs: Any) -> List[ContainerInterface]: 261 """Specialized find function to find only the modified container objects 262 that also match certain criteria. 263 264 This is faster than the normal find methods since it won't ever load all 265 containers, but only the modified ones. Since containers must be fully 266 loaded before they are modified, you are guaranteed that any operations 267 on the resulting containers will not trigger additional containers to 268 load lazily. 269 270 :param kwargs: A dictionary of keyword arguments containing 271 keys and values that need to match the metadata of the container. An 272 asterisk can be used to denote a wildcard. 273 :param ignore_case: Whether casing should be ignored when matching string 274 values of metadata. 275 :return: A list of containers matching the search criteria, or an empty 276 list if nothing was found. 277 """ 278 279 # Find the metadata of the containers and grab the actual containers from there. 280 # 281 # We could apply the "is in self._containers" filter and the "isDirty" filter 282 # to this metadata find function as well to filter earlier, but since the 283 # filters in findContainersMetadata are applied in arbitrary order anyway 284 # this will have very little effect except to prevent a list copy. 285 results_metadata = self.findContainersMetadata(ignore_case = ignore_case, **kwargs) 286 287 result = [] 288 for metadata in results_metadata: 289 if metadata["id"] not in self._containers: # Not yet loaded, so it can't be dirty. 290 continue 291 candidate = self._containers[metadata["id"]] 292 if candidate.isDirty(): 293 result.append(self._containers[metadata["id"]]) 294 return result 295 296 def getEmptyInstanceContainer(self) -> InstanceContainer: 297 """This is a small convenience to make it easier to support complex structures in ContainerStacks.""" 298 299 return self._emptyInstanceContainer 300 301 def setExplicitReadOnly(self, container_id: str) -> None: 302 self._explicit_read_only_container_ids.add(container_id) 303 304 def isExplicitReadOnly(self, container_id: str) -> bool: 305 return container_id in self._explicit_read_only_container_ids 306 307 def isReadOnly(self, container_id: str) -> bool: 308 """Returns whether a profile is read-only or not. 309 310 Whether it is read-only depends on the source where the container is 311 obtained from. 312 :return: True if the container is read-only, or False if it can be 313 modified. 314 """ 315 316 if self.isExplicitReadOnly(container_id): 317 return True 318 provider = self.source_provider.get(container_id) 319 if not provider: 320 return False # If no provider had the container, that means that the container was only in memory. Then it's always modifiable. 321 return provider.isReadOnly(container_id) 322 323 # Gets the container file path with for the container with the given ID. Returns None if the container/file doesn't 324 # exist. 325 def getContainerFilePathById(self, container_id: str) -> Optional[str]: 326 provider = self.source_provider.get(container_id) 327 if not provider: 328 return None 329 return provider.getContainerFilePathById(container_id) 330 331 def isLoaded(self, container_id: str) -> bool: 332 """Returns whether a container is completely loaded or not. 333 334 If only its metadata is known, it is not yet completely loaded. 335 :return: True if all data about this container is known, False if only 336 metadata is known or the container is completely unknown. 337 """ 338 339 return container_id in self._containers 340 341 def loadAllMetadata(self) -> None: 342 """Load the metadata of all available definition containers, instance 343 containers and container stacks. 344 """ 345 346 self._clearQueryCache() 347 gc.disable() 348 resource_start_time = time.time() 349 for provider in self._providers: # Automatically sorted by the priority queue. 350 for container_id in list(provider.getAllIds()): # Make copy of all IDs since it might change during iteration. 351 if container_id not in self.metadata: 352 self._application.processEvents() # Update the user interface because loading takes a while. Specifically the loading screen. 353 metadata = provider.loadMetadata(container_id) 354 if not self._isMetadataValid(metadata): 355 Logger.log("w", "Invalid metadata for container {container_id}: {metadata}".format(container_id = container_id, metadata = metadata)) 356 if container_id in self.metadata: 357 del self.metadata[container_id] 358 continue 359 self.metadata[container_id] = metadata 360 self.source_provider[container_id] = provider 361 Logger.log("d", "Loading metadata into container registry took %s seconds", time.time() - resource_start_time) 362 gc.enable() 363 ContainerRegistry.allMetadataLoaded.emit() 364 365 @UM.FlameProfiler.profile 366 def load(self) -> None: 367 """Load all available definition containers, instance containers and 368 container stacks. 369 370 :note This method does not clear the internal list of containers. This means that any containers 371 that were already added when the first call to this method happened will not be re-added. 372 """ 373 374 # Disable garbage collection to speed up the loading (at the cost of memory usage). 375 gc.disable() 376 resource_start_time = time.time() 377 378 with self.lockCache(): # Because we might be writing cache files. 379 for provider in self._providers: 380 for container_id in list(provider.getAllIds()): # Make copy of all IDs since it might change during iteration. 381 if container_id not in self._containers: 382 # Update UI while loading. 383 self._application.processEvents() # Update the user interface because loading takes a while. Specifically the loading screen. 384 try: 385 self._containers[container_id] = provider.loadContainer(container_id) 386 except: 387 Logger.logException("e", "Failed to load container %s", container_id) 388 raise 389 self.metadata[container_id] = self._containers[container_id].getMetaData() 390 self.source_provider[container_id] = provider 391 self.containerLoadComplete.emit(container_id) 392 393 gc.enable() 394 Logger.log("d", "Loading data into container registry took %s seconds", time.time() - resource_start_time) 395 396 @UM.FlameProfiler.profile 397 def addContainer(self, container: ContainerInterface) -> bool: 398 container_id = container.getId() 399 if container_id in self._containers: 400 return True # Container was already there, consider that a success 401 402 if hasattr(container, "metaDataChanged"): 403 container.metaDataChanged.connect(self._onContainerMetaDataChanged) 404 405 self.metadata[container_id] = container.getMetaData() 406 self._containers[container_id] = container 407 if container_id not in self.source_provider: 408 self.source_provider[container_id] = None # Added during runtime. 409 self._clearQueryCacheByContainer(container) 410 411 # containerAdded is a custom signal and can trigger direct calls to its subscribers. This should be avoided 412 # because with the direct calls, the subscribers need to know everything about what it tries to do to avoid 413 # triggering this signal again, which eventually can end up exceeding the max recursion limit. 414 # We avoid the direct calls here to make sure that the subscribers do not need to take into account any max 415 # recursion problem. 416 self._application.callLater(self.containerAdded.emit, container) 417 return True 418 419 @UM.FlameProfiler.profile 420 def removeContainer(self, container_id: str) -> None: 421 # Here we only need to check metadata because a container may not be loaded but its metadata must have been 422 # loaded first. 423 if container_id not in self.metadata: 424 Logger.log("w", "Tried to delete container {container_id}, which doesn't exist or isn't loaded.".format(container_id = container_id)) 425 return # Ignore. 426 427 # CURA-6237 428 # Do not try to operate on invalid containers because removeContainer() needs to load it if it's not loaded yet 429 # (see below), but an invalid container cannot be loaded. 430 if container_id in self._wrong_container_ids: 431 Logger.log("w", "Container [%s] is faulty, it won't be able to be loaded, so no need to remove, skip.") 432 # delete the metadata if present 433 if container_id in self.metadata: 434 del self.metadata[container_id] 435 return 436 437 container = None 438 if container_id in self._containers: 439 container = self._containers[container_id] 440 if hasattr(container, "metaDataChanged"): 441 container.metaDataChanged.disconnect(self._onContainerMetaDataChanged) 442 del self._containers[container_id] 443 if container_id in self.metadata: 444 if container is None: 445 # We're in a bit of a weird state now. We want to notify the rest of the code that the container 446 # has been deleted, but due to lazy loading, it hasn't been loaded yet. The issues is that in order 447 # to notify the rest of the code, we need to actually *have* the container. So an empty instance 448 # container is created, which is emitted with the containerRemoved signal and contains the metadata 449 container = EmptyInstanceContainer(container_id) 450 container.metaData = self.metadata[container_id] 451 del self.metadata[container_id] 452 if container_id in self.source_provider: 453 if self.source_provider[container_id] is not None: 454 cast(ContainerProvider, self.source_provider[container_id]).removeContainer(container_id) 455 del self.source_provider[container_id] 456 457 if container is not None: 458 self._clearQueryCacheByContainer(container) 459 self.containerRemoved.emit(container) 460 461 Logger.log("d", "Removed container %s", container_id) 462 463 @UM.FlameProfiler.profile 464 def renameContainer(self, container_id: str, new_name: str, new_id: Optional[str] = None) -> None: 465 Logger.log("d", "Renaming container %s to %s", container_id, new_name) 466 # Same as removeContainer(), metadata is always loaded but containers may not, so always check metadata. 467 if container_id not in self.metadata: 468 Logger.log("w", "Unable to rename container %s, because it does not exist", container_id) 469 return 470 471 container = self._containers.get(container_id) 472 if container is None: 473 container = self.findContainers(id = container_id)[0] 474 container = cast(ContainerInterface, container) 475 476 if new_name == container.getName(): 477 Logger.log("w", "Unable to rename container %s, because the name (%s) didn't change", container_id, new_name) 478 return 479 480 self.containerRemoved.emit(container) 481 482 try: 483 container.setName(new_name) #type: ignore 484 except TypeError: #Some containers don't allow setting the name. 485 return 486 if new_id is not None: 487 source_provider = self.source_provider[container.getId()] 488 del self._containers[container.getId()] 489 del self.metadata[container.getId()] 490 del self.source_provider[container.getId()] 491 if source_provider is not None: 492 source_provider.removeContainer(container.getId()) 493 container.getMetaData()["id"] = new_id 494 self._containers[container.getId()] = container 495 self.metadata[container.getId()] = container.getMetaData() 496 self.source_provider[container.getId()] = None # to be saved with saveSettings 497 498 self._clearQueryCacheByContainer(container) 499 self.containerAdded.emit(container) 500 501 @UM.FlameProfiler.profile 502 def uniqueName(self, original: str) -> str: 503 """Creates a new unique name for a container that doesn't exist yet. 504 505 It tries if the original name you provide exists, and if it doesn't 506 it'll add a " 1" or " 2" after the name to make it unique. 507 508 :param original: The original name that may not be unique. 509 :return: A unique name that looks a lot like the original but may have 510 a number behind it to make it unique. 511 """ 512 513 original = original.replace("*", "") # Filter out wildcards, since this confuses the ContainerQuery. 514 name = original.strip() 515 516 num_check = re.compile(r"(.*?)\s*#\d+$").match(name) 517 if num_check: #There is a number in the name. 518 name = num_check.group(1) #Filter out the number. 519 520 if not name: #Wait, that deleted everything! 521 name = "Profile" 522 elif not self.findContainersMetadata(id = original.strip(), ignore_case = True) and not self.findContainersMetadata(name = original.strip()): 523 # Check if the stripped version of the name is unique (note that this can still have the number in it) 524 return original.strip() 525 526 unique_name = name 527 i = 1 528 while self.findContainersMetadata(id = unique_name, ignore_case = True) or self.findContainersMetadata(name = unique_name): #A container already has this name. 529 i += 1 #Try next numbering. 530 unique_name = "%s #%d" % (name, i) #Fill name like this: "Extruder #2". 531 return unique_name 532 533 @classmethod 534 def addContainerType(cls, container: "PluginObject") -> None: 535 """Add a container type that will be used to serialize/deserialize containers. 536 537 :param container: An instance of the container type to add. 538 """ 539 540 plugin_id = container.getPluginId() 541 metadata = PluginRegistry.getInstance().getMetaData(plugin_id) 542 if "settings_container" not in metadata or "mimetype" not in metadata["settings_container"]: 543 raise Exception("Plugin {plugin} has incorrect metadata: Expected a 'settings_container' block with a 'mimetype' entry".format(plugin = plugin_id)) 544 cls.addContainerTypeByName(container.__class__, plugin_id, metadata["settings_container"]["mimetype"]) 545 546 @classmethod 547 def addContainerTypeByName(cls, container_type: type, type_name: str, mime_type: str) -> None: 548 """Used to associate mime types with object to be created 549 :param container_type: ContainerStack or derivative 550 :param type_name: 551 :param mime_type: 552 """ 553 554 cls.__container_types[type_name] = container_type 555 cls.mime_type_map[mime_type] = container_type 556 557 @classmethod 558 def getMimeTypeForContainer(cls, container_type: type) -> Optional[MimeType]: 559 """Retrieve the mime type corresponding to a certain container type 560 561 :param container_type: The type of container to get the mime type for. 562 563 :return: A MimeType object that matches the mime type of the container or None if not found. 564 """ 565 566 try: 567 mime_type_name = UM.Dictionary.findKey(cls.mime_type_map, container_type) 568 if mime_type_name: 569 return MimeTypeDatabase.getMimeType(mime_type_name) 570 except ValueError: 571 Logger.log("w", "Unable to find mimetype for container %s", container_type) 572 return None 573 574 @classmethod 575 def getContainerForMimeType(cls, mime_type): 576 """Get the container type corresponding to a certain mime type. 577 578 :param mime_type: The mime type to get the container type for. 579 580 :return: A class object of a container type that corresponds to the specified mime type or None if not found. 581 """ 582 583 return cls.mime_type_map.get(mime_type.name, None) 584 585 @classmethod 586 def getContainerTypes(cls): 587 """Get all the registered container types 588 589 :return: A dictionary view object that provides access to the container types. 590 The key is the plugin ID, the value the container type. 591 """ 592 593 return cls.__container_types.items() 594 595 def saveContainer(self, container: "ContainerInterface", provider: Optional["ContainerProvider"] = None) -> None: 596 """Save single dirty container""" 597 598 if not hasattr(provider, "saveContainer"): 599 provider = self.getDefaultSaveProvider() 600 if not container.isDirty(): 601 return 602 603 provider.saveContainer(container) #type: ignore 604 container.setDirty(False) 605 self.source_provider[container.getId()] = provider 606 607 def saveDirtyContainers(self) -> None: 608 """Save all the dirty containers by calling the appropriate container providers""" 609 610 # Lock file for "more" atomically loading and saving to/from config dir. 611 with self.lockFile(): 612 for instance in self.findDirtyContainers(container_type = InstanceContainer): 613 self.saveContainer(instance) 614 615 for stack in self.findContainerStacks(): 616 self.saveContainer(stack) 617 618 # Clear the internal query cache 619 def _clearQueryCache(self, *args: Any, **kwargs: Any) -> None: 620 ContainerQuery.ContainerQuery.cache.clear() 621 622 def _clearQueryCacheByContainer(self, container: ContainerInterface) -> None: 623 """Clear the query cache by using container type. 624 This is a slightly smarter way of clearing the cache. Only queries that are of the same type (or without one) 625 are cleared. 626 """ 627 628 # Remove all case-insensitive matches since we won't find those with the below "<=" subset check. 629 # TODO: Properly check case-insensitively in the dict's values. 630 for key in list(ContainerQuery.ContainerQuery.cache): 631 if not key[0]: 632 del ContainerQuery.ContainerQuery.cache[key] 633 634 # Remove all cache items that this container could fall in. 635 for key in list(ContainerQuery.ContainerQuery.cache): 636 query_metadata = dict(zip(key[1::2], key[2::2])) 637 if query_metadata.items() <= container.getMetaData().items(): 638 del ContainerQuery.ContainerQuery.cache[key] 639 640 def _onContainerMetaDataChanged(self, *args: ContainerInterface, **kwargs: Any) -> None: 641 """Called when any container's metadata changed. 642 643 This function passes it on to the containerMetaDataChanged signal. Sadly 644 that doesn't work automatically between pyqtSignal and UM.Signal. 645 """ 646 647 container = args[0] 648 # Always emit containerMetaDataChanged, even if the dictionary didn't actually change: The contents of the dictionary might have changed in-place! 649 self.metadata[container.getId()] = container.getMetaData() # refresh the metadata 650 self.containerMetaDataChanged.emit(*args, **kwargs) 651 652 def _isMetadataValid(self, metadata: Optional[Dict[str, Any]]) -> bool: 653 """Validate a metadata object. 654 655 If the metadata is invalid, the container is not allowed to be in the 656 registry. 657 :param metadata: A metadata object. 658 :return: Whether this metadata was valid. 659 """ 660 661 return metadata is not None 662 663 def getLockFilename(self) -> str: 664 """Get the lock filename including full path 665 Dependent on when you call this function, Resources.getConfigStoragePath may return different paths 666 """ 667 668 return Resources.getStoragePath(Resources.Resources, self._application.getApplicationLockFilename()) 669 670 def getCacheLockFilename(self) -> str: 671 """Get the cache lock filename including full path.""" 672 673 return Resources.getStoragePath(Resources.Cache, self._application.getApplicationLockFilename()) 674 675 def lockFile(self) -> LockFile: 676 """Contextmanager to create a lock file and remove it afterwards.""" 677 678 return LockFile( 679 self.getLockFilename(), 680 timeout = 10, 681 wait_msg = "Waiting for lock file in local config dir to disappear..." 682 ) 683 684 def lockCache(self) -> LockFile: 685 """Context manager to create a lock file for the cache directory and remove 686 it afterwards. 687 """ 688 689 return LockFile( 690 self.getCacheLockFilename(), 691 timeout = 10, 692 wait_msg = "Waiting for lock file in cache directory to disappear." 693 ) 694 695 __container_types = { 696 "definition": DefinitionContainer, 697 "instance": InstanceContainer, 698 "stack": ContainerStack, 699 } 700 701 mime_type_map = { 702 "application/x-uranium-definitioncontainer": DefinitionContainer, 703 "application/x-uranium-instancecontainer": InstanceContainer, 704 "application/x-uranium-containerstack": ContainerStack, 705 "application/x-uranium-extruderstack": ContainerStack 706 } # type: Dict[str, Type[ContainerInterface]] 707 708 __instance = None # type: ContainerRegistry 709 710 @classmethod 711 def getInstance(cls, *args, **kwargs) -> "ContainerRegistry": 712 return cls.__instance 713 714 715PluginRegistry.addType("settings_container", ContainerRegistry.addContainerType) 716