Coverage for apio/profile.py: 78%
267 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 01:55 +0000
1# -*- coding: utf-8 -*-
2# -- This file is part of the Apio project
3# -- (C) 2016-2019 FPGAwars
4# -- Author Jesús Arroyo
5# -- License GPLv2
6"""Manage the apio profile file"""
8import json
9import sys
10from enum import Enum
11from dataclasses import dataclass
12from datetime import datetime
13from typing import Dict, Optional, Any, List, Tuple
14from pathlib import Path
15import requests
16from jsonschema import validate
17from jsonschema.exceptions import ValidationError
18from apio.common import apio_console
19from apio.common.apio_console import cout, cerror
20from apio.common.apio_themes import THEMES_TABLE
21from apio.common.apio_styles import INFO, EMPH3, ERROR
22from apio.utils import util, jsonc
25def _check_json_dict(value: Any, desc: str) -> None:
26 """Raises a ValueError if 'value' is not a dict. For validating the
27 shape of values from user editable json files."""
28 if not isinstance(value, dict):
29 raise ValueError(
30 f"Expected {desc} to be a json dict, "
31 f"found {type(value).__name__}"
32 )
35# -- JSON schema for validating a remote config file.
36REMOTE_CONFIG_SCHEMA = {
37 "$schema": "https://json-schema.org/draft/2020-12/schema",
38 "type": "object",
39 "required": ["packages"],
40 "properties": {
41 # -- Packages
42 "packages": {
43 "type": "object",
44 "patternProperties": {
45 "^.*$": {
46 "type": "object",
47 "required": ["repository", "release"],
48 "properties": {
49 # -- Repository
50 "repository": {
51 "type": "object",
52 "required": ["organization", "name"],
53 "properties": {
54 # -- Repo organization. e.g. "fpgawars"
55 "organization": {"type": "string"},
56 # -- Repo name, e.g. 'examples'
57 "name": {"type": "string"},
58 },
59 "additionalProperties": False,
60 },
61 # -- Release.
62 "release": {
63 "type": "object",
64 "required": [
65 "tag",
66 "package",
67 ],
68 "properties": {
69 # -- Tag
70 "tag": {
71 "type": "string",
72 "pattern": r"^\d{4}\-\d{2}\-\d{2}$",
73 },
74 # -- Package
75 "package": {"type": "string"},
76 },
77 "additionalProperties": False,
78 },
79 },
80 "additionalProperties": False,
81 }
82 },
83 "additionalProperties": False,
84 }
85 },
86 "additionalProperties": False,
87}
90class RemoteConfigPolicy(Enum):
91 """Represents possible requirements from the remote config."""
93 # -- Config is being used but can be a cached value, as long that it's
94 # -- not too old.
95 CACHED_OK = 1
96 # -- Config is being used and a fresh copy is that was fetch in this
97 # -- invocation of Apio is required.
98 GET_FRESH = 2
101@dataclass(frozen=True)
102class PackageRemoteConfig:
103 """Contains a package info from the remote config."""
105 # -- E.g. "tools-oss-cad-suite"
106 repo_name: str
107 # -- E.g. "FPGAwars"
108 repo_organization: str
109 # -- E.g. "0.2.3"
110 release_version: str
111 # -- E.g. "${YYYY-MM-DD}""
112 release_tag: str
113 # -- E.g. "apio-oss-cad-suite-${PLATFORM}-${YYYYMMDD}.zip"
114 release_file: str
117def get_datetime_stamp(dt: Optional[datetime] = None) -> str:
118 """Returns a string with time now as yyyy-mm-dd-hh-mm"""
119 if dt is None:
120 dt = datetime.now()
121 return dt.strftime("%Y-%m-%d-%H-%M")
124def days_between_datetime_stamps(ts1: str, ts2: str, default: Any) -> int:
125 """Given two values generated by get_datetime_stamp(), return the
126 number of days from ts1 to ts2. The value can be negative if ts2 is
127 earlier than ts1. Returns the given 'default' value if either timestamp
128 is invalid."""
130 # -- The parsing format.
131 fmt = "%Y-%m-%d-%H-%M"
133 # -- Convert to timedates
134 try:
135 datetime1 = datetime.strptime(ts1, fmt)
136 datetime2 = datetime.strptime(ts2, fmt)
137 except ValueError:
138 return default
140 # -- Round to beginning of day.
141 day1 = datetime(datetime1.year, datetime1.month, datetime1.day)
142 day2 = datetime(datetime2.year, datetime2.month, datetime2.day)
144 # -- Compute the diff in days.
145 delta_days: int = (day2 - day1).days
147 # -- All done.
148 assert isinstance(delta_days, int)
149 return delta_days
152def minutes_between_datetime_stamps(ts1: str, ts2: str, default: Any) -> int:
153 """Return the number of minutes from ts1 to ts2 or default if any of
154 the timestamps is invalid."""
156 # -- The parsing format.
157 fmt = "%Y-%m-%d-%H-%M"
159 # -- Convert to timedates
160 try:
161 datetime1 = datetime.strptime(ts1, fmt)
162 datetime2 = datetime.strptime(ts2, fmt)
163 except ValueError:
164 return default
166 # -- Calculate the diff in minutes.
167 delta_minutes = int((datetime2 - datetime1).total_seconds() / 60)
168 assert isinstance(delta_minutes, int), type(delta_minutes)
169 return delta_minutes
172class Profile:
173 """Class for managing the apio profile file
174 ex. ~/.apio/profile.json
175 """
177 # pylint: disable=too-many-instance-attributes
179 # -- Only these instance vars are allowed.
180 __slots__ = (
181 "_profile_path",
182 "_packages_index_path",
183 "remote_config_url",
184 "remote_config_ttl_days",
185 "remote_config_retry_minutes",
186 "_remote_config_policy",
187 "_cached_remote_config",
188 "preferences",
189 "installed_packages",
190 )
192 def __init__(
193 self,
194 home_dir: Path,
195 packages_dir: Path,
196 remote_config_url_template: str,
197 remote_config_ttl_days: int,
198 remote_config_retry_minutes: int,
199 remote_config_policy: RemoteConfigPolicy,
200 ):
201 """remote_config_url_template is a url string with the
202 placeholder {major} and {minor} for the apio's major and minor
203 version. '"""
205 # pylint: disable=too-many-arguments
206 # pylint: disable=too-many-positional-arguments
208 # -- Sanity check
209 assert isinstance(remote_config_ttl_days, int)
210 assert 0 < remote_config_ttl_days <= 30
212 # -- Sanity check
213 assert isinstance(remote_config_retry_minutes, int)
214 assert 0 < remote_config_retry_minutes <= (60 * 24)
216 # -- Resolve and cache the remote config url. Replaced the placeholders
217 # -- with the major and minor versions of apio. Path version is
218 # -- not used.
219 ver_tuple = util.get_apio_version_tuple()
220 url = remote_config_url_template
221 url = url.replace("{major}", str(ver_tuple[0]))
222 url = url.replace("{minor}", str(ver_tuple[1]))
223 self.remote_config_url = url
225 # -- Save remote url ttl setting.
226 self.remote_config_ttl_days = remote_config_ttl_days
228 # -- Save the remote config fetch retry minutes.
229 self.remote_config_retry_minutes = remote_config_retry_minutes
231 # -- Save remote config policy.
232 self._remote_config_policy = remote_config_policy
234 # -- Verify that we resolved all the placeholders.
235 assert "{" not in self.remote_config_url, self.remote_config_url
237 if util.is_debug(1): 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true
238 cout(f"Remote config url: {self.remote_config_url}")
240 # ---- Set the default parameters
242 # User preferences
243 self.preferences = {}
245 # -- Installed package versions
246 self.installed_packages = {}
248 # -- A copy of remote config.
249 self._cached_remote_config = {}
251 # -- Cache the profile file path
252 # -- Ex. '/home/obijuan/.apio/profile.json'
253 self._profile_path = home_dir / "profile.json"
255 # -- Cache the packages index file path
256 # -- Ex. '/home/obijuan/.apio/packages/installed_packages.json'
257 self._packages_index_path = packages_dir / "installed_packages.json"
259 # -- Read the profile from file, if exists.
260 self._load_profile_file()
262 # -- Read the installed packages file, if exists.
263 self._load_installed_packages_file()
265 # -- Apply config policy
266 self._apply_remote_config_policy()
268 def _apply_remote_config_policy(self) -> None:
269 """Called after loading the profile file, to apply the remote config
270 policy for this invocation."""
272 # -- Case 1: A fresh config is required for the current command.
273 if self._remote_config_policy == RemoteConfigPolicy.GET_FRESH:
274 self._fetch_and_update_remote_config(error_is_fatal=True)
275 return
277 # -- Case 2: A fresh config is optional but there is no cached
278 # -- config so practically it's required.
279 assert self._remote_config_policy == RemoteConfigPolicy.CACHED_OK
280 if not self._cached_remote_config:
281 if util.is_debug(1): 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true
282 cout("Saved remote config is not available.", style=INFO)
283 self._fetch_and_update_remote_config(error_is_fatal=True)
284 return
286 # -- Case 3: May need to fetch a new config but can continue with
287 # -- the cached config in case of a fetch failure.
288 #
289 # -- Get the cached config metadata.
290 cashed_config_metadata = self._cached_remote_config.get("metadata", {})
291 last_fetch_timestamp = cashed_config_metadata.get("loaded-at", "")
292 last_fetch_url = cashed_config_metadata.get("loaded-from", "")
294 # -- Determine if we need a new config because the remote config URL
295 # -- was changed (e.g. with APIO_REMOTE_CONFIG_URL)
296 url_changed = last_fetch_url != self.remote_config_url
298 # -- Determine if there is time related reason to fetch a new config.
299 datetime_stamp_now = get_datetime_stamp()
300 days_since_last_fetch = days_between_datetime_stamps(
301 last_fetch_timestamp, datetime_stamp_now, default=99999
302 )
303 time_valid = 0 <= days_since_last_fetch < self.remote_config_ttl_days
305 # -- Determine if we already tried recently to refresh this config and
306 # -- failed.
307 refresh_failure_timestamp = cashed_config_metadata.get(
308 "refresh-failure-on", ""
309 )
310 minutes_since_refresh_failure = minutes_between_datetime_stamps(
311 refresh_failure_timestamp, datetime_stamp_now, default=99999
312 )
313 refresh_failed_recently = (
314 0
315 <= minutes_since_refresh_failure
316 < self.remote_config_retry_minutes
317 )
319 # -- Dump info for debugging.
320 if util.is_debug(1): 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 cout(
322 f"{days_since_last_fetch=}, {time_valid=}, {url_changed=}",
323 f"{minutes_since_refresh_failure=}, "
324 f"{refresh_failed_recently=}",
325 style=EMPH3,
326 )
328 # -- Fetch the new config if needed.
329 if url_changed or not time_valid: 329 ↛ 330line 329 didn't jump to line 330 because the condition on line 329 was never true
330 if not refresh_failed_recently:
331 self._fetch_and_update_remote_config(error_is_fatal=False)
333 @property
334 def remote_config(self) -> Dict:
335 """Returns the remote config that is applicable for this invocation.
336 Should not called if the context was initialized with NO_CONFIG."""
337 return self._cached_remote_config
339 def add_package(self, name: str, version: str, platform_id: str, url: str):
340 """Add a package to the profile class"""
342 # -- Updated the installed package data.
343 self.installed_packages[name] = {
344 "version": version,
345 "platform": platform_id,
346 "loaded-by": util.get_apio_version_str(),
347 "loaded-at": get_datetime_stamp(),
348 "loaded-from": url,
349 }
350 # self._save()
351 self._save_installed_packages()
353 def set_preferences_theme(self, theme: str):
354 """Set prefer theme name."""
355 self.preferences["theme"] = theme
356 self._save()
357 self.apply_color_preferences()
359 def remove_package(self, name: str):
360 """Remove a package from the profile file"""
362 if name in self.installed_packages.keys(): 362 ↛ exitline 362 didn't return from function 'remove_package' because the condition on line 362 was always true
363 del self.installed_packages[name]
364 # self._save()
365 self._save_installed_packages()
367 @staticmethod
368 def apply_color_preferences():
369 """Apply currently preferred theme."""
370 # -- Make sure the console is configured, with the default theme,
371 # -- before reading the preferences. Reading the preferences resolves
372 # -- the apio home dir which may exit with a console error message,
373 # -- for example if the home dir path contains a space.
374 apio_console.configure()
376 # -- If not specified, read the theme from file.
377 theme: str = Profile.read_preferences_theme()
379 # -- Apply to the apio console.
380 apio_console.configure(theme_name=theme)
382 @staticmethod
383 def read_preferences_theme(*, default: str = "light") -> str:
384 """Returns the value of the theme preference or default if not
385 specified. This is a static method because we may need this value
386 before creating the profile object, for example when printing command
387 help.
388 """
390 profile_path = util.resolve_home_dir() / "profile.json"
392 if not profile_path.exists():
393 return default
395 try:
396 with open(profile_path, "r", encoding="utf8") as f:
397 # -- Get the colors preferences value, if exists.
398 data = json.load(f)
399 preferences = data.get("preferences", {})
400 theme = preferences.get("theme", default)
401 except (OSError, ValueError, AttributeError):
402 # -- A corrupt profile file. Not reporting it here since
403 # -- _load_profile_file() reports it with a proper error message.
404 return default
406 # -- Fall back to the default for unknown theme names or values,
407 # -- e.g. from a hand edited or old profile file, since
408 # -- apio_console.configure() accepts only known theme names.
409 if not isinstance(theme, str) or theme not in THEMES_TABLE:
410 return default
412 return theme
414 def get_installed_package_info(self, package_name: str) -> Tuple[str, str]:
415 """Return (package_version, platform_id) of the given installed
416 package. Values are replaced with "" if not installed or a value is
417 missing."""
418 package_info = self.installed_packages.get(package_name, {})
419 package_version = package_info.get("version", "")
420 platform_id = package_info.get("platform", "")
421 return (package_version, platform_id)
423 def get_package_config(
424 self,
425 package_name: str,
426 ) -> PackageRemoteConfig:
427 """Given a package name, return the remote config information with the
428 version and fetch information.
429 """
431 # -- Extract package's remote config.
432 package_config = self.remote_config["packages"][package_name]
433 repo_name = package_config["repository"]["name"]
434 repo_organization = package_config["repository"]["organization"]
435 release_tag = package_config["release"]["tag"]
436 release_version = release_tag.replace("-", ".")
437 release_file = package_config["release"]["package"]
439 return PackageRemoteConfig(
440 repo_name=repo_name,
441 repo_organization=repo_organization,
442 release_version=release_version,
443 release_tag=release_tag,
444 release_file=release_file,
445 )
447 def _load_profile_file(self):
448 """Load the profile file if exists, e.g.
449 /home/obijuan/.apio/profile.json)
450 """
452 # -- If profile file doesn't exist then nothing to do.
453 if not self._profile_path.exists():
454 return
456 # -- Read the profile file as a json dict and extract its fields.
457 # -- Handle invalid content gracefully, e.g. a corrupt or hand
458 # -- edited file, since this runs on every apio command.
459 try:
460 with open(self._profile_path, "r", encoding="utf8") as f:
461 data = json.load(f)
462 _check_json_dict(data, "the file content")
464 # -- Determine if the cached remote config is usable.
465 remote_config = data.get("remote-config", {})
466 config_apio_version = remote_config.get("metadata", {}).get(
467 "loaded-by", ""
468 )
469 config_usable = config_apio_version == util.get_apio_version_str()
471 # -- Extract the fields. If remote config is of a different
472 # -- apio version, drop it.
473 self.preferences = data.get("preferences", {})
474 self.installed_packages = data.get("installed-packages", {})
475 self._cached_remote_config = remote_config if config_usable else {}
477 # -- Validate the shape of the extracted fields.
478 _check_json_dict(self.preferences, "'preferences'")
479 _check_json_dict(self.installed_packages, "'installed-packages'")
480 for name, info in self.installed_packages.items():
481 _check_json_dict(info, f"package '{name}'")
482 except (OSError, ValueError, AttributeError) as e:
483 cerror(f"Invalid profile file {self._profile_path}", f"{e}")
484 cout(
485 "You can delete the file and let apio recreate it.",
486 style=INFO,
487 )
488 sys.exit(1)
490 def _load_installed_packages_file(self):
491 """Load the installed packages index file if exists, e.g.
492 /home/obijuan/.apio/packages/installed_packages.json)
493 """
495 if self._packages_index_path.exists():
497 # -- Read the file as a json dict. Handle invalid content
498 # -- gracefully, since this runs on every apio command.
499 try:
500 with open(
501 self._packages_index_path, "r", encoding="utf8"
502 ) as f:
503 self.installed_packages = json.load(f)
504 _check_json_dict(self.installed_packages, "the file content")
505 for name, info in self.installed_packages.items():
506 _check_json_dict(info, f"package '{name}'")
507 except (OSError, ValueError) as e:
508 cerror(
509 f"Invalid packages index file "
510 f"{self._packages_index_path}",
511 f"{e}",
512 )
513 cout(
514 "You can delete the file and let apio recreate it.",
515 style=INFO,
516 )
517 sys.exit(1)
519 def _save(self):
520 """Save the profile file"""
522 # -- Create the enclosing folder, if it does not exist yet
523 path = self._profile_path.parent
524 if not path.exists(): 524 ↛ 525line 524 didn't jump to line 525 because the condition on line 524 was never true
525 path.mkdir()
527 # -- Construct the json dict.
528 data = {}
529 if self.preferences:
530 data["preferences"] = self.preferences
532 if self._cached_remote_config: 532 ↛ 536line 532 didn't jump to line 536 because the condition on line 532 was always true
533 data["remote-config"] = self._cached_remote_config
535 # -- Write to profile file.
536 with open(self._profile_path, "w", encoding="utf8") as f:
537 json.dump(data, f, indent=4)
539 # -- Dump for debugging.
540 if util.is_debug(1): 540 ↛ 541line 540 didn't jump to line 541 because the condition on line 540 was never true
541 cout("Saved profile:", style=EMPH3)
542 cout(json.dumps(data, indent=2))
544 def _save_installed_packages(self):
545 """Save the installed packages file"""
547 # -- Create the enclosing folder, if it does not exist yet
548 path = self._packages_index_path.parent
549 if not path.exists(): 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 path.mkdir()
552 # -- Write to profile file.
553 with open(self._packages_index_path, "w", encoding="utf8") as f:
554 json.dump(self.installed_packages, f, indent=4)
556 # -- Dump for debugging.
557 if util.is_debug(1): 557 ↛ 558line 557 didn't jump to line 558 because the condition on line 557 was never true
558 cout("Saved installed packages index:", style=EMPH3)
559 cout(json.dumps(self.installed_packages, indent=2))
561 def _handle_config_refresh_failure(
562 self, *, msg: List[str], error_is_fatal: bool
563 ):
564 """Called to handle a failure of a remote config refresh."""
565 # -- Handle hard error.
566 if error_is_fatal:
567 cout(*msg, style=ERROR)
568 sys.exit(1)
570 # -- Handle soft error. We can continue with a cached config.
571 # -- Sanity check, a cached config exists.
572 assert self._cached_remote_config, "No cached remote config"
574 # -- Print the soft warning.
575 cout(*msg, style=INFO)
576 cout("Will try again at a latter time.", style=INFO)
578 # -- Memorize the time of the attempt so we don't retry too often.
579 metadata = self._cached_remote_config["metadata"]
580 metadata["refresh-failure-on"] = get_datetime_stamp()
581 self._save()
583 def _fetch_and_update_remote_config(self, *, error_is_fatal: bool) -> None:
584 """Returns the apio remote config JSON dict."""
586 # -- Fetch the config text. Returns None if error_is_fatal=False and
587 # -- fetch failed.
588 config_text: Optional[str] = self._fetch_remote_config_text(
589 error_is_fatal=error_is_fatal
590 )
592 if config_text is None: 592 ↛ 595line 592 didn't jump to line 595 because the condition on line 592 was never true
593 # -- Sanity check, If error_is_fatal, _fetch_remote_config_text()
594 # -- wouldn't return with None.
595 assert not error_is_fatal
596 return
598 # -- Print the file's content for debugging
599 if util.is_debug(1): 599 ↛ 600line 599 didn't jump to line 600 because the condition on line 599 was never true
600 cout(config_text)
602 # -- Convert the jsonc to json by removing '//' comments.
603 config_text = jsonc.to_json(config_text)
605 # -- Parse the remote JSON config file into a dict.
606 try:
607 remote_config = json.loads(config_text)
609 # -- Handle parsing error.
610 except json.decoder.JSONDecodeError as exc:
611 self._handle_config_refresh_failure(
612 msg=[
613 "Failed to parse the latest Apio remote config file.",
614 f"{exc}",
615 ],
616 error_is_fatal=error_is_fatal,
617 )
618 return
620 # -- Do some checks and fail if invalid. This is not an exhaustive
621 # -- check.
622 ok = self._check_downloaded_remote_config(
623 remote_config, error_is_fatal=error_is_fatal
624 )
625 if not ok: 625 ↛ 626line 625 didn't jump to line 626 because the condition on line 625 was never true
626 return
628 # -- Append remote config metadata. This also clear the
629 # -- "refresh-failure-on" field if exists.
630 metadata_dict = {}
631 metadata_dict["loaded-by"] = util.get_apio_version_str()
632 metadata_dict["loaded-at"] = get_datetime_stamp()
633 metadata_dict["loaded-from"] = self.remote_config_url
634 remote_config["metadata"] = metadata_dict
636 self._cached_remote_config = remote_config
637 self._save()
639 def _check_downloaded_remote_config(
640 self, remote_config: Dict, error_is_fatal: bool
641 ) -> bool:
642 """Check the downloaded remote config has a valid structure."""
643 try:
644 validate(instance=remote_config, schema=REMOTE_CONFIG_SCHEMA)
645 except ValidationError as e:
646 # -- Error.
647 msg = ["Fetched remote config failed validation.", str(e)]
648 self._handle_config_refresh_failure(
649 msg=msg, error_is_fatal=error_is_fatal
650 )
651 return False
653 # -- Ok.
654 return True
656 def _fetch_remote_config_text(self, error_is_fatal: bool) -> Optional[str]:
657 """Fetches and returns the apio remote config JSON text. In case
658 of an error, returns None."""
660 # pylint: disable=broad-exception-caught
662 # -- Announce the remote config url
663 cout(f"Fetching '{self.remote_config_url}'")
665 # -- If the URL has a file protocol, read from the file. This
666 # -- is used mostly for testing of a new package version.
667 if self.remote_config_url.startswith("file://"): 667 ↛ 689line 667 didn't jump to line 689 because the condition on line 667 was always true
668 file_path = self.remote_config_url[7:]
669 try:
670 with open(file_path, encoding="utf-8") as f:
671 file_text = f.read()
672 except Exception as e:
673 # -- Since local config file can be fixed and doesn't depend
674 # -- on availability of a remote server, we make this a fatal
675 # -- error instead of a soft error.
676 self._handle_config_refresh_failure(
677 msg=["Failed to read a local config file.", str(e)],
678 error_is_fatal=True,
679 )
681 # -- Local file read OK.
682 return file_text
684 # -- Here is the normal case where the config url is not of a local
685 # -- file but at a remote URL.
687 # -- Fetch the remote config. With timeout = 10, this failed a
688 # -- few times on github workflow tests so increased to 25.
689 try:
690 resp: requests.Response = requests.get(
691 self.remote_config_url, timeout=25
692 )
693 error_msg = None
694 except Exception as e:
695 error_msg = str(e)
697 # -- Error codes such as 404 don't cause an exception so we handle
698 # -- them here separately.
699 if (error_msg is None) and (resp.status_code != 200):
700 error_msg = (
701 f"Expected HTTP status code 200, got {resp.status_code}."
702 )
704 # -- If an error was found then handle it.
705 if error_msg is not None:
706 self._handle_config_refresh_failure(
707 msg=[
708 "Downloading of the latest Apio remote config "
709 "file failed.",
710 error_msg,
711 ],
712 error_is_fatal=error_is_fatal,
713 )
714 return None
716 # -- Done ok.
717 assert resp.text is not None
718 return resp.text