import logging import os import ap_git import json import jsonschema import hashlib from pathlib import Path from threading import Lock from utils import TaskRunner from .vehicles_manager import VehiclesManager as vehm class VersionInfo: """ Class to wrap version info properties. """ def __init__(self, remote_info: 'RemoteInfo', commit_ref: str, release_type: str, version_number: str, ap_build_artifacts_url) -> None: self.remote_info = remote_info self.commit_ref = commit_ref self.release_type = release_type self.version_number = version_number self.ap_build_artifacts_url = ap_build_artifacts_url # Generate version_id as remote-sanitized_commit_ref-hash # Commit ref is sanitized for URL safety by replacing '/' with '-' # Hash is used to ensure unique ID after sanitization of commit_ref, # as different commit refs may become identical after replacing '/' commit_ref_sanitized = commit_ref.replace('/', '-') commit_ref_hash = hashlib.md5(commit_ref.encode()).hexdigest()[:8] self.version_id = ( f"{remote_info.name}-{commit_ref_sanitized}-{commit_ref_hash}" ) class RemoteInfo: """ Class to wrap remote info properties. """ def __init__(self, name: str, url: str) -> None: self.name = name self.url = url def to_dict(self): return { 'name': self.name, 'url': self.url, } class VersionsFetcher: """ Class to fetch the version-to-build metadata from remotes.json and provide methods to view the same """ __singleton = None def __init__(self, remotes_json_path: str, ap_repo: ap_git.GitRepo): """ Initializes the VersionsFetcher instance with a given remotes.json path. Parameters: remotes_json_path (str): Path to the remotes.json file. ap_repo (GitRepo): ArduPilot local git repository. This local repository is shared between the VersionsFetcher and the APSourceMetadataFetcher. Raises: RuntimeError: If an instance of this class already exists, enforcing a singleton pattern. """ if vehm.get_singleton() is None: raise RuntimeError("VehiclesManager should be initialised first") # Enforce singleton pattern by raising an error if # an instance already exists. if VersionsFetcher.__singleton: raise RuntimeError("VersionsFetcher must be a singleton.") self.logger = logging.getLogger(__name__) self.__remotes_json_path = remotes_json_path self.__ensure_remotes_json() self.__access_lock_versions_metadata = Lock() self.__versions_metadata = [] tasks = ( (self.fetch_ap_releases, 1200), (self.fetch_whitelisted_tags, 1200), ) self.__task__runner = TaskRunner(tasks=tasks) self.repo = ap_repo VersionsFetcher.__singleton = self def start(self) -> None: """ Start auto-fetch jobs. """ self.logger.info( "Starting VersionsFetcher background auto-fetch jobs." ) self.__task__runner.start() def stop(self) -> None: """ Stop auto-fetch jobs. """ self.logger.info( "Stopping VersionsFetcher background auto-fetch jobs." ) self.__task__runner.stop() def get_all_remotes_info(self) -> list[RemoteInfo]: """ Return the list of RemoteInfo objects constructed from the information in the remotes.json file Returns: list: RemoteInfo objects for all remotes mentioned in remotes.json """ return [ RemoteInfo( name=remote.get('name', None), url=remote.get('url', None) ) for remote in self.__get_versions_metadata() ] def get_remote_info(self, remote_name: str) -> RemoteInfo: """ Return the RemoteInfo for the given remote name, None otherwise. Returns: RemoteInfo: The remote information object. """ return next( ( remote for remote in self.get_all_remotes_info() if remote.name == remote_name ), None ) def get_versions_for_vehicle(self, vehicle_id: str) -> list[VersionInfo]: """ Return the list of dictionaries containing the info about the versions listed to be built for a particular vehicle. Parameters: vehicle_id (str): the vehicle ID to fetch versions list for Returns: list: VersionInfo objects for all versions allowed to be built for the said vehicle. """ if vehicle_id is None: raise ValueError("Vehicle ID is a required parameter.") vehicle = vehm.get_singleton().get_vehicle_by_id(vehicle_id) if vehicle is None: raise ValueError(f"Invalid vehicle ID '{vehicle_id}'.") vehicle_name = vehicle.name versions_list = [] for remote in self.__get_versions_metadata(): remote_info = RemoteInfo( name=remote.get('name', None), url=remote.get('url', None) ) for vehicle in remote['vehicles']: if vehicle['name'] != vehicle_name: continue for release in vehicle['releases']: versions_list.append(VersionInfo( remote_info=remote_info, commit_ref=release.get('commit_reference', None), release_type=release.get('release_type', None), version_number=release.get('version_number', None), ap_build_artifacts_url=release.get( 'ap_build_artifacts_url', None ) )) return versions_list def is_version_listed(self, vehicle_id: str, version_id: str) -> bool: """ Check if a version with given properties mentioned in remotes.json Parameters: vehicle_id (str): ID of the vehicle for which version is listed version_id (str): version ID Returns: bool: True if the said version is mentioned in remotes.json, False otherwise """ if vehicle_id is None: raise ValueError("vehicle_id is a required parameter.") if version_id is None: raise ValueError("version_id is a required parameter.") return version_id in [ version_info.version_id for version_info in self.get_versions_for_vehicle(vehicle_id=vehicle_id) ] def get_version_info(self, vehicle_id: str, version_id: str) -> VersionInfo: """ Find first version matching the given properties in remotes.json Parameters: vehicle_id (str): ID of the vehicle for which version is listed version_id (str): version ID Returns: VersionInfo: Object for the version matching the properties, None if not found """ return next( ( version for version in self.get_versions_for_vehicle( vehicle_id=vehicle_id ) if version.version_id == version_id ), None ) def reload_remotes_json(self) -> None: """ Read remotes.json, validate its structure against the schema and cache it in memory """ # load file containing vehicles listed to be built for each # remote along with the branches/tags/commits on which the # firmware can be built remotes_json_schema_path = os.path.join( os.path.dirname(__file__), 'remotes.schema.json' ) with open(self.__remotes_json_path, 'r') as f, \ open(remotes_json_schema_path, 'r') as s: f_content = f.read() # Early return if file is empty if not f_content: return versions_metadata = json.loads(f_content) schema = json.loads(s.read()) # validate schema jsonschema.validate(instance=versions_metadata, schema=schema) self.__set_versions_metadata(versions_metadata=versions_metadata) # update git repo with latest remotes list self.__sync_remotes_with_ap_repo() def __ensure_remotes_json(self) -> None: """ Ensures remotes.json exists and is a valid JSON file. """ p = Path(self.__remotes_json_path) if not p.exists(): # Ensure parent directory exists Path.mkdir(p.parent, parents=True, exist_ok=True) # write empty json list with open(p, 'w') as f: f.write('[]') def __set_versions_metadata(self, versions_metadata: list) -> None: """ Set versions metadata property with the one passed as parameter This requires to acquire the access lock to avoid overwriting the object while it is being read """ if versions_metadata is None: raise ValueError("versions_metadata is a required parameter. " "Cannot be None.") with self.__access_lock_versions_metadata: self.__versions_metadata = versions_metadata def __get_versions_metadata(self) -> list: """ Read versions metadata property This requires to acquire the access lock to avoid reading the list while it is being modified Returns: list: the versions metadata list """ with self.__access_lock_versions_metadata: return self.__versions_metadata def __sync_remotes_with_ap_repo(self): """ Update the remotes in ArduPilot local repository with the latest remotes list. """ remotes = tuple( (remote.name, remote.url) for remote in self.get_all_remotes_info() ) self.repo.remote_add_bulk(remotes=remotes, force=True) def fetch_ap_releases(self) -> None: """ Execute the fetch_releases.py script to update remotes.json with Ardupilot's official releases """ from scripts import fetch_releases fetch_releases.run( base_dir=os.path.join( os.path.dirname(self.__remotes_json_path), '..', ), remote_name="ardupilot", ) self.reload_remotes_json() return def fetch_whitelisted_tags(self) -> None: """ Execute the fetch_whitelisted_tags.py script to update remotes.json with tags from whitelisted repos """ from scripts import fetch_whitelisted_tags fetch_whitelisted_tags.run( base_dir=os.path.join( os.path.dirname(self.__remotes_json_path), '..', ) ) self.reload_remotes_json() return @staticmethod def get_singleton(): return VersionsFetcher.__singleton