Coverage for apio/profile.py: 75%
239 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-24 03:51 +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
20from apio.common.apio_styles import INFO, EMPH3, ERROR
21from apio.utils import util, jsonc
23# -- JSON schema for validating a remote config file.
24REMOTE_CONFIG_SCHEMA = {
25 "$schema": "https://json-schema.org/draft/2020-12/schema",
26 "type": "object",
27 "required": ["packages"],
28 "properties": {
29 # -- Packages
30 "packages": {
31 "type": "object",
32 "patternProperties": {
33 "^.*$": {
34 "type": "object",
35 "required": ["repository", "release"],
36 "properties": {
37 # -- Repository
38 "repository": {
39 "type": "object",
40 "required": ["organization", "name"],
41 "properties": {
42 # -- Repo organization. e.g. "fpgawars"
43 "organization": {"type": "string"},
44 # -- Repo name, e.g. 'examples'
45 "name": {"type": "string"},
46 },
47 "additionalProperties": False,
48 },
49 # -- Release.
50 "release": {
51 "type": "object",
52 "required": [
53 "tag",
54 "package",
55 ],
56 "properties": {
57 # -- Tag
58 "tag": {
59 "type": "string",
60 "pattern": r"^\d{4}\-\d{2}\-\d{2}$",
61 },
62 # -- Package
63 "package": {"type": "string"},
64 },
65 "additionalProperties": False,
66 },
67 },
68 "additionalProperties": False,
69 }
70 },
71 "additionalProperties": False,
72 }
73 },
74 "additionalProperties": False,
75}
78class RemoteConfigPolicy(Enum):
79 """Represents possible requirements from the remote config."""
81 # -- Config is being used but can be a cached value, as long that it's
82 # -- not too old.
83 CACHED_OK = 1
84 # -- Config is being used and a fresh copy is that was fetch in this
85 # -- invocation of Apio is required.
86 GET_FRESH = 2
89@dataclass(frozen=True)
90class PackageRemoteConfig:
91 """Contains a package info from the remote config."""
93 # -- E.g. "tools-oss-cad-suite"
94 repo_name: str
95 # -- E.g. "FPGAwars"
96 repo_organization: str
97 # -- E.g. "0.2.3"
98 release_version: str
99 # -- E.g. "${YYYY-MM-DD}""
100 release_tag: str
101 # -- E.g. "apio-oss-cad-suite-${PLATFORM}-${YYYYMMDD}.zip"
102 release_file: str
105def get_datetime_stamp(dt: Optional[datetime] = None) -> str:
106 """Returns a string with time now as yyyy-mm-dd-hh-mm"""
107 if dt is None:
108 dt = datetime.now()
109 return dt.strftime("%Y-%m-%d-%H-%M")
112def days_between_datetime_stamps(ts1: str, ts2: str, default: Any) -> int:
113 """Given two values generated by get_datetime_stamp(), return the
114 number of days from ts1 to ts2. The value can be negative if ts2 is
115 earlier than ts1. Returns the given 'default' value if either timestamp
116 is invalid."""
118 # -- The parsing format.
119 fmt = "%Y-%m-%d-%H-%M"
121 # -- Convert to timedates
122 try:
123 datetime1 = datetime.strptime(ts1, fmt)
124 datetime2 = datetime.strptime(ts2, fmt)
125 except ValueError:
126 return default
128 # -- Round to beginning of day.
129 day1 = datetime(datetime1.year, datetime1.month, datetime1.day)
130 day2 = datetime(datetime2.year, datetime2.month, datetime2.day)
132 # -- Compute the diff in days.
133 delta_days: int = (day2 - day1).days
135 # -- All done.
136 assert isinstance(delta_days, int)
137 return delta_days
140def minutes_between_datetime_stamps(ts1: str, ts2: str, default: Any) -> int:
141 """Return the number of minutes from ts1 to ts2 or default if any of
142 the timestamps is invalid."""
144 # -- The parsing format.
145 fmt = "%Y-%m-%d-%H-%M"
147 # -- Convert to timedates
148 try:
149 datetime1 = datetime.strptime(ts1, fmt)
150 datetime2 = datetime.strptime(ts2, fmt)
151 except ValueError:
152 return default
154 # -- Calculate the diff in minutes.
155 delta_minutes = int((datetime2 - datetime1).total_seconds() / 60)
156 assert isinstance(delta_minutes, int), type(delta_minutes)
157 return delta_minutes
160class Profile:
161 """Class for managing the apio profile file
162 ex. ~/.apio/profile.json
163 """
165 # pylint: disable=too-many-instance-attributes
167 # -- Only these instance vars are allowed.
168 __slots__ = (
169 "_profile_path",
170 "_packages_index_path",
171 "remote_config_url",
172 "remote_config_ttl_days",
173 "remote_config_retry_minutes",
174 "_remote_config_policy",
175 "_cached_remote_config",
176 "preferences",
177 "installed_packages",
178 )
180 def __init__(
181 self,
182 home_dir: Path,
183 packages_dir: Path,
184 remote_config_url_template: str,
185 remote_config_ttl_days: int,
186 remote_config_retry_minutes: int,
187 remote_config_policy: RemoteConfigPolicy,
188 ):
189 """remote_config_url_template is a url string with the
190 placeholder {major} and {minor} for the apio's major and minor
191 version. '"""
193 # pylint: disable=too-many-arguments
194 # pylint: disable=too-many-positional-arguments
196 # -- Sanity check
197 assert isinstance(remote_config_ttl_days, int)
198 assert 0 < remote_config_ttl_days <= 30
200 # -- Sanity check
201 assert isinstance(remote_config_retry_minutes, int)
202 assert 0 < remote_config_retry_minutes <= (60 * 24)
204 # -- Resolve and cache the remote config url. Replaced the placeholders
205 # -- with the major and minor versions of apio. Path version is
206 # -- not used.
207 ver_tuple = util.get_apio_version_tuple()
208 url = remote_config_url_template
209 url = url.replace("{major}", str(ver_tuple[0]))
210 url = url.replace("{minor}", str(ver_tuple[1]))
211 self.remote_config_url = url
213 # -- Save remote url ttl setting.
214 self.remote_config_ttl_days = remote_config_ttl_days
216 # -- Save the remote config fetch retry minutes.
217 self.remote_config_retry_minutes = remote_config_retry_minutes
219 # -- Save remote config policy.
220 self._remote_config_policy = remote_config_policy
222 # -- Verify that we resolved all the placeholders.
223 assert "{" not in self.remote_config_url, self.remote_config_url
225 if util.is_debug(1): 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true
226 cout(f"Remote config url: {self.remote_config_url}")
228 # ---- Set the default parameters
230 # User preferences
231 self.preferences = {}
233 # -- Installed package versions
234 self.installed_packages = {}
236 # -- A copy of remote config.
237 self._cached_remote_config = {}
239 # -- Cache the profile file path
240 # -- Ex. '/home/obijuan/.apio/profile.json'
241 self._profile_path = home_dir / "profile.json"
243 # -- Cache the packages index file path
244 # -- Ex. '/home/obijuan/.apio/packages/installed_packages.json'
245 self._packages_index_path = packages_dir / "installed_packages.json"
247 # -- Read the profile from file, if exists.
248 self._load_profile_file()
250 # -- Read the installed packages file, if exists.
251 self._load_installed_packages_file()
253 # -- Apply config policy
254 self._apply_remote_config_policy()
256 def _apply_remote_config_policy(self) -> None:
257 """Called after loading the profile file, to apply the remote config
258 policy for this invocation."""
260 # -- Case 1: A fresh config is required for the current command.
261 if self._remote_config_policy == RemoteConfigPolicy.GET_FRESH:
262 self._fetch_and_update_remote_config(error_is_fatal=True)
263 return
265 # -- Case 2: A fresh config is optional but there is no cached
266 # -- config so practically it's required.
267 assert self._remote_config_policy == RemoteConfigPolicy.CACHED_OK
268 if not self._cached_remote_config:
269 if util.is_debug(1): 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 cout("Saved remote config is not available.", style=INFO)
271 self._fetch_and_update_remote_config(error_is_fatal=True)
272 return
274 # -- Case 3: May need to fetch a new config but can continue with
275 # -- the cached config in case of a fetch failure.
276 #
277 # -- Get the cached config metadata.
278 cashed_config_metadata = self._cached_remote_config.get("metadata", {})
279 last_fetch_timestamp = cashed_config_metadata.get("loaded-at", "")
280 last_fetch_url = cashed_config_metadata.get("loaded-from", "")
282 # -- Determine if we need a new config because the remote config URL
283 # -- was changed (e.g. with APIO_REMOTE_CONFIG_URL)
284 url_changed = last_fetch_url != self.remote_config_url
286 # -- Determine if there is time related reason to fetch a new config.
287 datetime_stamp_now = get_datetime_stamp()
288 days_since_last_fetch = days_between_datetime_stamps(
289 last_fetch_timestamp, datetime_stamp_now, default=99999
290 )
291 time_valid = 0 <= days_since_last_fetch < self.remote_config_ttl_days
293 # -- Determine if we already tried recently to refresh this config and
294 # -- failed.
295 refresh_failure_timestamp = cashed_config_metadata.get(
296 "refresh-failure-on", ""
297 )
298 minutes_since_refresh_failure = minutes_between_datetime_stamps(
299 refresh_failure_timestamp, datetime_stamp_now, default=99999
300 )
301 refresh_failed_recently = (
302 0
303 <= minutes_since_refresh_failure
304 < self.remote_config_retry_minutes
305 )
307 # -- Dump info for debugging.
308 if util.is_debug(1): 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 cout(
310 f"{days_since_last_fetch=}, {time_valid=}, {url_changed=}",
311 f"{minutes_since_refresh_failure=}, "
312 f"{refresh_failed_recently=}",
313 style=EMPH3,
314 )
316 # -- Fetch the new config if needed.
317 if url_changed or not time_valid: 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true
318 if not refresh_failed_recently:
319 self._fetch_and_update_remote_config(error_is_fatal=False)
321 @property
322 def remote_config(self) -> Dict:
323 """Returns the remote config that is applicable for this invocation.
324 Should not called if the context was initialized with NO_CONFIG."""
325 return self._cached_remote_config
327 def add_package(self, name: str, version: str, platform_id: str, url: str):
328 """Add a package to the profile class"""
330 # -- Updated the installed package data.
331 self.installed_packages[name] = {
332 "version": version,
333 "platform": platform_id,
334 "loaded-by": util.get_apio_version_str(),
335 "loaded-at": get_datetime_stamp(),
336 "loaded-from": url,
337 }
338 # self._save()
339 self._save_installed_packages()
341 def set_preferences_theme(self, theme: str):
342 """Set prefer theme name."""
343 self.preferences["theme"] = theme
344 self._save()
345 self.apply_color_preferences()
347 def remove_package(self, name: str):
348 """Remove a package from the profile file"""
350 if name in self.installed_packages.keys(): 350 ↛ exitline 350 didn't return from function 'remove_package' because the condition on line 350 was always true
351 del self.installed_packages[name]
352 # self._save()
353 self._save_installed_packages()
355 @staticmethod
356 def apply_color_preferences():
357 """Apply currently preferred theme."""
358 # -- If not specified, read the theme from file.
359 theme: str = Profile.read_preferences_theme()
361 # -- Apply to the apio console.
362 apio_console.configure(theme_name=theme)
364 @staticmethod
365 def read_preferences_theme(*, default: str = "light") -> str:
366 """Returns the value of the theme preference or default if not
367 specified. This is a static method because we may need this value
368 before creating the profile object, for example when printing command
369 help.
370 """
372 profile_path = util.resolve_home_dir() / "profile.json"
374 if not profile_path.exists():
375 return default
377 with open(profile_path, "r", encoding="utf8") as f:
378 # -- Get the colors preferences value, if exists.
379 data = json.load(f)
380 preferences = data.get("preferences", {})
381 theme = preferences.get("theme", default)
383 return theme
385 def get_installed_package_info(self, package_name: str) -> Tuple[str, str]:
386 """Return (package_version, platform_id) of the given installed
387 package. Values are replaced with "" if not installed or a value is
388 missing."""
389 package_info = self.installed_packages.get(package_name, {})
390 package_version = package_info.get("version", "")
391 platform_id = package_info.get("platform", "")
392 return (package_version, platform_id)
394 def get_package_config(
395 self,
396 package_name: str,
397 ) -> PackageRemoteConfig:
398 """Given a package name, return the remote config information with the
399 version and fetch information.
400 """
402 # -- Extract package's remote config.
403 package_config = self.remote_config["packages"][package_name]
404 repo_name = package_config["repository"]["name"]
405 repo_organization = package_config["repository"]["organization"]
406 release_tag = package_config["release"]["tag"]
407 release_version = release_tag.replace("-", ".")
408 release_file = package_config["release"]["package"]
410 return PackageRemoteConfig(
411 repo_name=repo_name,
412 repo_organization=repo_organization,
413 release_version=release_version,
414 release_tag=release_tag,
415 release_file=release_file,
416 )
418 def _load_profile_file(self):
419 """Load the profile file if exists, e.g.
420 /home/obijuan/.apio/profile.json)
421 """
423 # -- If profile file doesn't exist then nothing to do.
424 if not self._profile_path.exists():
425 return
427 # -- Read the profile file as a json dict.
428 with open(self._profile_path, "r", encoding="utf8") as f:
429 data = json.load(f)
431 # -- Determine if the cached remote config is usable.
432 remote_config = data.get("remote-config", {})
433 config_apio_version = remote_config.get("metadata", {}).get(
434 "loaded-by", ""
435 )
436 config_usable = config_apio_version == util.get_apio_version_str()
438 # -- Extract the fields. If remote config is of a different apio
439 # -- version, drop it.
440 self.preferences = data.get("preferences", {})
441 self.installed_packages = data.get("installed-packages", {})
442 self._cached_remote_config = remote_config if config_usable else {}
444 def _load_installed_packages_file(self):
445 """Load the installed packages index file if exists, e.g.
446 /home/obijuan/.apio/packages/installed_packages.json)
447 """
449 if self._packages_index_path.exists():
451 # -- Read the file as a json dict.
452 with open(self._packages_index_path, "r", encoding="utf8") as f:
453 self.installed_packages = json.load(f)
455 def _save(self):
456 """Save the profile file"""
458 # -- Create the enclosing folder, if it does not exist yet
459 path = self._profile_path.parent
460 if not path.exists(): 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true
461 path.mkdir()
463 # -- Construct the json dict.
464 data = {}
465 if self.preferences:
466 data["preferences"] = self.preferences
468 if self._cached_remote_config: 468 ↛ 472line 468 didn't jump to line 472 because the condition on line 468 was always true
469 data["remote-config"] = self._cached_remote_config
471 # -- Write to profile file.
472 with open(self._profile_path, "w", encoding="utf8") as f:
473 json.dump(data, f, indent=4)
475 # -- Dump for debugging.
476 if util.is_debug(1): 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true
477 cout("Saved profile:", style=EMPH3)
478 cout(json.dumps(data, indent=2))
480 def _save_installed_packages(self):
481 """Save the installed packages file"""
483 # -- Create the enclosing folder, if it does not exist yet
484 path = self._packages_index_path.parent
485 if not path.exists(): 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true
486 path.mkdir()
488 # -- Write to profile file.
489 with open(self._packages_index_path, "w", encoding="utf8") as f:
490 json.dump(self.installed_packages, f, indent=4)
492 # -- Dump for debugging.
493 if util.is_debug(1): 493 ↛ 494line 493 didn't jump to line 494 because the condition on line 493 was never true
494 cout("Saved installed packages index:", style=EMPH3)
495 cout(json.dumps(self.installed_packages, indent=2))
497 def _handle_config_refresh_failure(
498 self, *, msg: List[str], error_is_fatal: bool
499 ):
500 """Called to handle a failure of a remote config refresh."""
501 # -- Handle hard error.
502 if error_is_fatal:
503 cout(*msg, style=ERROR)
504 sys.exit(1)
506 # -- Handle soft error. We can continue with a cached config.
507 # -- Sanity check, a cached config exists.
508 assert self._cached_remote_config, "No cached remote config"
510 # -- Print the soft warning.
511 cout(*msg, style=INFO)
512 cout("Will try again at a latter time.", style=INFO)
514 # -- Memorize the time of the attempt so we don't retry too often.
515 metadata = self._cached_remote_config["metadata"]
516 metadata["refresh-failure-on"] = get_datetime_stamp()
517 self._save()
519 def _fetch_and_update_remote_config(self, *, error_is_fatal: bool) -> None:
520 """Returns the apio remote config JSON dict."""
522 # -- Fetch the config text. Returns None if error_is_fatal=False and
523 # -- fetch failed.
524 config_text: Optional[str] = self._fetch_remote_config_text(
525 error_is_fatal=error_is_fatal
526 )
528 if config_text is None: 528 ↛ 531line 528 didn't jump to line 531 because the condition on line 528 was never true
529 # -- Sanity check, If error_is_fatal, _fetch_remote_config_text()
530 # -- wouldn't return with None.
531 assert not error_is_fatal
532 return
534 # -- Print the file's content for debugging
535 if util.is_debug(1): 535 ↛ 536line 535 didn't jump to line 536 because the condition on line 535 was never true
536 cout(config_text)
538 # -- Convert the jsonc to json by removing '//' comments.
539 config_text = jsonc.to_json(config_text)
541 # -- Parse the remote JSON config file into a dict.
542 try:
543 remote_config = json.loads(config_text)
545 # -- Handle parsing error.
546 except json.decoder.JSONDecodeError as exc:
547 self._handle_config_refresh_failure(
548 msg=[
549 "Failed to parse the latest Apio remote config file.",
550 f"{exc}",
551 ],
552 error_is_fatal=error_is_fatal,
553 )
554 return
556 # -- Do some checks and fail if invalid. This is not an exhaustive
557 # -- check.
558 ok = self._check_downloaded_remote_config(
559 remote_config, error_is_fatal=error_is_fatal
560 )
561 if not ok: 561 ↛ 562line 561 didn't jump to line 562 because the condition on line 561 was never true
562 return
564 # -- Append remote config metadata. This also clear the
565 # -- "refresh-failure-on" field if exists.
566 metadata_dict = {}
567 metadata_dict["loaded-by"] = util.get_apio_version_str()
568 metadata_dict["loaded-at"] = get_datetime_stamp()
569 metadata_dict["loaded-from"] = self.remote_config_url
570 remote_config["metadata"] = metadata_dict
572 self._cached_remote_config = remote_config
573 self._save()
575 def _check_downloaded_remote_config(
576 self, remote_config: Dict, error_is_fatal: bool
577 ) -> bool:
578 """Check the downloaded remote config has a valid structure."""
579 try:
580 validate(instance=remote_config, schema=REMOTE_CONFIG_SCHEMA)
581 except ValidationError as e:
582 # -- Error.
583 msg = ["Fetched remote config failed validation.", str(e)]
584 self._handle_config_refresh_failure(
585 msg=msg, error_is_fatal=error_is_fatal
586 )
587 return False
589 # -- Ok.
590 return True
592 def _fetch_remote_config_text(self, error_is_fatal: bool) -> Optional[str]:
593 """Fetches and returns the apio remote config JSON text. In case
594 of an error, returns None."""
596 # pylint: disable=broad-exception-caught
598 # -- Announce the remote config url
599 cout(f"Fetching '{self.remote_config_url}'")
601 # -- If the URL has a file protocol, read from the file. This
602 # -- is used mostly for testing of a new package version.
603 if self.remote_config_url.startswith("file://"): 603 ↛ 625line 603 didn't jump to line 625 because the condition on line 603 was always true
604 file_path = self.remote_config_url[7:]
605 try:
606 with open(file_path, encoding="utf-8") as f:
607 file_text = f.read()
608 except Exception as e:
609 # -- Since local config file can be fixed and doesn't depend
610 # -- on availability of a remote server, we make this a fatal
611 # -- error instead of a soft error.
612 self._handle_config_refresh_failure(
613 msg=["Failed to read a local config file.", str(e)],
614 error_is_fatal=True,
615 )
617 # -- Local file read OK.
618 return file_text
620 # -- Here is the normal case where the config url is not of a local
621 # -- file but at a remote URL.
623 # -- Fetch the remote config. With timeout = 10, this failed a
624 # -- few times on github workflow tests so increased to 25.
625 try:
626 resp: requests.Response = requests.get(
627 self.remote_config_url, timeout=25
628 )
629 error_msg = None
630 except Exception as e:
631 error_msg = str(e)
633 # -- Error codes such as 404 don't cause an exception so we handle
634 # -- them here separately.
635 if (error_msg is None) and (resp.status_code != 200):
636 error_msg = (
637 f"Expected HTTP status code 200, got {resp.status_code}."
638 )
640 # -- If an error was found then handle it.
641 if error_msg is not None:
642 self._handle_config_refresh_failure(
643 msg=[
644 "Downloading of the latest Apio remote config "
645 "file failed.",
646 error_msg,
647 ],
648 error_is_fatal=error_is_fatal,
649 )
650 return None
652 # -- Done ok.
653 assert resp.text is not None
654 return resp.text