Coverage for apio/managers/package_manager.py: 75%
293 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 03:53 +0000
1# -*- coding: utf-8 -*-
2# -- This file is part of the Apio project
3# -- (C) 2016-2021 FPGAwars
4# -- Author Jesús Arroyo
5# -- License GPLv2
6"""Package install/uninstall functionality.
7Used by the 'apio packages' command.
8"""
10import os
11import json
12from enum import Enum, unique
13from datetime import datetime
14from dataclasses import dataclass
15from typing import Any
16from pathlib import Path
17import shutil
18from apio.common.apio_console import cout, cstyle, fatal_error
19from apio.common.apio_styles import SUCCESS, EMPH3
20from apio.common.debug_util import is_debug
21from apio.managers.downloader import FileDownloader
22from apio.utils import util
23from apio.utils.apio_platforms import ApioPlatform
24from apio.managers.remote_config import RemoteConfig, PackageRemoteConfig
27@unique
28class RequiredPackageStatus(Enum):
29 """Represents the classification of a required package status."""
31 # -- NOTE: The string values here are using facing by the
32 # -- 'apio packages list' command.
34 PACKAGE_UNINSTALLED = "Uninstalled"
35 PACKAGE_DIR_MISSING = "Package dir missing"
36 PACKAGE_DIR_IS_A_FILE = "Package dir is a file"
37 PACKAGE_VERSION_MISMATCH = "Version mismatch"
38 PACKAGE_PLATFORM_MISMATCH = "Platform mismatch"
39 PACKAGE_APIO_VERSION_MISMATCH = "Apio version mismatch"
40 PACKAGE_URL_MISMATCH = "Source URL mismatch"
41 PACKAGE_OK = "OK"
43 @property
44 def is_ok(self) -> bool:
45 """Returns True if the status is of a legit package."""
46 return self == self.PACKAGE_OK
48 @property
49 def is_inconsistency(self) -> bool:
50 """Is it an inconsistency that requires fixing before installing the
51 uninstalled packages."""
52 return self not in (self.PACKAGE_UNINSTALLED, self.PACKAGE_OK)
55@unique
56class OrphanType(Enum):
57 """Represents the types of orphans (leftovers items)."""
59 # -- NOTE: The string values here are using facing by the
60 # -- 'apio packages list' command.
62 # -- Non required package in installed packages index, potentially
63 # -- it also has an entry with same name in the packages folder..
64 ORPHAN_PACKAGE = "Orphan package"
65 # -- Package dir that doesn't match a required or an orphan package.
66 ORPHAN_FILE = "Orphan file"
67 # -- A file in the packages dir that doesn't match a name of an orphan
68 # -- package.
69 ORPHAN_DIR = "Orphan dir"
72@dataclass
73class PackagesScanResults:
74 """Represents results of packages scan."""
76 # -- Names and statuses of required packages.
77 required_packages: dict[str, RequiredPackageStatus]
79 # -- Name and types of package, dir, and file orphans.
80 orphans: dict[str, OrphanType]
82 def packages_installed_ok(self) -> bool:
83 """Returns true if all the required packages are installed ok,
84 regardless of other fixable errors."""
85 return all(status.is_ok for status in self.required_packages.values())
87 def num_inconsistencies_to_fix(self) -> int:
88 """Returns the number of inconsistencies that require fixing before
89 installing any missing package."""
90 required_packages_errors = sum(
91 1
92 for status in self.required_packages.values()
93 if not status.is_inconsistency
94 )
95 orphans_errors = len(self.orphans)
96 return required_packages_errors + orphans_errors
98 def is_all_ok(self) -> bool:
99 """Return True if all packages are installed properly with no
100 issues."""
101 return self.packages_installed_ok() and len(self.orphans) == 0
103 def dump(self):
104 """Dump the content of this object. For debugging."""
105 cout()
106 cout("Package scan results:")
107 cout(f" required {self.required_packages}")
108 cout(f" orphans {self.orphans}")
111def get_datetime_stamp(dt: datetime | None = None) -> str:
112 """Returns a string with time now as yyyy-mm-dd-hh-mm"""
113 if dt is None: 113 ↛ 115line 113 didn't jump to line 115 because the condition on line 113 was always true
114 dt = datetime.now()
115 return dt.strftime("%Y-%m-%d-%H-%M")
118class PackageManager:
119 """Context for package managements operations.
120 This class provides the information needed for package management
121 operations. This is a subset of the information contained by ApioContext
122 and we use it, instead of passing the ApioContext, because we need to
123 perform package management operations (e.g. updating packages) before
124 the ApioContext object is fully initialized.
125 """
127 def __init__(
128 self,
129 remote_config: RemoteConfig,
130 required_packages: dict,
131 platform: ApioPlatform,
132 apio_home_dir: Path,
133 packages_dir: Path,
134 ):
136 # pylint: disable=too-many-arguments
137 # pylint: disable=too-many-positional-arguments
139 # -- Same as ApioContext.remote_config
140 self.remote_config = remote_config
141 # -- Same as ApioContext.required_packages
142 self.required_packages = required_packages
143 # -- platform: ApioPlatform
144 self.platform = platform
145 # -- Same as ApioContext.apio_home_dir
146 self.apio_home_dir = apio_home_dir
147 # -- Same as ApioContext.packages_dir
148 self.packages_dir = packages_dir
150 # -- Sanity checks.
151 assert isinstance(self.remote_config, RemoteConfig)
152 assert self.required_packages
153 assert self.platform
154 assert self.packages_dir
156 # -- Initialized installed packages, a copy of
157 # -- installed-packages.json.
158 self.installed_packages: dict[str, Any] = {}
160 # -- Cache the packages index file path
161 # -- Ex. '/home/obijuan/.apio/packages/installed_packages.json'
162 self._packages_index_path = packages_dir / "installed_packages.json"
164 # -- Read the installed packages file, if exists.
165 self._maybe_load_installed_packages_file()
167 def required_package_dir(self, package_name: str) -> Path:
168 """Return the local root directory of the package with given name"""
169 # -- Validate the package name
170 assert package_name in self.required_packages, package_name
171 # -- Construct the path
172 return self.packages_dir / package_name
174 def _construct_package_download_url(
175 self,
176 package_remote_config: PackageRemoteConfig,
177 ) -> str:
178 """Construct the download URL for the given package name and
179 version."""
181 # -- Create vars mapping.
182 url_vars = {
183 "${PLATFORM}": self.platform.id,
184 "${YYYYMMDD}": package_remote_config.release_tag.replace("-", ""),
185 }
186 if is_debug(1): 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 cout(f"Package URL vars: {url_vars}")
189 # -- Define the url parts.
190 url_parts = [
191 "https://github.com/",
192 package_remote_config.repo_organization,
193 "/",
194 package_remote_config.repo_name,
195 "/releases/download/",
196 package_remote_config.release_tag,
197 "/",
198 package_remote_config.release_file,
199 ]
201 if is_debug(1): 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 cout(f"package url parts = {url_parts}")
204 # -- Concatenate the URL parts.
205 url = "".join(url_parts)
207 if is_debug(1): 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 cout(f"Combined package url: {url}")
210 # -- Replace placeholders with values.
211 for name, val in url_vars.items():
212 url = url.replace(name, val)
214 if is_debug(1): 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true
215 cout(f"Resolved package url: {url}")
217 # -- All done.
218 return url
220 def _download_package_file(
221 self, url: str, dir_path: Path, package_name: str
222 ) -> Path:
223 """Download the given file (url). Return the path of local destination
224 file. Exits with a user message and error code if any error.
226 * INPUTS:
227 * url: File to download
228 * OUTPUTS:
229 * The path of the destination file
230 """
232 filepath: Path | None = None
234 try:
235 # -- Object for downloading the file
236 downloader = FileDownloader(url, dir_path)
238 # -- Get the destination path
239 filepath = downloader.destination
241 downloader.download()
243 # -- If the user press Ctrl-C (Abort)
244 except KeyboardInterrupt:
246 # -- Remove the file
247 if filepath and filepath.is_file():
248 filepath.unlink()
250 # -- Inform the user
251 fatal_error("User aborted download")
253 except IOError as exc:
254 fatal_error(
255 f"Failed to download package '{package_name}'", cause=exc
256 )
258 # -- Return the destination path
259 return filepath
261 def _delete_package_dir(self, package_name: str, verbose: bool) -> None:
262 """Delete the directory of the package with given name."""
263 package_path = self.packages_dir / package_name
265 # -- If doesn't exist, ignore silently.
266 if not package_path.exists():
267 return
269 if verbose: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 cout(f"Deleting {str(package_path)}")
272 if package_path.is_dir(): 272 ↛ 277line 272 didn't jump to line 277 because the condition on line 272 was always true
273 # -- Sanity check the path and delete.
274 assert "packages" in str(package_path).lower(), package_path
275 shutil.rmtree(package_path)
276 else:
277 package_path.unlink()
279 # -- Confirm
280 if package_path.exists(): 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true
281 fatal_error(
282 f"Package dir deletion failed: {str(package_path.absolute())}"
283 )
285 def scan_and_fix_inconsistencies(self) -> bool:
286 """Scan the packages and fix if there are errors. Returns true
287 if the packages are installed ok."""
289 # -- Scan the packages.
290 scan: PackagesScanResults = self.scan_packages()
292 # -- If there are fixable errors, fix them.
293 if scan.num_inconsistencies_to_fix() > 0: 293 ↛ 299line 293 didn't jump to line 299 because the condition on line 293 was always true
294 self._fix_inconsistencies(scan)
296 # -- Return a flag that indicates if all packages are installed ok. We
297 # -- use a scan from before the fixing but the fixing does not touch
298 # -- installed ok packages.
299 return scan.packages_installed_ok()
301 def install_missing_packages_on_the_fly(self, verbose=False) -> None:
302 """Install on the fly any missing packages. Does not print a thing if
303 all packages are already ok. This function is intended for on demand
304 package fetching by commands such as apio build, and thus is allowed
305 to use fetched remote config instead of fetching a fresh one. Exists
306 with error code if any error."""
308 # -- Scan and fix broken package.
309 # -- Since this is a on-the-fly operation, we don't require a fresh
310 # -- remote config file for required packages versions.
311 installed_ok = self.scan_and_fix_inconsistencies()
313 # -- If the packages are installed we are done, we are done.
314 if installed_ok: 314 ↛ 325line 314 didn't jump to line 325 because the condition on line 314 was always true
315 # -- Final sanity check of the packages.
316 self.check_packages_post_install()
317 # -- Installed ok.
318 return
320 # -- Here when we need to install some packages. Since we just fixed
321 # -- we can't have broken or packages with version mismatch, just
322 # -- installed ok, and not installed.
323 # --
324 # -- Get lists of installed and required packages.
325 installed_packages = self.installed_packages
326 required_packages_names = self.required_packages.keys()
328 # -- Install any required package that is not installed.
329 for package_name in required_packages_names:
330 if package_name not in installed_packages:
331 self.install_package(
332 package_name=package_name,
333 force_reinstall=False,
334 verbose=verbose,
335 )
337 # -- Here all packages should be ok but we check again just in case.
338 scan_results = self.scan_packages()
339 if not scan_results.is_all_ok():
340 fatal_error(
341 "Packages issues detected. Use "
342 + "'apio packages list' to investigate."
343 )
345 # -- Final sanity check of the packages.
346 self.check_packages_post_install()
348 def install_package(
349 self,
350 *,
351 package_name: str,
352 force_reinstall: bool,
353 verbose: bool,
354 ) -> None:
355 """Install a given package.
357 'package_name' is the package name, e.g. 'examples' or 'oss-cad-suite'.
358 'force' indicates if to perform the installation even if a matching
359 package is already installed.
360 'explicit' indicates that the user specified the package name(s)
361 explicitly and thus expect more feedback in case of a 'no change'
362 'verbose' indicates if to print extra information.
364 Returns normally if no error, exits the program with an error status
365 and a user message if an error is detected.
366 """
368 # -- Force verbose if debug.
369 if is_debug(1): 369 ↛ 370line 369 didn't jump to line 370 because the condition on line 369 was never true
370 verbose = True
372 # -- Caller is responsible to check check that package name is valid
373 # -- on this platform.
374 assert package_name in self.required_packages, package_name
376 # -- Set up installation announcement
377 pending_announcement: str | None = cstyle(
378 f"Installing apio package '{package_name}'", style=EMPH3
379 )
381 # -- If in chatty mode, announce now and clear. Otherwise we will
382 # -- announce later only if actually installing.
383 if verbose and pending_announcement: 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true
384 cout(pending_announcement)
385 pending_announcement = None
387 # -- Get package remote config from the cache. Caller can refresh the
388 # -- cache with the latest remote config if desired.
389 package_config: PackageRemoteConfig = (
390 self.remote_config.get_package_config(package_name)
391 )
393 # -- Get the version we should have.
394 target_version = package_config.release_version
396 # -- If not forcing and the target version already installed then
397 # -- nothing to do and we leave quietly.
398 if not force_reinstall:
399 # -- Get the package status
400 package_status = self.classify_required_package_status(
401 package_name
402 )
404 if verbose: 404 ↛ 405line 404 didn't jump to line 405 because the condition on line 404 was never true
405 cout(
406 f"Package {package_name} status is"
407 + f" '{package_status.value}'"
408 )
410 # -- If the package is OK then nothing to do.
411 if package_status.is_ok:
412 if verbose: 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 cout(
414 f"Package {package_name} version {target_version} "
415 + "is already installed OK",
416 style=SUCCESS,
417 )
418 return
420 # -- Here we need to fetch and install so can be more chatty.
422 # -- Here we actually do the work. Announce if we haven't done it yet.
423 if pending_announcement: 423 ↛ 427line 423 didn't jump to line 427 because the condition on line 423 was always true
424 cout(pending_announcement)
425 pending_announcement = None
427 cout(f"Fetching version {target_version} ({self.platform.id})")
429 # -- Construct the download URL.
430 download_url = self._construct_package_download_url(package_config)
431 if verbose: 431 ↛ 432line 431 didn't jump to line 432 because the condition on line 431 was never true
432 cout(f"Download URL: {download_url}")
434 # -- Prepare the packages directory.
435 self.packages_dir.mkdir(exist_ok=True)
437 # -- Prepare the package directory.
438 package_dir = self.packages_dir / package_name
439 cout(f"Package dir: {package_dir}")
441 # -- Download the package file from the remote server.
442 local_package_file = self._download_package_file(
443 download_url, self.packages_dir, package_name
444 )
445 if verbose: 445 ↛ 446line 445 didn't jump to line 446 because the condition on line 445 was never true
446 cout(f"Local package file: {local_package_file}")
448 # -- Delete the old package dir, if exists, to avoid name conflicts and
449 # -- left over files.
450 self._delete_package_dir(package_name, verbose)
452 # -- Unpack the package. This creates a new package dir.
453 util.unpack_tgz(local_package_file, package_dir)
455 # -- Remove the package file. We don't need it anymore.
456 if verbose: 456 ↛ 457line 456 didn't jump to line 457 because the condition on line 456 was never true
457 cout(f"Deleting package file {local_package_file}")
458 local_package_file.unlink()
460 # -- Add package and save.
461 self.add_package(
462 package_name, target_version, self.platform.id, download_url
463 )
465 # -- Inform the user!
466 cout(f"Package '{package_name}' installed successfully", style=SUCCESS)
468 def _fix_inconsistencies(self, scan: PackagesScanResults) -> None:
469 """If the package scan result contains errors, fix them. This
470 does not install missing packages, just fixing inconsistencies."""
472 for package_name, package_status in scan.required_packages.items():
473 if package_status.is_inconsistency:
474 cout(f"Uninstalling broken package '{package_name}'")
475 self._delete_package_dir(package_name, verbose=False)
476 self.remove_package(package_name)
478 for orphan_name, orphan_type in scan.orphans.items():
479 # -- Delete an unknown entry in the installed packages index.
480 if orphan_type == orphan_type.ORPHAN_PACKAGE: 480 ↛ 481line 480 didn't jump to line 481 because the condition on line 480 was never true
481 cout(f"Uninstalling unknown package '{orphan_name}'")
482 self.remove_package(orphan_name)
484 # -- Delete an unknown dir in the package dir.
485 elif orphan_type == orphan_type.ORPHAN_DIR: 485 ↛ 492line 485 didn't jump to line 492 because the condition on line 485 was always true
486 cout(f"Deleting unknown package dir '{orphan_name}'")
487 dir_path = self.packages_dir / orphan_name
488 assert "packages" in str(dir_path).lower(), dir_path
489 shutil.rmtree(dir_path)
491 # -- Delete an unknown file in the packages dir.
492 elif orphan_type == orphan_type.ORPHAN_FILE:
493 cout(f"Deleting unknown package file '{orphan_name}'")
494 file_path = self.packages_dir / orphan_name
495 assert "packages" in str(file_path).lower(), dir_path
496 file_path.unlink()
498 # -- Unexpected orphan type.
499 else:
500 raise ValueError(f"Unknown orphan type: {orphan_type}")
502 def read_package_build_info(self, package_name: str) -> dict[str, Any]:
503 """Returns the BUILD-INFO.json of the package as a dict. Fatal
504 error if doesn't exist or can't parse."""
506 build_info_path = (
507 self.required_package_dir(package_name) / "BUILD-INFO.json"
508 )
510 # pylint: disable=broad-exception-caught
512 try:
513 with open(build_info_path, encoding="utf-8") as f:
514 build_info = json.load(f)
515 except Exception as e:
516 fatal_error(
517 f"Reading/parsing [{build_info_path}] failed.", cause=e
518 )
520 return build_info
522 def get_yosys_release_tag(self) -> str:
523 """Return the version tag (e.g. "2026-03-21") of the underlying Yosys.
524 This value is extract from the BUILD-INFO.json file of the apio
525 oss-cad-suite package."""
526 build_info = self.read_package_build_info("oss-cad-suite")
527 return build_info["yosys-release-tag"]
529 def check_packages_post_install(self):
530 """Called after the Apio packages were installed or fixed and are
531 believed to be correct. Performs additional validation of the apio
532 packages and exits with an error on any error."""
534 # -- Read the build info of the two packages.
535 build_info1 = self.read_package_build_info("oss-cad-suite")
536 build_info2 = self.read_package_build_info("openxc7")
538 # -- Extract the version of the underlying yosys
539 yosys_release_tag1 = build_info1["yosys-release-tag"]
540 yosys_release_tag2 = build_info2["yosys-release-tag"]
542 # -- Compare the versions.
543 if yosys_release_tag1 != yosys_release_tag2:
544 fatal_error(
545 'The packages "oss-cad-suite" and "openxc7" were built with '
546 + 'different "yosys-release-tag".',
547 "Their respective BUILD-INFO.json files "
548 + f'contain "{yosys_release_tag1}" vs "{yosys_release_tag2}"',
549 info="This typically happens due to corrupt packages or "
550 + "bad remote configuration by the Apio team.",
551 )
553 def classify_required_package_status(
554 self, name: str
555 ) -> RequiredPackageStatus:
556 """Classify existing or missing entry under the package directory."""
558 # pylint: disable=too-many-return-statements
560 # -- Check tha the package is a required one.
561 assert name in self.required_packages, name
563 # -- Construct the package path.
564 package_path: Path = self.required_package_dir(name)
566 if name not in self.installed_packages:
567 return RequiredPackageStatus.PACKAGE_UNINSTALLED
569 if not package_path.exists():
570 return RequiredPackageStatus.PACKAGE_DIR_MISSING
572 if not package_path.is_dir(): 572 ↛ 573line 572 didn't jump to line 573 because the condition on line 572 was never true
573 return RequiredPackageStatus.PACKAGE_DIR_IS_A_FILE
575 # -- Get installed package info or "" if not installed.
576 (
577 installed_version,
578 installed_platform_id,
579 installed_platform_version,
580 installed_src_url,
581 ) = self.get_installed_package_info(name)
583 # -- Get the package's remote config
584 package_config: PackageRemoteConfig = (
585 self.remote_config.get_package_config(name)
586 )
588 if ( 588 ↛ 592line 588 didn't jump to line 592 because the condition on line 588 was never true
589 not installed_version
590 or installed_version != package_config.release_version
591 ):
592 return RequiredPackageStatus.PACKAGE_VERSION_MISMATCH
594 if ( 594 ↛ 598line 594 didn't jump to line 598 because the condition on line 594 was never true
595 not installed_platform_id
596 or installed_platform_id != self.platform.id
597 ):
598 return RequiredPackageStatus.PACKAGE_PLATFORM_MISMATCH
600 if installed_platform_version != util.get_apio_version_str(): 600 ↛ 601line 600 didn't jump to line 601 because the condition on line 600 was never true
601 return RequiredPackageStatus.PACKAGE_APIO_VERSION_MISMATCH
603 true_src_url = self._construct_package_download_url(package_config)
605 if installed_src_url != true_src_url: 605 ↛ 606line 605 didn't jump to line 606 because the condition on line 605 was never true
606 return RequiredPackageStatus.PACKAGE_URL_MISMATCH
608 return RequiredPackageStatus.PACKAGE_OK
610 def scan_packages(self) -> PackagesScanResults:
611 """Scans the available and installed packages and returns
612 the findings as a PackageScanResults object."""
614 result = PackagesScanResults({}, {})
616 # -- Scan the required packages.
617 for package_name in self.required_packages:
618 package_status: RequiredPackageStatus = (
619 self.classify_required_package_status(package_name)
620 )
621 result.required_packages[package_name] = package_status
623 # -- Scan the installed packages and identify orphan packages.
624 for package_name in self.installed_packages:
625 if package_name not in self.required_packages: 625 ↛ 626line 625 didn't jump to line 626 because the condition on line 625 was never true
626 result.orphans[package_name] = OrphanType.ORPHAN_PACKAGE
628 # -- Scan the packages directory and identify orphan dirs and files.
629 for path in self.packages_dir.glob("*"):
630 base_name = os.path.basename(path)
631 assert isinstance(base_name, str), type(base_name)
633 # -- Ignore the installed packages index file.
634 if base_name == "installed_packages.json":
635 continue
637 # -- I the dir entry is of a required or orphan package, skip it,
638 # -- it will be covered by the package handling.
639 if (
640 base_name in result.required_packages
641 or base_name in result.orphans
642 ):
643 continue
645 # -- Classify the orphan as a dir or file.
646 if path.is_dir(): 646 ↛ 649line 646 didn't jump to line 649 because the condition on line 646 was always true
647 result.orphans[base_name] = OrphanType.ORPHAN_DIR
648 else:
649 result.orphans[base_name] = OrphanType.ORPHAN_FILE
651 # -- All done
652 return result
654 def _maybe_load_installed_packages_file(self):
655 """Load the installed packages index file if exists, e.g.
656 ~/.apio/packages/installed_packages.json, populates
657 self.installed_packages with the data read.
658 """
660 # -- Do nothing if the file doesn't exist.
661 if not self._packages_index_path.exists():
662 return
664 # -- Read the file as a json dict. Handle invalid content
665 # -- gracefully, since this runs on every apio command.
666 try:
667 with open(self._packages_index_path, "r", encoding="utf8") as f:
668 self.installed_packages = json.load(f)
670 # -- Perform a shallow sanity check.
671 # -- TODO: Do a full json validation.
672 assert isinstance(
673 self.installed_packages, dict
674 ), "Install packages not a dict"
675 for name, info in self.installed_packages.items():
676 assert isinstance(
677 info, dict
678 ), f"installed package '{name}' not a dict"
680 except (OSError, ValueError, AssertionError) as e:
681 fatal_error(
682 "Invalid installed packages index file "
683 + f"{self._packages_index_path}",
684 cause=e,
685 info="You can delete the file, "
686 + "Apio will recreate it automatically.",
687 )
689 def _save_installed_packages(self):
690 """Save the installed packages file"""
692 # -- Create the enclosing folder, if it does not exist yet
693 parent = self._packages_index_path.parent
694 if not parent.exists(): 694 ↛ 695line 694 didn't jump to line 695 because the condition on line 694 was never true
695 parent.mkdir()
697 # -- Write to installed packages file.
698 with open(self._packages_index_path, "w", encoding="utf8") as f:
699 json.dump(self.installed_packages, f, indent=4)
701 # -- Dump for debugging.
702 if is_debug(1): 702 ↛ 703line 702 didn't jump to line 703 because the condition on line 702 was never true
703 cout("Saved installed packages index:", style=EMPH3)
704 cout(json.dumps(self.installed_packages, indent=2))
706 def get_installed_package_version(self, package_name: str) -> str:
707 """Return the version of the given installed package. Fatal error
708 if the package is not installed have its info is corrupt.
709 """
710 package_info = self.installed_packages.get(package_name)
711 assert package_info is not None, package_name
712 package_version = package_info.get("version")
713 assert package_version is not None
714 return package_version
716 def get_installed_package_info(
717 self, package_name: str
718 ) -> tuple[str, str, str, str]:
719 """Return (package_version, platform_id) of the given installed
720 package. Values are replaced with "" if not installed or a value is
721 missing."""
722 package_info = self.installed_packages.get(package_name, {})
723 package_version = package_info.get("version", "")
724 platform_id = package_info.get("platform", "")
725 platform_version = package_info.get("loaded-by", "")
726 package_source_url = package_info.get("loaded-from", "")
727 return (
728 package_version,
729 platform_id,
730 platform_version,
731 package_source_url,
732 )
734 def add_package(self, name: str, version: str, platform_id: str, url: str):
735 """Add a package to the installed packages and save."""
737 # -- Updated the installed package data.
738 self.installed_packages[name] = {
739 "version": version,
740 "platform": platform_id,
741 "loaded-by": util.get_apio_version_str(),
742 "loaded-at": get_datetime_stamp(),
743 "loaded-from": url,
744 }
745 # self._save()
746 self._save_installed_packages()
748 def remove_package(self, name: str):
749 """Remove a package from the installed packages file. Do nothing
750 if not in installed packages."""
752 if name in self.installed_packages.keys(): 752 ↛ exitline 752 didn't return from function 'remove_package' because the condition on line 752 was always true
753 del self.installed_packages[name]
754 # self._save()
755 self._save_installed_packages()
757 def get_required_package_spec(self, package_name: str) -> dict:
758 """Returns the information of the package with given name.
759 The information is a JSON dict originated at packages.json().
760 Exits with an error message if the package is not defined.
761 """
762 package_info = self.required_packages.get(package_name, None)
763 if package_info is None: 763 ↛ 764line 763 didn't jump to line 764 because the condition on line 763 was never true
764 fatal_error(f"Unknown package '{package_name}'")
766 return package_info