Coverage for apio/managers/package_manager.py: 77%
316 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-07 05:08 +0000
1# This file is part of the Apio project
2# (C) 2016-present FPGAwars
3# Project authors:
4# * Jesús Arroyo (Jesus89) (2016-2019)
5# * Juan Gonzalez (obijuan) (2019-2024)
6# * The Apio Team (2024-present)
7# License: GPLv2
9"""Package install/uninstall functionality.
10Used by the 'apio packages' command.
11"""
13import os
14import json
15from enum import Enum, unique
16from datetime import datetime
17from dataclasses import dataclass
18from typing import Any
19from pathlib import Path
20import shutil
21import tarfile
22from rich.progress import track
23from apio.common.apio_console import cout, cstyle, console, fatal_error
24from apio.common.apio_styles import SUCCESS, EMPH3
25from apio.common.debug_util import is_debug
26from apio.managers.downloader import FileDownloader
27from apio.utils import util
28from apio.common.apio_platforms import ApioPlatform
29from apio.managers.remote_config import RemoteConfig
30from apio.common.proto.stubs.apio_config_pb2 import ApioPackageSpec
31from apio.common.proto.stubs.apio_remote_config_pb2 import (
32 RemoteConfigPackageSpec,
33)
36@unique
37class RequiredPackageStatus(Enum):
38 """Represents the classification of a required package status."""
40 # -- NOTE: The string values here are using facing by the
41 # -- 'apio packages list' command.
43 PACKAGE_UNINSTALLED = "Uninstalled"
44 PACKAGE_DIR_MISSING = "Package dir missing"
45 PACKAGE_DIR_IS_A_FILE = "Package dir is a file"
46 PACKAGE_VERSION_MISMATCH = "Version mismatch"
47 PACKAGE_PLATFORM_MISMATCH = "Platform mismatch"
48 PACKAGE_APIO_VERSION_MISMATCH = "Apio version mismatch"
49 PACKAGE_URL_MISMATCH = "Source URL mismatch"
50 PACKAGE_OK = "OK"
52 @property
53 def is_ok(self) -> bool:
54 """Returns True if the status is of a legit package."""
55 return self == self.PACKAGE_OK
57 @property
58 def is_inconsistency(self) -> bool:
59 """Is it an inconsistency that requires fixing before installing the
60 uninstalled packages."""
61 return self not in (self.PACKAGE_UNINSTALLED, self.PACKAGE_OK)
64@unique
65class OrphanType(Enum):
66 """Represents the types of orphans (leftovers items)."""
68 # -- NOTE: The string values here are using facing by the
69 # -- 'apio packages list' command.
71 # -- Non required package in installed packages index, potentially
72 # -- it also has an entry with same name in the packages folder..
73 ORPHAN_PACKAGE = "Orphan package"
74 # -- Package dir that doesn't match a required or an orphan package.
75 ORPHAN_FILE = "Orphan file"
76 # -- A file in the packages dir that doesn't match a name of an orphan
77 # -- package.
78 ORPHAN_DIR = "Orphan dir"
81@dataclass
82class PackagesScanResults:
83 """Represents results of packages scan."""
85 # -- Names and statuses of required packages.
86 platform_packages: dict[str, RequiredPackageStatus]
88 # -- Name and types of package, dir, and file orphans.
89 orphans: dict[str, OrphanType]
91 def packages_installed_ok(self) -> bool:
92 """Returns true if all the required packages are installed ok,
93 regardless of other fixable errors."""
94 return all(status.is_ok for status in self.platform_packages.values())
96 def num_inconsistencies_to_fix(self) -> int:
97 """Returns the number of inconsistencies that require fixing before
98 installing any missing package."""
99 required_packages_inconsistencies = sum(
100 1
101 for status in self.platform_packages.values()
102 if status.is_inconsistency
103 )
104 # -- All orphan errors are considered to be inconsistencies.
105 orphans_errors = len(self.orphans)
106 return required_packages_inconsistencies + orphans_errors
108 def is_all_ok(self) -> bool:
109 """Return True if all packages are installed properly with no
110 issues."""
111 return self.packages_installed_ok() and len(self.orphans) == 0
113 def dump(self):
114 """Dump the content of this object. For debugging."""
115 cout()
116 cout("Package scan results:")
117 cout(f" required {self.platform_packages}")
118 cout(f" orphans {self.orphans}")
121def get_datetime_stamp(dt: datetime | None = None) -> str:
122 """Returns a string with time now as yyyy-mm-dd-hh-mm"""
123 if dt is None: 123 ↛ 125line 123 didn't jump to line 125 because the condition on line 123 was always true
124 dt = datetime.now()
125 return dt.strftime("%Y-%m-%d-%H-%M")
128def unpack_tgz(archive_file_path: Path, dest_dir: Path) -> None:
129 """Unpack a .tgz archive into the given dest directory. Exits with an
130 error if an archive item would be written outside of dest_dir, e.g.
131 '../foo', '/foo', or a link to outside of dest_dir."""
133 # -- Get the file name
134 archive_name = archive_file_path.name
136 # -- Check it's a ".tgz" file
137 if not archive_name.endswith(".tgz"):
138 fatal_error(f"Cannot unarchive'{archive_name}', it's not a .tgz file.")
140 # -- We use the tarfile 'data' filter which rejects items that would be
141 # -- written outside of the dest dir. Search 'extraction_filter' in
142 # -- https://docs.python.org/3/library/tarfile.html
144 # -- Extract all items while animating the progress bar.
145 with tarfile.open(archive_file_path) as tar_file:
147 items = tar_file.getmembers()
149 # -- Check all the items before extracting any of them. This way an
150 # -- unsafe archive is rejected before anything is written, and the
151 # -- error is not lost in the progress bar output.
152 for item in items:
153 # -- Pre-check that we don't write outside dest_dir, e.g. with
154 # -- ".." path in the tar file.
155 try:
156 tarfile.data_filter(item, str(dest_dir))
157 except tarfile.FilterError as e:
158 fatal_error(
159 f"Unsafe item '{item.name}' in archive '{archive_name}'.",
160 cause=e,
161 )
163 for i in track(
164 range(len(items)),
165 description="Unpacking ",
166 console=console(),
167 ):
168 item = items[i]
169 # -- Extract the item. The filter checks again, now that the
170 # -- previous items are on disk, e.g. 'a/../x' after the link
171 # -- 'a -> .'.
172 try:
173 tar_file.extract(item, dest_dir, filter="data")
174 except tarfile.FilterError as e:
175 fatal_error(
176 f"Unsafe item '{item.name}' in archive '{archive_name}'.",
177 cause=e,
178 )
181class PackageManager:
182 """Context for package managements operations.
183 This class provides the information needed for package management
184 operations. This is a subset of the information contained by ApioContext
185 and we use it, instead of passing the ApioContext, because we need to
186 perform package management operations (e.g. updating packages) before
187 the ApioContext object is fully initialized.
188 """
190 def __init__(
191 self,
192 remote_config: RemoteConfig,
193 platform_packages: dict[str, ApioPackageSpec],
194 platform: ApioPlatform,
195 apio_home_dir: Path,
196 packages_dir: Path,
197 ):
199 # pylint: disable=too-many-arguments
200 # pylint: disable=too-many-positional-arguments
202 # -- Same as ApioContext.remote_config
203 self.remote_config = remote_config
204 # -- Same as ApioContext.platform_packages
205 self.platform_packages = platform_packages
206 # -- platform: ApioPlatform
207 self.platform = platform
208 # -- Same as ApioContext.apio_home_dir
209 self.apio_home_dir = apio_home_dir
210 # -- Same as ApioContext.packages_dir
211 self.packages_dir = packages_dir
213 # -- Sanity checks.
214 assert isinstance(self.remote_config, RemoteConfig)
215 assert self.platform_packages
216 assert self.platform
217 assert self.packages_dir
219 # -- Initialized installed packages, a copy of
220 # -- installed-packages.json.
221 self.installed_packages: dict[str, Any] = {}
223 # -- Cache the packages index file path
224 # -- Ex. '/home/obijuan/.apio/packages/installed_packages.json'
225 self._packages_index_path = packages_dir / "installed_packages.json"
227 # -- Read the installed packages file, if exists.
228 self._maybe_load_installed_packages_file()
230 def required_package_dir(self, package_name: str) -> Path:
231 """Return the local root directory of the package with given name"""
232 # -- Validate the package name
233 assert package_name in self.platform_packages, package_name
234 # -- Construct the path
235 return self.packages_dir / package_name
237 def _construct_package_download_url(
238 self,
239 package_remote_config: RemoteConfigPackageSpec,
240 ) -> str:
241 """Construct the download URL for the given package name and
242 version."""
244 # -- Create vars mapping.
245 url_vars = {
246 "${PLATFORM}": self.platform.id,
247 "${YYYYMMDD}": package_remote_config.release.tag.replace("-", ""),
248 }
249 if is_debug(1): 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 cout(f"Package URL vars: {url_vars}")
252 # -- Define the url parts.
253 url_parts = [
254 "https://github.com/",
255 package_remote_config.repository.organization,
256 "/",
257 package_remote_config.repository.name,
258 "/releases/download/",
259 package_remote_config.release.tag,
260 "/",
261 package_remote_config.release.package,
262 ]
264 if is_debug(1): 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 cout(f"package url parts = {url_parts}")
267 # -- Concatenate the URL parts.
268 url = "".join(url_parts)
270 if is_debug(1): 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true
271 cout(f"Combined package url: {url}")
273 # -- Replace placeholders with values.
274 for name, val in url_vars.items():
275 url = url.replace(name, val)
277 if is_debug(1): 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true
278 cout(f"Resolved package url: {url}")
280 # -- All done.
281 return url
283 def _download_package_file(
284 self, url: str, dir_path: Path, package_name: str
285 ) -> Path:
286 """Download the given file (url). Return the path of local destination
287 file. Exits with a user message and error code if any error.
289 * INPUTS:
290 * url: File to download
291 * OUTPUTS:
292 * The path of the destination file
293 """
295 filepath: Path | None = None
297 try:
298 # -- Object for downloading the file
299 downloader = FileDownloader(url, dir_path)
301 # -- Get the destination path
302 filepath = downloader.destination
304 downloader.download()
306 # -- If the user press Ctrl-C (Abort)
307 except KeyboardInterrupt:
309 # -- Remove the file
310 if filepath and filepath.is_file():
311 filepath.unlink()
313 # -- Inform the user
314 fatal_error("User aborted download")
316 except IOError as exc:
317 fatal_error(
318 f"Failed to download package '{package_name}'", cause=exc
319 )
321 # -- Return the destination path
322 return filepath
324 def _delete_package_dir(self, package_name: str, verbose: bool) -> None:
325 """Delete the directory of the package with given name."""
326 package_path = self.packages_dir / package_name
328 # -- If doesn't exist, ignore silently.
329 if not package_path.exists():
330 return
332 if verbose: 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true
333 cout(f"Deleting {str(package_path)}")
335 if package_path.is_dir(): 335 ↛ 340line 335 didn't jump to line 340 because the condition on line 335 was always true
336 # -- Sanity check the path and delete.
337 assert "packages" in str(package_path).lower(), package_path
338 shutil.rmtree(package_path)
339 else:
340 package_path.unlink()
342 # -- Confirm
343 if package_path.exists(): 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true
344 fatal_error(
345 f"Package dir deletion failed: {str(package_path.absolute())}"
346 )
348 def scan_and_fix_inconsistencies(self) -> bool:
349 """Scan the packages and fix if there are errors. Returns true
350 if the packages are installed ok."""
352 # -- Scan the packages.
353 scan: PackagesScanResults = self.scan_packages()
355 # -- If there are fixable errors, fix them.
356 if scan.num_inconsistencies_to_fix() > 0:
357 self._fix_inconsistencies(scan)
359 # -- Return a flag that indicates if all packages are installed ok. We
360 # -- use a scan from before the fixing but the fixing does not touch
361 # -- installed ok packages.
362 return scan.packages_installed_ok()
364 def install_missing_packages_on_the_fly(self, verbose=False) -> None:
365 """Install on the fly any missing packages. Does not print a thing if
366 all packages are already ok. This function is intended for on demand
367 package fetching by commands such as apio build, and thus is allowed
368 to use fetched remote config instead of fetching a fresh one. Exists
369 with error code if any error."""
371 # -- Scan and fix broken package.
372 # -- Since this is a on-the-fly operation, we don't require a fresh
373 # -- remote config file for required packages versions.
374 installed_ok = self.scan_and_fix_inconsistencies()
376 # -- If the packages are installed we are done, we are done.
377 if installed_ok: 377 ↛ 388line 377 didn't jump to line 388 because the condition on line 377 was always true
378 # -- Final sanity check of the packages.
379 self.check_packages_post_install()
380 # -- Installed ok.
381 return
383 # -- Here when we need to install some packages. Since we just fixed
384 # -- we can't have broken or packages with version mismatch, just
385 # -- installed ok, and not installed.
386 # --
387 # -- Get lists of installed and required packages.
388 installed_packages = self.installed_packages
389 platform_packages_names = self.platform_packages.keys()
391 # -- Install any required package that is not installed.
392 for package_name in platform_packages_names:
393 if package_name not in installed_packages:
394 self.install_package(
395 package_name=package_name,
396 force_reinstall=False,
397 verbose=verbose,
398 )
400 # -- Here all packages should be ok but we check again just in case.
401 scan_results = self.scan_packages()
402 if not scan_results.is_all_ok():
403 fatal_error(
404 "Packages issues detected. Use "
405 + "'apio packages list' to investigate."
406 )
408 # -- Final sanity check of the packages.
409 self.check_packages_post_install()
411 def install_package(
412 self,
413 *,
414 package_name: str,
415 force_reinstall: bool,
416 verbose: bool,
417 ) -> None:
418 """Install a given package.
420 'package_name' is the package name, e.g. 'examples' or 'oss-cad-suite'.
421 'force' indicates if to perform the installation even if a matching
422 package is already installed.
423 'explicit' indicates that the user specified the package name(s)
424 explicitly and thus expect more feedback in case of a 'no change'
425 'verbose' indicates if to print extra information.
427 Returns normally if no error, exits the program with an error status
428 and a user message if an error is detected.
429 """
431 # -- Force verbose if debug.
432 if is_debug(1): 432 ↛ 433line 432 didn't jump to line 433 because the condition on line 432 was never true
433 verbose = True
435 # -- Caller is responsible to check check that package name is valid
436 # -- on this platform.
437 assert package_name in self.platform_packages, package_name
439 # -- Set up installation announcement
440 pending_announcement: str | None = cstyle(
441 f"Installing apio package '{package_name}'", style=EMPH3
442 )
444 # -- If in chatty mode, announce now and clear. Otherwise we will
445 # -- announce later only if actually installing.
446 if verbose and pending_announcement: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true
447 cout(pending_announcement)
448 pending_announcement = None
450 # -- Get package remote config from the cache. Caller can refresh the
451 # -- cache with the latest remote config if desired.
452 package_config: RemoteConfigPackageSpec = (
453 self.remote_config.get_package_config(package_name)
454 )
456 # -- Get the version we should have.
457 target_version = package_config.release.tag.replace("-", ".")
459 # -- If not forcing and the target version already installed then
460 # -- nothing to do and we leave quietly.
461 if not force_reinstall:
462 # -- Get the package status
463 package_status = self.classify_required_package_status(
464 package_name
465 )
467 if verbose: 467 ↛ 468line 467 didn't jump to line 468 because the condition on line 467 was never true
468 cout(
469 f"Package {package_name} status is"
470 + f" '{package_status.value}'"
471 )
473 # -- If the package is OK then nothing to do.
474 if package_status.is_ok:
475 if verbose: 475 ↛ 476line 475 didn't jump to line 476 because the condition on line 475 was never true
476 cout(
477 f"Package {package_name} version {target_version} "
478 + "is already installed OK",
479 style=SUCCESS,
480 )
481 return
483 # -- Here we need to fetch and install so can be more chatty.
485 # -- Here we actually do the work. Announce if we haven't done it yet.
486 if pending_announcement: 486 ↛ 490line 486 didn't jump to line 490 because the condition on line 486 was always true
487 cout(pending_announcement)
488 pending_announcement = None
490 cout(f"Fetching version {target_version} ({self.platform.id})")
492 # -- Construct the download URL.
493 download_url = self._construct_package_download_url(package_config)
494 if verbose: 494 ↛ 495line 494 didn't jump to line 495 because the condition on line 494 was never true
495 cout(f"Download URL: {download_url}")
497 # -- Prepare the packages directory.
498 self.packages_dir.mkdir(exist_ok=True)
500 # -- Prepare the package directory.
501 package_dir = self.packages_dir / package_name
502 cout(f"Package dir: {package_dir}")
504 # -- Download the package file from the remote server.
505 local_package_file = self._download_package_file(
506 download_url, self.packages_dir, package_name
507 )
508 if verbose: 508 ↛ 509line 508 didn't jump to line 509 because the condition on line 508 was never true
509 cout(f"Local package file: {local_package_file}")
511 # -- Delete the old package dir, if exists, to avoid name conflicts and
512 # -- left over files.
513 self._delete_package_dir(package_name, verbose)
515 # -- Unpack the package. This creates a new package dir.
516 unpack_tgz(local_package_file, package_dir)
518 # -- Remove the package file. We don't need it anymore.
519 if verbose: 519 ↛ 520line 519 didn't jump to line 520 because the condition on line 519 was never true
520 cout(f"Deleting package file {local_package_file}")
521 local_package_file.unlink()
523 # -- Add package and save.
524 self.add_package(
525 package_name, target_version, self.platform.id, download_url
526 )
528 # -- Inform the user!
529 cout(f"Package '{package_name}' installed successfully", style=SUCCESS)
531 def _fix_inconsistencies(self, scan: PackagesScanResults) -> None:
532 """If the package scan result contains errors, fix them. This
533 does not install missing packages, just fixing inconsistencies."""
535 for package_name, package_status in scan.platform_packages.items():
536 if package_status.is_inconsistency:
537 cout(f"Uninstalling broken package '{package_name}'")
538 self._delete_package_dir(package_name, verbose=False)
539 self.remove_package(package_name)
541 for orphan_name, orphan_type in scan.orphans.items():
542 # -- Delete an unknown entry in the installed packages index.
543 if orphan_type == orphan_type.ORPHAN_PACKAGE: 543 ↛ 544line 543 didn't jump to line 544 because the condition on line 543 was never true
544 cout(f"Uninstalling unknown package '{orphan_name}'")
545 self.remove_package(orphan_name)
547 # -- Delete an unknown dir in the package dir.
548 elif orphan_type == orphan_type.ORPHAN_DIR: 548 ↛ 555line 548 didn't jump to line 555 because the condition on line 548 was always true
549 cout(f"Deleting unknown package dir '{orphan_name}'")
550 dir_path = self.packages_dir / orphan_name
551 assert "packages" in str(dir_path).lower(), dir_path
552 shutil.rmtree(dir_path)
554 # -- Delete an unknown file in the packages dir.
555 elif orphan_type == orphan_type.ORPHAN_FILE:
556 cout(f"Deleting unknown package file '{orphan_name}'")
557 file_path = self.packages_dir / orphan_name
558 assert "packages" in str(file_path).lower(), file_path
559 file_path.unlink()
561 # -- Unexpected orphan type.
562 else:
563 raise ValueError(f"Unknown orphan type: {orphan_type}")
565 def read_package_build_info(self, package_name: str) -> dict[str, Any]:
566 """Returns the BUILD-INFO.json of the package as a dict. Fatal
567 error if doesn't exist or can't parse."""
569 build_info_path = (
570 self.required_package_dir(package_name) / "BUILD-INFO.json"
571 )
573 # pylint: disable=broad-exception-caught
575 try:
576 with open(build_info_path, encoding="utf-8") as f:
577 build_info = json.load(f)
578 except Exception as e:
579 fatal_error(
580 f"Reading/parsing [{build_info_path}] failed.", cause=e
581 )
583 return build_info
585 def get_yosys_release_tag(self) -> str:
586 """Return the version tag (e.g. "2026-03-21") of the underlying Yosys.
587 This value is extract from the BUILD-INFO.json file of the apio
588 oss-cad-suite package."""
589 build_info = self.read_package_build_info("oss-cad-suite")
590 return build_info["yosys-release-tag"]
592 def check_packages_post_install(self):
593 """Called after the Apio packages were installed or fixed and are
594 believed to be correct. Performs additional validation of the apio
595 packages and exits with an error on any error."""
597 # -- Read the build info of the two packages.
598 build_info1 = self.read_package_build_info("oss-cad-suite")
599 build_info2 = self.read_package_build_info("openxc7")
601 # -- Extract the version of the underlying yosys
602 yosys_release_tag1 = build_info1["yosys-release-tag"]
603 yosys_release_tag2 = build_info2["yosys-release-tag"]
605 # -- Compare the versions.
606 if yosys_release_tag1 != yosys_release_tag2:
607 fatal_error(
608 'The packages "oss-cad-suite" and "openxc7" were built with '
609 + 'different "yosys-release-tag".',
610 "Their respective BUILD-INFO.json files "
611 + f'contain "{yosys_release_tag1}" vs "{yosys_release_tag2}"',
612 info="This typically happens due to corrupt packages or "
613 + "bad remote configuration by the Apio team.",
614 )
616 def classify_required_package_status(
617 self, name: str
618 ) -> RequiredPackageStatus:
619 """Classify existing or missing entry under the package directory."""
621 # pylint: disable=too-many-return-statements
623 # -- Check tha the package is a required one.
624 assert name in self.platform_packages, name
626 # -- Construct the package path.
627 package_path: Path = self.required_package_dir(name)
629 if name not in self.installed_packages:
630 return RequiredPackageStatus.PACKAGE_UNINSTALLED
632 if not package_path.exists():
633 return RequiredPackageStatus.PACKAGE_DIR_MISSING
635 if not package_path.is_dir(): 635 ↛ 636line 635 didn't jump to line 636 because the condition on line 635 was never true
636 return RequiredPackageStatus.PACKAGE_DIR_IS_A_FILE
638 # -- Get installed package info or "" if not installed.
639 (
640 installed_version,
641 installed_platform_id,
642 installed_platform_version,
643 installed_src_url,
644 ) = self.get_installed_package_info(name)
646 # -- Get the package's remote config
647 package_config: RemoteConfigPackageSpec = (
648 self.remote_config.get_package_config(name)
649 )
651 if ( 651 ↛ 656line 651 didn't jump to line 656 because the condition on line 651 was never true
652 not installed_version
653 or installed_version
654 != package_config.release.tag.replace("-", ".")
655 ):
656 return RequiredPackageStatus.PACKAGE_VERSION_MISMATCH
658 if ( 658 ↛ 662line 658 didn't jump to line 662 because the condition on line 658 was never true
659 not installed_platform_id
660 or installed_platform_id != self.platform.id
661 ):
662 return RequiredPackageStatus.PACKAGE_PLATFORM_MISMATCH
664 if installed_platform_version != util.get_apio_version_str(): 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true
665 return RequiredPackageStatus.PACKAGE_APIO_VERSION_MISMATCH
667 true_src_url = self._construct_package_download_url(package_config)
669 if installed_src_url != true_src_url: 669 ↛ 670line 669 didn't jump to line 670 because the condition on line 669 was never true
670 return RequiredPackageStatus.PACKAGE_URL_MISMATCH
672 return RequiredPackageStatus.PACKAGE_OK
674 def scan_packages(self) -> PackagesScanResults:
675 """Scans the available and installed packages and returns
676 the findings as a PackageScanResults object."""
678 result = PackagesScanResults({}, {})
680 # -- Scan the required packages.
681 for package_name in self.platform_packages:
682 package_status: RequiredPackageStatus = (
683 self.classify_required_package_status(package_name)
684 )
685 result.platform_packages[package_name] = package_status
687 # -- Scan the installed packages and identify orphan packages.
688 for package_name in self.installed_packages:
689 if package_name not in self.platform_packages: 689 ↛ 690line 689 didn't jump to line 690 because the condition on line 689 was never true
690 result.orphans[package_name] = OrphanType.ORPHAN_PACKAGE
692 # -- Scan the packages directory and identify orphan dirs and files.
693 for path in self.packages_dir.glob("*"):
694 base_name = os.path.basename(path)
695 assert isinstance(base_name, str), type(base_name)
697 # -- Ignore the installed packages index file.
698 if base_name == "installed_packages.json":
699 continue
701 # -- I the dir entry is of a required or orphan package, skip it,
702 # -- it will be covered by the package handling.
703 if (
704 base_name in result.platform_packages
705 or base_name in result.orphans
706 ):
707 continue
709 # -- Classify the orphan as a dir or file.
710 if path.is_dir(): 710 ↛ 713line 710 didn't jump to line 713 because the condition on line 710 was always true
711 result.orphans[base_name] = OrphanType.ORPHAN_DIR
712 else:
713 result.orphans[base_name] = OrphanType.ORPHAN_FILE
715 # -- All done
716 if is_debug(1): 716 ↛ 717line 716 didn't jump to line 717 because the condition on line 716 was never true
717 result.dump()
719 return result
721 def _maybe_load_installed_packages_file(self):
722 """Load the installed packages index file if exists, e.g.
723 ~/.apio/packages/installed_packages.json, populates
724 self.installed_packages with the data read.
725 """
727 # -- Do nothing if the file doesn't exist.
728 if not self._packages_index_path.exists():
729 return
731 # -- Read the file as a json dict. Handle invalid content
732 # -- gracefully, since this runs on every apio command.
733 try:
734 with open(self._packages_index_path, "r", encoding="utf8") as f:
735 self.installed_packages = json.load(f)
737 # -- Perform a shallow sanity check.
738 # -- TODO: Do a full json validation.
739 assert isinstance(
740 self.installed_packages, dict
741 ), "Install packages not a dict"
742 for name, info in self.installed_packages.items():
743 assert isinstance(
744 info, dict
745 ), f"installed package '{name}' not a dict"
747 except (OSError, ValueError, AssertionError) as e:
748 fatal_error(
749 "Invalid installed packages index file "
750 + f"{self._packages_index_path}",
751 cause=e,
752 info="You can delete the file, "
753 + "Apio will recreate it automatically.",
754 )
756 def _save_installed_packages(self):
757 """Save the installed packages file"""
759 # -- Create the enclosing folder, if it does not exist yet
760 parent = self._packages_index_path.parent
761 if not parent.exists(): 761 ↛ 762line 761 didn't jump to line 762 because the condition on line 761 was never true
762 parent.mkdir()
764 # -- Write to installed packages file.
765 with open(self._packages_index_path, "w", encoding="utf8") as f:
766 json.dump(self.installed_packages, f, indent=4)
768 # -- Dump for debugging.
769 if is_debug(1): 769 ↛ 770line 769 didn't jump to line 770 because the condition on line 769 was never true
770 cout("Saved installed packages index:", style=EMPH3)
771 cout(json.dumps(self.installed_packages, indent=2))
773 def get_installed_package_version(self, package_name: str) -> str:
774 """Return the version of the given installed package. Fatal error
775 if the package is not installed have its info is corrupt.
776 """
777 package_info = self.installed_packages.get(package_name)
778 assert package_info is not None, package_name
779 package_version = package_info.get("version")
780 assert package_version is not None
781 return package_version
783 def get_installed_package_info(
784 self, package_name: str
785 ) -> tuple[str, str, str, str]:
786 """Return (package_version, platform_id) of the given installed
787 package. Values are replaced with "" if not installed or a value is
788 missing."""
789 package_info = self.installed_packages.get(package_name, {})
790 package_version = package_info.get("version", "")
791 platform_id = package_info.get("platform", "")
792 platform_version = package_info.get("loaded-by", "")
793 package_source_url = package_info.get("loaded-from", "")
794 return (
795 package_version,
796 platform_id,
797 platform_version,
798 package_source_url,
799 )
801 def add_package(self, name: str, version: str, platform_id: str, url: str):
802 """Add a package to the installed packages and save."""
804 # -- Updated the installed package data.
805 self.installed_packages[name] = {
806 "version": version,
807 "platform": platform_id,
808 "loaded-by": util.get_apio_version_str(),
809 "loaded-at": get_datetime_stamp(),
810 "loaded-from": url,
811 }
812 # self._save()
813 self._save_installed_packages()
815 def remove_package(self, name: str):
816 """Remove a package from the installed packages file. Do nothing
817 if not in installed packages."""
819 if name in self.installed_packages.keys(): 819 ↛ exitline 819 didn't return from function 'remove_package' because the condition on line 819 was always true
820 del self.installed_packages[name]
821 # self._save()
822 self._save_installed_packages()
824 def get_required_package_spec(self, package_name: str) -> ApioPackageSpec:
825 """Returns the information of the package with given name.
826 The information is a proto buffer originated at packages.json().
827 Exits with an error message if the package is not defined.
828 """
829 package_info = self.platform_packages.get(package_name, None)
830 if package_info is None: 830 ↛ 831line 830 didn't jump to line 831 because the condition on line 830 was never true
831 fatal_error(f"Unknown package '{package_name}'")
833 return package_info