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