Coverage for apio/apio_context.py: 87%
256 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 03:53 +0000
1"""The apio context."""
3# -*- coding: utf-8 -*-
4# -- This file is part of the Apio project
5# -- (C) 2016-2019 FPGAwars
6# -- Author Jesús Arroyo
7# -- License GPLv2
9import os
10import platform
11from dataclasses import dataclass
12from enum import Enum, unique
13from pathlib import Path
14import json5
15from apio.common.apio_console import cout, cstyle, fatal_error
16from apio.common.apio_styles import INFO, EMPH1, EMPH2, EMPH3
17from apio.common.common_util import env_build_path
18from apio.managers.profile import Profile
19from apio.managers.remote_config import RemoteConfig, RemoteConfigPolicy
20from apio.utils import util, env_options, apio_platforms
21from apio.utils.apio_platforms import ApioPlatform
22from apio.managers.project import Project, load_project_from_file
23from apio.managers.package_manager import PackageManager
24from apio.managers.apio_definitions import ApioDefinitions
25from apio.utils.resource_util import (
26 ProjectResources,
27 collect_project_resources,
28 validate_config,
29 validate_packages,
30)
32# ---------- RESOURCES
33RESOURCES_DIR = "resources"
36# ---------------------------------------
37# ---- File: resources/packages.jsonc
38# --------------------------------------
39# -- This file contains all the information regarding the available apio
40# -- packages: Repository, version, name...
41PACKAGES_JSONC = "packages.jsonc"
44# -----------------------------------------
45# ---- File: resources/config.jsonc
46# -----------------------------------------
47# -- General config information.
48CONFIG_JSONC = "config.jsonc"
51@dataclass(frozen=True)
52class EnvMutations:
53 """Contains mutations to the system env."""
55 # -- List of env vars to unset.
56 unset_vars: list[str]
58 # -- PATH items to add.
59 paths: list[str]
61 # -- Dict with env vars name/value to set.
62 set_vars: dict[str, str]
65@unique
66class ProjectPolicy(Enum):
67 """Represents the possible context policies regarding loading apio.ini.
68 and project related information."""
70 # -- Project information is not loaded.
71 NO_PROJECT = 1
72 # -- Project information is loaded if apio.ini is found.
73 PROJECT_OPTIONAL = 2
74 # -- Apio.ini is required and project information must be loaded.
75 PROJECT_REQUIRED = 3
78@unique
79class PackagesPolicy(Enum):
80 """Represents the possible context policies regarding loading apio.ini.
81 and project related information."""
83 # -- Do not change the package state, they may exist or not, updated or
84 # -- not. This policy requires project policy NO_PROJECT and with it,
85 # -- the definitions are not loaded.
86 IGNORE_PACKAGES = 1
87 # -- Normal policy, verify that the packages are installed correctly and
88 # -- update them if needed.
89 ENSURE_PACKAGES = 2
92class ApioContext:
93 """Apio context. Class for accessing apio resources and configurations."""
95 # pylint: disable=too-many-instance-attributes
97 # -- List of allowed instance vars.
98 __slots__ = (
99 "project_policy",
100 "apio_home_dir",
101 "apio_packages_dir",
102 "config",
103 "profile",
104 "remote_config",
105 "package_manager",
106 "platform",
107 "platform_id",
108 "scons_shell_id",
109 "all_packages",
110 "required_packages",
111 "env_was_already_set",
112 "_project_dir",
113 "_project",
114 "_project_resources",
115 "definitions",
116 )
118 def __init__(
119 self,
120 *,
121 project_policy: ProjectPolicy,
122 remote_config_policy: RemoteConfigPolicy,
123 packages_policy: PackagesPolicy,
124 project_dir_arg: Path | None = None,
125 env_arg: str | None = None,
126 report_env=True,
127 ):
128 """Initializes the ApioContext object.
130 'project_policy', 'config_policy', and 'packages_policy' are modifiers
131 that controls the initialization of the context.
133 'project_dir_arg' is an optional user specification of the project dir.
134 Must be None if project_policy is NO_PROJECT.
136 'env_arg' is an optional command line option value that select the
137 apio.ini env if the project is loaded. it makes sense only when
138 project_policy is PROJECT_REQUIRED (enforced by an assertion).
140 If an apio.ini project is loaded, the method prints to the user the
141 selected env and board, unless if report_env = False.
142 """
144 # pylint: disable=too-many-arguments
145 # pylint: disable=too-many-statements
147 # -- Sanity check the policies.
148 assert isinstance(project_policy, ProjectPolicy)
149 assert isinstance(remote_config_policy, RemoteConfigPolicy)
150 assert isinstance(packages_policy, PackagesPolicy)
152 if packages_policy == PackagesPolicy.IGNORE_PACKAGES:
153 assert project_policy == ProjectPolicy.NO_PROJECT
155 # -- Inform as soon as possible about the list of apio env options
156 # -- that modify its default behavior.
157 defined_env_options = env_options.get_defined()
158 if defined_env_options: 158 ↛ 165line 158 didn't jump to line 165 because the condition on line 158 was always true
159 cout(
160 f"Active env options [{', '.join(defined_env_options)}].",
161 style=INFO,
162 )
164 # -- Store the project_policy
165 assert isinstance(
166 project_policy, ProjectPolicy
167 ), "Not an ApioContextScope"
168 self.project_policy = project_policy
170 # -- Sanity check, env_arg makes sense only when project_policy is
171 # -- PROJECT_REQUIRED.
172 if env_arg is not None:
173 assert project_policy == ProjectPolicy.PROJECT_REQUIRED
175 # -- A flag to indicate if the system env was already set in this
176 # -- apio session. Used to avoid multiple repeated settings that
177 # -- make the path longer and longer.
178 self.env_was_already_set = False
180 # -- Determine if we need to load the project, and if so, set
181 # -- self._project_dir to the project dir, otherwise, leave it None.
182 self._project_dir: Path | None = None
183 if project_policy == ProjectPolicy.PROJECT_REQUIRED:
184 self._project_dir = util.user_directory_or_cwd(
185 project_dir_arg, description="Project", must_exist=True
186 )
187 elif project_policy == ProjectPolicy.PROJECT_OPTIONAL:
188 project_dir = util.user_directory_or_cwd(
189 project_dir_arg, description="Project", must_exist=False
190 )
191 if (project_dir / "apio.ini").exists():
192 self._project_dir = project_dir
193 else:
194 assert (
195 project_policy == ProjectPolicy.NO_PROJECT
196 ), f"Unexpected project policy: {project_policy}"
197 assert (
198 project_dir_arg is None
199 ), "project_dir_arg specified for project policy None"
201 # -- Determine apio home and packages dirs
202 self.apio_home_dir: Path = util.resolve_home_dir()
203 self.apio_packages_dir: Path = util.resolve_packages_dir(
204 self.apio_home_dir
205 )
207 # -- Get the jsonc source dirs.
208 resources_dir = util.get_path_in_apio_package(RESOURCES_DIR)
210 # -- Read and validate the config information
211 self.config = self._load_resource_file(CONFIG_JSONC, resources_dir)
212 validate_config(self.config)
214 # -- Read the user profile from ~/.apio/profile.json.
215 self.profile = Profile(
216 self.apio_home_dir,
217 )
219 # -- Read remote config information, from local cache or remotely..
220 remote_config_url = env_options.get(
221 env_options.APIO_REMOTE_CONFIG_URL,
222 default=self.config["remote-config-url"],
223 )
224 remote_config_ttl_days = self.config["remote-config-ttl-days"]
225 remote_config_retry_minutes = self.config[
226 "remote-config-retry-minutes"
227 ]
229 self.remote_config = RemoteConfig(
230 self.apio_home_dir,
231 str(remote_config_url),
232 remote_config_ttl_days,
233 remote_config_retry_minutes,
234 remote_config_policy,
235 )
237 # -- Get the underlying platform information.
238 self.platform: ApioPlatform = apio_platforms.get_apio_platform()
239 self.platform_id: str = self.platform.id
241 # -- Determine the shell id that scons will use.
242 # -- See _determine_scons_shell_id() for possible values.
243 self.scons_shell_id = self._determine_scons_shell_id(self.platform)
245 # -- Read the apio packages information
246 self.all_packages = self._load_resource_file(
247 PACKAGES_JSONC, resources_dir
248 )
249 validate_packages(self.all_packages)
251 # -- Expand in place the env templates in all_packages.
252 ApioContext._resolve_package_envs(
253 self.all_packages, self.apio_packages_dir
254 )
256 # -- The subset of packages that are applicable to this platform.
257 self.required_packages = self._select_required_packages_for_platform(
258 self.all_packages,
259 self.platform_id,
260 )
262 # -- Instantiate the package manager. All self.* args were already
263 # -- initialized above.
264 self.package_manager: PackageManager = PackageManager(
265 remote_config=self.remote_config,
266 required_packages=self.required_packages,
267 platform=self.platform,
268 apio_home_dir=self.apio_home_dir,
269 packages_dir=self.apio_packages_dir,
270 )
272 # -- Apply package policy
274 # -- Case 1: IGNORE_PACKAGES
275 if packages_policy == PackagesPolicy.IGNORE_PACKAGES:
276 self.definitions = None
278 # -- Case 2: ENSURE_PACKAGES
279 else:
280 assert packages_policy == PackagesPolicy.ENSURE_PACKAGES
282 # -- Install missing packages. At this point, the fields that are
283 # -- required by self.package_manager are already initialized.
284 # --
285 # -- TODO: Set verbose=True if APIO_DEBUG is above some level.
286 self.package_manager.install_missing_packages_on_the_fly(
287 verbose=False
288 )
290 # -- Load the boards, fpgas, and programmer definitions, including
291 # -- optional custom overrides in project's dir.
292 self.definitions = ApioDefinitions(
293 self.get_package_dir("definitions"),
294 self._project_dir,
295 )
297 # -- If we determined that we need to load the project, load the
298 # -- apio.ini data.
299 self._project: Project | None = None
300 self._project_resources: ProjectResources | None = None
302 if self._project_dir:
303 # -- If we have a project, we must also have definitions.
304 assert self.definitions is not None
306 # -- Load the project object
307 self._project = load_project_from_file(
308 self._project_dir, env_arg, self.definitions.boards
309 )
310 assert self.has_project, "init(): project not loaded"
311 # -- Inform the user about the active env, if needed..
312 if report_env:
313 self.report_env()
314 # -- Collect and validate the project resources.
315 # -- The project is already validated to have the required "board.
316 self._project_resources = collect_project_resources(
317 self._project.get_str_option("board"),
318 self.definitions,
319 )
320 else:
321 assert not self.has_project, "init(): project loaded"
323 def report_env(self):
324 """Report to the user the env and board used. Asserts that the
325 project is loaded."""
326 # -- Do not call if project is not loaded.
327 assert self.has_project
329 # -- Env name string in color
330 styled_env_name = cstyle(self.project.env_name, style=EMPH1)
332 # -- Board id string in color
333 styled_board_id = cstyle(
334 self.project.get_str_option("board"), style=EMPH1
335 )
337 # -- Report.
338 cout(f"Using env {styled_env_name} ({styled_board_id})")
340 @property
341 def has_project(self):
342 """Returns True if the project is loaded."""
343 return self._project is not None
345 @property
346 def project_dir(self):
347 """Returns the project dir. Should be called only if has_project_loaded
348 is true."""
349 assert self.has_project, "project_dir(): project is not loaded"
350 assert self._project_dir, "project_dir(): missing value."
351 return self._project_dir
353 @property
354 def project(self) -> Project:
355 """Return the project. Should be called only if has_project() is
356 True."""
357 # -- Failure here is a programming error, not a user error.
358 assert self.has_project, "project(): project is not loaded"
359 assert self._project is not None
360 return self._project
362 @property
363 def project_resources(self) -> ProjectResources:
364 """Return the project resources. Should be called only if
365 has_project() is True."""
366 # -- Failure here is a programming error, not a user error.
367 assert self.has_project, "project(): project is not loaded"
368 assert self._project_resources is not None
369 return self._project_resources
371 @property
372 def env_build_path(self) -> Path:
373 """Returns the relative path of the current env build directory from
374 the project dir. Should be called only when has_project is True."""
375 assert self.has_project, "project(): project is not loaded"
376 return env_build_path(self.project.env_name)
378 @classmethod
379 def _load_resource_file(cls, name: str, resources_dir: Path) -> dict:
380 """Load a .jsonc resource file and return its content as a
381 json dict."""
383 # pylint: disable=broad-exception-caught
385 # -- Construct file path.
386 filepath = resources_dir / name
388 # -- Read the and parse the jsonc file
389 try:
390 jsonc_text = filepath.read_text(encoding="utf-8")
391 json_dict = json5.loads(jsonc_text)
392 except Exception as e:
394 fatal_error(
395 f"Failed to read and parse resource file {name}", cause=e
396 )
398 # -- Return the object for the resource
399 return json_dict
401 @staticmethod
402 def _expand_env_values(template: str, package_path: Path) -> str:
403 """Fills a packages env value template as they appear in
404 packages.jsonc. Currently it recognizes only a single place holder
405 '%p' representing the package absolute path. The '%p" can appear only
406 at the beginning of the template.
408 E.g. '%p/bin' -> '/users/user/.apio/packages/drivers/bin'
410 NOTE: This format is very basic but is sufficient for the current
411 needs. If needed, extend or modify it.
412 """
414 # Case 1: No place holder -> no change.
415 if "%p" not in template: 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true
416 return template
418 # Case 2: The template contains only the placeholder.
419 if template == "%p": 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 return str(package_path)
422 # Case 3: The place holder is the prefix of the template's path.
423 if template.startswith("%p/"): 423 ↛ 427line 423 didn't jump to line 427 because the condition on line 423 was always true
424 return str(package_path / template[3:])
426 # Case 4: Unsupported.
427 raise RuntimeError(f"Invalid env template: [{template}]")
429 @staticmethod
430 def _resolve_package_envs(
431 packages_: dict[str, dict], packages_dir: Path
432 ) -> None:
433 """Resolve in-place the path and var value templates in the
434 given packages dictionary. For example, %p is replaced with
435 the package's absolute path."""
437 for package_name, package_config in packages_.items():
439 # -- Get the package root dir.
440 package_path = packages_dir / package_name
442 # -- Get the json 'env' section. We require it, even if empty,
443 # -- for clarity reasons.
444 assert "env" in package_config
445 package_env = package_config["env"]
447 # -- NOTE: There is no need to expand values in the "unset-env"
448 # -- section since it contains env names only.
450 # -- Expand the values in the "add-to-path" section, if any.
451 add_to_path_section = package_env.get("add-to-path", [])
452 for i, path_template in enumerate(add_to_path_section):
453 add_to_path_section[i] = ApioContext._expand_env_values(
454 path_template, package_path
455 )
457 # -- Expand the values in the "add-env-vars" section, if any.
458 add_env_vars_section = package_env.get("add-env-vars", {})
459 for var_name, var_value in add_env_vars_section.items():
460 add_env_vars_section[var_name] = (
461 ApioContext._expand_env_values(var_value, package_path)
462 )
464 # -- Expand the values in the "define-consts" section, if any.
465 define_consts_section = package_env.get("define-consts", {})
466 for const_name, const_value in define_consts_section.items():
467 define_consts_section[const_name] = (
468 ApioContext._expand_env_values(const_value, package_path)
469 )
471 def get_package_dir(self, package_name: str) -> Path:
472 """Returns the root path of a package with given name."""
474 return self.apio_packages_dir / package_name
476 def get_tmp_dir(self, create: bool = True) -> Path:
477 """Return the tmp dir under the apio home dir. If 'create' is true
478 create the dir and its parents if they do not exist."""
479 tmp_dir = self.apio_home_dir / "tmp"
480 if create:
481 tmp_dir.mkdir(parents=True, exist_ok=True)
482 return tmp_dir
484 @staticmethod
485 def _determine_scons_shell_id(apio_platform: ApioPlatform) -> str:
486 """
487 Returns a simplified string name of the shell that SCons will use
488 for executing shell-dependent commands. See code below for possible
489 values.
490 """
492 # pylint: disable=too-many-return-statements
494 # -- Handle windows.
495 if apio_platform.is_windows: 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true
496 comspec = os.environ.get("COMSPEC", "").lower()
497 if "powershell.exe" in comspec or "pwsh.exe" in comspec:
498 return "powershell"
499 if "cmd.exe" in comspec:
500 return "cmd"
501 return "unknown"
503 # -- Handle the rest (macOS, Linux, etc.)
504 shell_path = os.environ.get("SHELL", "").lower()
505 if "bash" in shell_path: 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true
506 return "bash"
507 if "zsh" in shell_path: 507 ↛ 508line 507 didn't jump to line 508 because the condition on line 507 was never true
508 return "zsh"
509 if "fish" in shell_path: 509 ↛ 510line 509 didn't jump to line 510 because the condition on line 509 was never true
510 return "fish"
511 if "dash" in shell_path: 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true
512 return "dash"
513 if "ksh" in shell_path: 513 ↛ 514line 513 didn't jump to line 514 because the condition on line 513 was never true
514 return "ksh"
515 if "csh" in shell_path or "tcsh" in shell_path: 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true
516 return "cshell"
517 return "unknown"
519 @staticmethod
520 def _select_required_packages_for_platform(
521 all_packages: dict[str, dict],
522 platform_id: str,
523 ) -> dict:
524 """Given a dictionary with the packages.jsonc packages infos,
525 returns subset dictionary with packages that are available for
526 'platform_id'.
527 """
529 # -- Dict of all supported platforms.
530 all_apio_platforms = apio_platforms.get_all_apio_platforms()
532 # -- If fails, this is a programming error.
533 assert platform_id in all_apio_platforms, platform
535 # -- Final dict with the output packages
536 filtered_packages = {}
538 # -- Check all the packages
539 for package_name in all_packages.keys():
541 # -- Get the package info.
542 package_info = all_packages[package_name]
544 # -- Get the list of platforms ids on which this package is
545 # -- available. The package is available on all platforms unless
546 # -- restricted by the ""restricted-to-platforms" field.
547 required_for_platforms = package_info.get(
548 "restricted-to-platforms", all_apio_platforms.keys()
549 )
551 # -- Sanity check that all platform ids are valid. If fails it's
552 # -- a programming error.
553 for p in required_for_platforms:
554 assert p in all_apio_platforms, platform
556 # -- If available for 'platform_id', add it.
557 if platform_id in required_for_platforms:
558 filtered_packages[package_name] = all_packages[package_name]
560 # -- Return the subset dict with the packages for 'platform_id'.
561 return filtered_packages
563 @property
564 def is_linux(self) -> bool:
565 """Returns True iff underlying platform is a Linux."""
566 return self.platform.is_linux
568 @property
569 def is_darwin(self) -> bool:
570 """Returns True iff underlying platform is a Mac OSX."""
571 return self.platform.is_darwin
573 @property
574 def is_windows(self) -> bool:
575 """Returns True iff underlying platform is a Windows."""
576 return self.platform.is_windows
578 def _get_env_mutations_for_packages(self) -> EnvMutations:
579 """Collects the env mutation for each of the defined packages,
580 in the order they are defined."""
582 unset_vars: list[str] = []
583 paths: list[str] = []
584 set_vars: dict[str, str] = {}
585 for _, package_config in self.required_packages.items():
586 # -- Get the json 'env' section. We require it, even if it's empty,
587 # -- for clarity reasons.
588 assert "env" in package_config
589 package_env = package_config["env"]
591 # -- Collect the env vars to delete.
592 delete_env_vars_section = package_env.get("delete-env-vars", [])
593 for var_name in delete_env_vars_section:
594 # -- Detect duplicates.
595 assert var_name not in unset_vars, var_name
596 unset_vars.append(var_name)
598 # -- Collect the path values.
599 package_paths = package_env.get("add-to-path", [])
600 paths.extend(package_paths)
602 # -- Collect the env vars to add (name, value) pairs.
603 add_env_vars_section = package_env.get("add-env-vars", {})
604 for var_name, var_value in add_env_vars_section.items():
605 # -- Detect duplicates.
606 assert var_name not in set_vars, var_name
607 set_vars[var_name] = var_value
609 return EnvMutations(unset_vars, paths, set_vars)
611 def _dump_env_mutations(self, mutations: EnvMutations) -> None:
612 """Dumps a user friendly representation of the env mutations."""
613 cout("Environment settings:", style=EMPH2)
615 # -- Print PATH mutations.
616 windows = self.is_windows
618 # -- Print unset vars.
619 for name in mutations.unset_vars:
620 styled_name = cstyle(name, style=EMPH3)
621 if windows: 621 ↛ 622line 621 didn't jump to line 622 because the condition on line 621 was never true
622 cout(f" set {styled_name}=")
623 else:
624 cout(f" unset {styled_name}")
626 # -- Dump paths.
627 for p in reversed(mutations.paths):
628 styled_name = cstyle("PATH", style=EMPH3)
629 if windows: 629 ↛ 630line 629 didn't jump to line 630 because the condition on line 629 was never true
630 cout(f" set {styled_name}={p};%PATH%")
631 else:
632 cout(f' {styled_name}="{p}:$PATH"')
634 # -- Print set vars.
635 for name, val in mutations.set_vars.items():
636 styled_name = cstyle(name, style=EMPH3)
637 if windows: 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true
638 cout(f" set {styled_name}={val}")
639 else:
640 cout(f' {styled_name}="{val}"')
642 def _apply_env_mutations(self, mutations: EnvMutations) -> None:
643 """Apply a given set of env mutations, while preserving their order."""
645 # -- Apply the unset var mutations
646 for name in mutations.unset_vars:
647 os.environ.pop(name, None)
649 # -- Apply the path mutations, while preserving order.
650 # -- NOTE: We treat the old path items as a single items.
651 old_val = os.environ["PATH"]
652 items = mutations.paths + [old_val]
653 new_val = os.pathsep.join(items)
654 os.environ["PATH"] = new_val
656 # -- Apply the set var mutations
657 for name, value in mutations.set_vars.items():
658 os.environ[name] = value
660 def set_env_for_packages(
661 self, *, quiet: bool = False, verbose: bool = False
662 ) -> None:
663 """Sets the environment variables for using all the that are
664 available for this platform, even if currently not installed.
666 The function sets the environment only on first call and in latter
667 calls skips the operation silently.
669 If quite is set, no output is printed. When verbose is set, additional
670 output such as the env vars mutations are printed, otherwise, a minimal
671 information is printed to make the user aware that they commands they
672 see are executed in a modified env settings.
673 """
675 # -- If this fails, this is a programming error. Quiet and verbose
676 # -- cannot be combined.
677 assert not (quiet and verbose), "Can't have both quite and verbose."
679 # -- Collect the env mutations for all packages.
680 mutations = self._get_env_mutations_for_packages()
682 if verbose:
683 self._dump_env_mutations(mutations)
685 # -- If this is the first call in this apio invocation, apply the
686 # -- mutations. These mutations are temporary for the lifetime of this
687 # -- process and does not affect the user's shell environment.
688 # -- The mutations are also inherited by child processes such as the
689 # -- scons processes.
690 if not self.env_was_already_set: 690 ↛ exitline 690 didn't return from function 'set_env_for_packages' because the condition on line 690 was always true
691 self._apply_env_mutations(mutations)
692 self.env_was_already_set = True
693 if not verbose and not quiet:
694 cout("Setting shell vars.")