Coverage for apio / profile.py: 75%
239 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-24 01:53 +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(
113 ts1: str, ts2: str, default: Any
114) -> Optional[int]:
115 """Given two values generated by get_datetime_stamp(), return the
116 number of days from ts1 to ts2. The value can be negative if ts2 is
117 earlier than ts1. Returns the given 'default' value if either timestamp
118 is invalid."""
120 # -- The parsing format.
121 fmt = "%Y-%m-%d-%H-%M"
123 # -- Convert to timedates
124 try:
125 datetime1 = datetime.strptime(ts1, fmt)
126 datetime2 = datetime.strptime(ts2, fmt)
127 except ValueError:
128 return default
130 # -- Round to beginning of day.
131 day1 = datetime(datetime1.year, datetime1.month, datetime1.day)
132 day2 = datetime(datetime2.year, datetime2.month, datetime2.day)
134 # -- Compute the diff in days.
135 delta_days: int = (day2 - day1).days
137 # -- All done.
138 assert isinstance(delta_days, int)
139 return delta_days
142def minutes_between_datetime_stamps(
143 ts1: str, ts2: str, default: Any
144) -> Optional[int]:
145 """Return the number of minutes from ts1 to ts2 or default if any of
146 the timestamps is invalid."""
148 # -- The parsing format.
149 fmt = "%Y-%m-%d-%H-%M"
151 # -- Convert to timedates
152 try:
153 datetime1 = datetime.strptime(ts1, fmt)
154 datetime2 = datetime.strptime(ts2, fmt)
155 except ValueError:
156 return default
158 # -- Calculate the diff in minutes.
159 delta_minutes = int((datetime2 - datetime1).total_seconds() / 60)
160 assert isinstance(delta_minutes, int), type(delta_minutes)
161 return delta_minutes
164class Profile:
165 """Class for managing the apio profile file
166 ex. ~/.apio/profile.json
167 """
169 # pylint: disable=too-many-instance-attributes
171 # -- Only these instance vars are allowed.
172 __slots__ = (
173 "_profile_path",
174 "_packages_index_path",
175 "remote_config_url",
176 "remote_config_ttl_days",
177 "remote_config_retry_minutes",
178 "_remote_config_policy",
179 "_cached_remote_config",
180 "preferences",
181 "installed_packages",
182 )
184 def __init__(
185 self,
186 home_dir: Path,
187 packages_dir: Path,
188 remote_config_url_template: str,
189 remote_config_ttl_days: int,
190 remote_config_retry_minutes: int,
191 remote_config_policy: RemoteConfigPolicy,
192 ):
193 """remote_config_url_template is a url string with the
194 placeholder {major} and {minor} for the apio's major and minor
195 version. '"""
197 # pylint: disable=too-many-arguments
198 # pylint: disable=too-many-positional-arguments
200 # -- Sanity check
201 assert isinstance(remote_config_ttl_days, int)
202 assert 0 < remote_config_ttl_days <= 30
204 # -- Sanity check
205 assert isinstance(remote_config_retry_minutes, int)
206 assert 0 < remote_config_retry_minutes <= (60 * 24)
208 # -- Resolve and cache the remote config url. Replaced the placeholders
209 # -- with the major and minor versions of apio. Path version is
210 # -- not used.
211 ver_tuple = util.get_apio_version_tuple()
212 url = remote_config_url_template
213 url = url.replace("{major}", str(ver_tuple[0]))
214 url = url.replace("{minor}", str(ver_tuple[1]))
215 self.remote_config_url = url
217 # -- Save remote url ttl setting.
218 self.remote_config_ttl_days = remote_config_ttl_days
220 # -- Save the remote config fetch retry minutes.
221 self.remote_config_retry_minutes = remote_config_retry_minutes
223 # -- Save remote config policy.
224 self._remote_config_policy = remote_config_policy
226 # -- Verify that we resolved all the placeholders.
227 assert "{" not in self.remote_config_url, self.remote_config_url
229 if util.is_debug(1): 229 ↛ 230line 229 didn't jump to line 230 because the condition on line 229 was never true
230 cout(f"Remote config url: {self.remote_config_url}")
232 # ---- Set the default parameters
234 # User preferences
235 self.preferences = {}
237 # -- Installed package versions
238 self.installed_packages = {}
240 # -- A copy of remote config.
241 self._cached_remote_config = {}
243 # -- Cache the profile file path
244 # -- Ex. '/home/obijuan/.apio/profile.json'
245 self._profile_path = home_dir / "profile.json"
247 # -- Cache the packages index file path
248 # -- Ex. '/home/obijuan/.apio/packages/installed_packages.json'
249 self._packages_index_path = packages_dir / "installed_packages.json"
251 # -- Read the profile from file, if exists.
252 self._load_profile_file()
254 # -- Read the installed packages file, if exists.
255 self._load_installed_packages_file()
257 # -- Apply config policy
258 self._apply_remote_config_policy()
260 def _apply_remote_config_policy(self) -> None:
261 """Called after loading the profile file, to apply the remote config
262 policy for this invocation."""
264 # -- Case 1: A fresh config is required for the current command.
265 if self._remote_config_policy == RemoteConfigPolicy.GET_FRESH:
266 self._fetch_and_update_remote_config(error_is_fatal=True)
267 return
269 # -- Case 2: A fresh config is optional but there is no cached
270 # -- config so practically it's required.
271 assert self._remote_config_policy == RemoteConfigPolicy.CACHED_OK
272 if not self._cached_remote_config:
273 if util.is_debug(1): 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 cout("Saved remote config is not available.", style=INFO)
275 self._fetch_and_update_remote_config(error_is_fatal=True)
276 return
278 # -- Case 3: May need to fetch a new config but can continue with
279 # -- the cached config in case of a fetch failure.
280 #
281 # -- Get the cached config metadata.
282 cashed_config_metadata = self._cached_remote_config.get("metadata", {})
283 last_fetch_timestamp = cashed_config_metadata.get("loaded-at", "")
284 last_fetch_url = cashed_config_metadata.get("loaded-from", "")
286 # -- Determine if we need a new config because the remote config URL
287 # -- was changed (e.g. with APIO_REMOTE_CONFIG_URL)
288 url_changed = last_fetch_url != self.remote_config_url
290 # -- Determine if there is time related reason to fetch a new config.
291 datetime_stamp_now = get_datetime_stamp()
292 days_since_last_fetch = days_between_datetime_stamps(
293 last_fetch_timestamp, datetime_stamp_now, default=99999
294 )
295 time_valid = 0 <= days_since_last_fetch < self.remote_config_ttl_days
297 # -- Determine if we already tried recently to refresh this config and
298 # -- failed.
299 refresh_failure_timestamp = cashed_config_metadata.get(
300 "refresh-failure-on", ""
301 )
302 minutes_since_refresh_failure = minutes_between_datetime_stamps(
303 refresh_failure_timestamp, datetime_stamp_now, default=99999
304 )
305 refresh_failed_recently = (
306 0
307 <= minutes_since_refresh_failure
308 < self.remote_config_retry_minutes
309 )
311 # -- Dump info for debugging.
312 if util.is_debug(1): 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true
313 cout(
314 f"{days_since_last_fetch=}, {time_valid=}, {url_changed=}",
315 f"{minutes_since_refresh_failure=}, "
316 f"{refresh_failed_recently=}",
317 style=EMPH3,
318 )
320 # -- Fetch the new config if needed.
321 if url_changed or not time_valid: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 if not refresh_failed_recently:
323 self._fetch_and_update_remote_config(error_is_fatal=False)
325 @property
326 def remote_config(self) -> Dict:
327 """Returns the remote config that is applicable for this invocation.
328 Should not called if the context was initialized with NO_CONFIG."""
329 return self._cached_remote_config
331 def add_package(self, name: str, version: str, platform_id: str, url: str):
332 """Add a package to the profile class"""
334 # -- Updated the installed package data.
335 self.installed_packages[name] = {
336 "version": version,
337 "platform": platform_id,
338 "loaded-by": util.get_apio_version_str(),
339 "loaded-at": get_datetime_stamp(),
340 "loaded-from": url,
341 }
342 # self._save()
343 self._save_installed_packages()
345 def set_preferences_theme(self, theme: str):
346 """Set prefer theme name."""
347 self.preferences["theme"] = theme
348 self._save()
349 self.apply_color_preferences()
351 def remove_package(self, name: str):
352 """Remove a package from the profile file"""
354 if name in self.installed_packages.keys(): 354 ↛ exitline 354 didn't return from function 'remove_package' because the condition on line 354 was always true
355 del self.installed_packages[name]
356 # self._save()
357 self._save_installed_packages()
359 @staticmethod
360 def apply_color_preferences():
361 """Apply currently preferred theme."""
362 # -- If not specified, read the theme from file.
363 theme: str = Profile.read_preferences_theme()
365 # -- Apply to the apio console.
366 apio_console.configure(theme_name=theme)
368 @staticmethod
369 def read_preferences_theme(*, default: str = "light") -> Optional[str]:
370 """Returns the value of the theme preference or default if not
371 specified. This is a static method because we may need this value
372 before creating the profile object, for example when printing command
373 help.
374 """
376 profile_path = util.resolve_home_dir() / "profile.json"
378 if not profile_path.exists():
379 return default
381 with open(profile_path, "r", encoding="utf8") as f:
382 # -- Get the colors preferences value, if exists.
383 data = json.load(f)
384 preferences = data.get("preferences", {})
385 theme = preferences.get("theme", None)
387 # -- Get the click context, if exists.
388 return theme if theme else default
390 def get_installed_package_info(self, package_name: str) -> Tuple[str, str]:
391 """Return (package_version, platform_id) of the given installed
392 package. Values are replaced with "" if not installed or a value is
393 missing."""
394 package_info = self.installed_packages.get(package_name, {})
395 package_version = package_info.get("version", "")
396 platform_id = package_info.get("platform", "")
397 return (package_version, platform_id)
399 def get_package_config(
400 self,
401 package_name: str,
402 ) -> PackageRemoteConfig:
403 """Given a package name, return the remote config information with the
404 version and fetch information.
405 """
407 # -- Extract package's remote config.
408 package_config = self.remote_config["packages"][package_name]
409 repo_name = package_config["repository"]["name"]
410 repo_organization = package_config["repository"]["organization"]
411 release_tag = package_config["release"]["tag"]
412 release_version = release_tag.replace("-", ".")
413 release_file = package_config["release"]["package"]
415 return PackageRemoteConfig(
416 repo_name=repo_name,
417 repo_organization=repo_organization,
418 release_version=release_version,
419 release_tag=release_tag,
420 release_file=release_file,
421 )
423 def _load_profile_file(self):
424 """Load the profile file if exists, e.g.
425 /home/obijuan/.apio/profile.json)
426 """
428 # -- If profile file doesn't exist then nothing to do.
429 if not self._profile_path.exists():
430 return
432 # -- Read the profile file as a json dict.
433 with open(self._profile_path, "r", encoding="utf8") as f:
434 data = json.load(f)
436 # -- Determine if the cached remote config is usable.
437 remote_config = data.get("remote-config", {})
438 config_apio_version = remote_config.get("metadata", {}).get(
439 "loaded-by", ""
440 )
441 config_usable = config_apio_version == util.get_apio_version_str()
443 # -- Extract the fields. If remote config is of a different apio
444 # -- version, drop it.
445 self.preferences = data.get("preferences", {})
446 self.installed_packages = data.get("installed-packages", {})
447 self._cached_remote_config = remote_config if config_usable else {}
449 def _load_installed_packages_file(self):
450 """Load the installed packages index file if exists, e.g.
451 /home/obijuan/.apio/packages/installed_packages.json)
452 """
454 if self._packages_index_path.exists():
456 # -- Read the file as a json dict.
457 with open(self._packages_index_path, "r", encoding="utf8") as f:
458 self.installed_packages = json.load(f)
460 def _save(self):
461 """Save the profile file"""
463 # -- Create the enclosing folder, if it does not exist yet
464 path = self._profile_path.parent
465 if not path.exists(): 465 ↛ 466line 465 didn't jump to line 466 because the condition on line 465 was never true
466 path.mkdir()
468 # -- Construct the json dict.
469 data = {}
470 if self.preferences:
471 data["preferences"] = self.preferences
473 if self._cached_remote_config: 473 ↛ 477line 473 didn't jump to line 477 because the condition on line 473 was always true
474 data["remote-config"] = self._cached_remote_config
476 # -- Write to profile file.
477 with open(self._profile_path, "w", encoding="utf8") as f:
478 json.dump(data, f, indent=4)
480 # -- Dump for debugging.
481 if util.is_debug(1): 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true
482 cout("Saved profile:", style=EMPH3)
483 cout(json.dumps(data, indent=2))
485 def _save_installed_packages(self):
486 """Save the installed packages file"""
488 # -- Create the enclosing folder, if it does not exist yet
489 path = self._packages_index_path.parent
490 if not path.exists(): 490 ↛ 491line 490 didn't jump to line 491 because the condition on line 490 was never true
491 path.mkdir()
493 # -- Write to profile file.
494 with open(self._packages_index_path, "w", encoding="utf8") as f:
495 json.dump(self.installed_packages, f, indent=4)
497 # -- Dump for debugging.
498 if util.is_debug(1): 498 ↛ 499line 498 didn't jump to line 499 because the condition on line 498 was never true
499 cout("Saved installed packages index:", style=EMPH3)
500 cout(json.dumps(self.installed_packages, indent=2))
502 def _handle_config_refresh_failure(
503 self, *, msg: List[str], error_is_fatal: bool
504 ):
505 """Called to handle a failure of a remote config refresh."""
506 # -- Handle hard error.
507 if error_is_fatal:
508 cout(*msg, style=ERROR)
509 sys.exit(1)
511 # -- Handle soft error. We can continue with a cached config.
512 # -- Sanity check, a cached config exists.
513 assert self._cached_remote_config, "No cached remote config"
515 # -- Print the soft warning.
516 cout(*msg, style=INFO)
517 cout("Will try again at a latter time.", style=INFO)
519 # -- Memorize the time of the attempt so we don't retry too often.
520 metadata = self._cached_remote_config["metadata"]
521 metadata["refresh-failure-on"] = get_datetime_stamp()
522 self._save()
524 def _fetch_and_update_remote_config(self, *, error_is_fatal: bool) -> None:
525 """Returns the apio remote config JSON dict."""
527 # -- Fetch the config text. Returns None if error_is_fatal=False and
528 # -- fetch failed.
529 config_text: Optional[str] = self._fetch_remote_config_text(
530 error_is_fatal=error_is_fatal
531 )
533 if config_text is None: 533 ↛ 536line 533 didn't jump to line 536 because the condition on line 533 was never true
534 # -- Sanity check, If error_is_fatal, _fetch_remote_config_text()
535 # -- wouldn't return with None.
536 assert not error_is_fatal
537 return
539 # -- Print the file's content 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(config_text)
543 # -- Convert the jsonc to json by removing '//' comments.
544 config_text = jsonc.to_json(config_text)
546 # -- Parse the remote JSON config file into a dict.
547 try:
548 remote_config = json.loads(config_text)
550 # -- Handle parsing error.
551 except json.decoder.JSONDecodeError as exc:
552 self._handle_config_refresh_failure(
553 msg=[
554 "Failed to parse the latest Apio remote config file.",
555 f"{exc}",
556 ],
557 error_is_fatal=error_is_fatal,
558 )
559 return
561 # -- Do some checks and fail if invalid. This is not an exhaustive
562 # -- check.
563 ok = self._check_downloaded_remote_config(
564 remote_config, error_is_fatal=error_is_fatal
565 )
566 if not ok: 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true
567 return
569 # -- Append remote config metadata. This also clear the
570 # -- "refresh-failure-on" field if exists.
571 metadata_dict = {}
572 metadata_dict["loaded-by"] = util.get_apio_version_str()
573 metadata_dict["loaded-at"] = get_datetime_stamp()
574 metadata_dict["loaded-from"] = self.remote_config_url
575 remote_config["metadata"] = metadata_dict
577 self._cached_remote_config = remote_config
578 self._save()
580 def _check_downloaded_remote_config(
581 self, remote_config: Dict, error_is_fatal: bool
582 ) -> bool:
583 """Check the downloaded remote config has a valid structure."""
584 try:
585 validate(instance=remote_config, schema=REMOTE_CONFIG_SCHEMA)
586 except ValidationError as e:
587 # -- Error.
588 msg = ["Fetched remote config failed validation.", str(e)]
589 self._handle_config_refresh_failure(
590 msg=msg, error_is_fatal=error_is_fatal
591 )
592 return False
594 # -- Ok.
595 return True
597 def _fetch_remote_config_text(self, error_is_fatal: bool) -> Optional[str]:
598 """Fetches and returns the apio remote config JSON text. In case
599 of an error, returns None."""
601 # pylint: disable=broad-exception-caught
603 # -- Announce the remote config url
604 cout(f"Fetching '{self.remote_config_url}'")
606 # -- If the URL has a file protocol, read from the file. This
607 # -- is used mostly for testing of a new package version.
608 if self.remote_config_url.startswith("file://"): 608 ↛ 630line 608 didn't jump to line 630 because the condition on line 608 was always true
609 file_path = self.remote_config_url[7:]
610 try:
611 with open(file_path, encoding="utf-8") as f:
612 file_text = f.read()
613 except Exception as e:
614 # -- Since local config file can be fixed and doesn't depend
615 # -- on availability of a remote server, we make this a fatal
616 # -- error instead of a soft error.
617 self._handle_config_refresh_failure(
618 msg=["Failed to read a local config file.", str(e)],
619 error_is_fatal=True,
620 )
622 # -- Local file read OK.
623 return file_text
625 # -- Here is the normal case where the config url is not of a local
626 # -- file but at a remote URL.
628 # -- Fetch the remote config. With timeout = 10, this failed a
629 # -- few times on github workflow tests so increased to 25.
630 try:
631 resp: requests.Response = requests.get(
632 self.remote_config_url, timeout=25
633 )
634 error_msg = None
635 except Exception as e:
636 error_msg = str(e)
638 # -- Error codes such as 404 don't cause an exception so we handle
639 # -- them here separately.
640 if (error_msg is None) and (resp.status_code != 200):
641 error_msg = (
642 f"Expected HTTP status code 200, got {resp.status_code}."
643 )
645 # -- If an error was found then handle it.
646 if error_msg is not None:
647 self._handle_config_refresh_failure(
648 msg=[
649 "Downloading of the latest Apio remote config "
650 "file failed.",
651 error_msg,
652 ],
653 error_is_fatal=error_is_fatal,
654 )
655 return None
657 # -- Done ok.
658 assert resp.text is not None
659 return resp.text