chore: enable default pyrefly checking

pixi.toml

@@ -20,4 +20,4 @@ vinca.path = "."
 
 [package.build]
 backend.name = "pixi-build-python"
-config.ignore-pypi-mapping = false
+config.ignore-pypi-mapping = false

pyproject.toml

@@ -67,5 +67,6 @@ include = [
 packages = ["vinca"]
 
 [tool.pyrefly]
-preset = "basic"
+preset = "default"
 project-includes = ["vinca"]
+project-excludes = ["vinca/test_*.py"]

vinca/config.py

@@ -1,6 +1,8 @@
-selected_platform = None
-ros_distro = None
-skip_testing = None
-parsed_args = None
-setup_pixi_version = None
-pixi_version = None
+from argparse import Namespace
+
+selected_platform: str | None = None
+ros_distro: str | None = None
+skip_testing: bool | None = None
+parsed_args: Namespace | None = None
+setup_pixi_version: str | None = None
+pixi_version: str | None = None

vinca/distro.py

@@ -237,9 +237,11 @@ def _get_direct_depends(self, pkg: str) -> set[str]:
  self._direct_depends_cache[pkg] = set(direct)
  return direct
 
- def _get_snapshot_recursive_depends(self, pkg, ignore_pkgs=None):
+ def _get_snapshot_recursive_depends(
+ self, pkg: str, ignore_pkgs: Optional[Iterable[str]] = None
+ ) -> set[str]:
  """Return ROS dependencies using only package manifests pinned by the snapshot."""
- dependencies = set()
+ dependencies: set[str] = set()
  ignored = set(ignore_pkgs or [])
  packages_to_check = {pkg}
  checked_packages = set()
@@ -263,7 +265,7 @@ def _get_snapshot_recursive_depends(self, pkg, ignore_pkgs=None):
  package_xml = self.get_release_package_xml(package_name)
  package = catkin_pkg.package.parse_package_string(package_xml)
  package.evaluate_conditions(os.environ)
- direct_dependencies = {
+ direct_dependencies: set[str] = {
  dependency.name
  for attribute in dependency_attributes
  for dependency in getattr(package, attribute)

vinca/generate_gha.py

@@ -97,10 +97,8 @@ def add_additional_recipes(args):
  os.path.join(args.dir, "..", "additional_recipes")
  )
 
- print("Searching additional recipes in ", additional_recipes_path)
-
  if not os.path.exists(additional_recipes_path):
- return
+ return []
 
  with open("vinca.yaml", "r") as vinca_yaml:
  vinca_conf = yaml.safe_load(vinca_yaml)
@@ -222,7 +220,7 @@ def build_unix_pipeline(
  setup_pixi_version: str = DEFAULT_SETUP_PIXI_VERSION,
  pixi_version: str = DEFAULT_PIXI_VERSION,
 ):
- blurb = {"jobs": {}, "name": pipeline_name}
+ blurb: dict[str, Any] = {"jobs": {}, "name": pipeline_name}
 
  if workflow is None:
  workflow = blurb
@@ -343,7 +341,7 @@ def build_win_pipeline(
 ):
  vm_imagename = "windows-2022"
  # Build Win pipeline
- blurb = {"jobs": {}, "name": "build_win"}
+ blurb: dict[str, Any] = {"jobs": {}, "name": "build_win"}
 
  if workflow is None:
  workflow = blurb
@@ -423,7 +421,10 @@ def build_win_pipeline(
 
 
 def get_full_tree():
- recipes_dir = config.parsed_args.dir
+ parsed_args = config.parsed_args
+ if parsed_args is None:
+ raise RuntimeError("Pipeline arguments must be parsed before generating a tree")
+ recipes_dir = parsed_args.dir
 
  vinca_yaml = os.path.join(os.path.dirname(recipes_dir), "vinca.yaml")
 
@@ -499,7 +500,11 @@ def main():
 
  names_to_build = {pkg["package"]["name"] for pkg in metas}
  print("Names to build: ", names_to_build)
- tg_slimmed = [el for el in tg if el in names_to_build]
+ tg_slimmed = [
+ element
+ for element in tg
+ if isinstance(element, str) and element in names_to_build
+ ]
 
  stages = []
  current_stage = []

vinca/generate_gitlab.py

@@ -1,6 +1,7 @@
 import glob
 import os
 import sys
+from typing import Any
 
 import networkx as nx
 import yaml
@@ -79,7 +80,7 @@ def main():
 
  print(stages)
 
- gitlab_template = {"image": "condaforge/linux-anvil-cos7-x86_64"}
+ gitlab_template: dict[str, Any] = {"image": "condaforge/linux-anvil-cos7-x86_64"}
 
  stage_names = []
  for i, s in enumerate(stages):

vinca/main.py

@@ -263,6 +263,8 @@ def get_pkg(pkg_name):
  print(f"Could not generate output for {pkg_shortname}")
  continue
 
+ output: dict[str, Any] | None = None
+
  try:
  output = generate_output(
  pkg_shortname,
@@ -692,7 +694,7 @@ def main():
  ):
  with open(add_rec) as fi:
  add_rec_y = yaml.load(fi)
- if config.parsed_args.platform == "emscripten-wasm32":
+ if arguments.platform == "emscripten-wasm32":
  additional_recipe_names.add(add_rec_y["package"]["name"])
  else:
  if add_rec_y["package"]["name"] not in [

vinca/migrate.py

@@ -78,22 +78,24 @@ def create_migration_instructions(arch, packages_to_migrate, trigger_branch):
 
  latest = {}
  for pkg in ros_pkgs:
- current = current_version = None
+ current = None
+ current_version: tuple[int, ...] | None = None
  for pkey in packages:
  if packages[pkey]["name"] == pkg:
-tmp = packages[pkey]["version"].split(".")
+parts = packages[pkey]["version"].split(".")
  version = []
- for el in tmp:
- if el.isdecimal():
- version.append(int(el))
+ for element in parts:
+ if element.isdecimal():
+ version.append(int(element))
  else:
- x = re.search(r"[^0-9]", version).start()
- version.append(int(el[:x]))
+ match = re.search(r"[^0-9]", element)
+ if match is not None:
+ version.append(int(element[: match.start()]))
 
-version = tuple(version)
+parsed_version = tuple(version)
 
- if not current or version > current_version:
- current_version = version
+ if current_version is None or parsed_version > current_version:
+ current_version = parsed_version
  current = pkey
  latest[pkg] = current
 
@@ -132,22 +134,24 @@ def create_migration_instructions(arch, packages_to_migrate, trigger_branch):
  if os.path.exists("recipes"):
  shutil.rmtree("recipes")
 
- mutex_path = os.path.join(
- config.parsed_args.dir, "additional_recipes/ros-distro-mutex"
- )
- if os.path.exists(mutex_path):
- goal_folder = os.path.join(
- config.parsed_args.dir, "recipes", "ros-distro-mutex"
+ parsed_args = config.parsed_args
+ if parsed_args is None:
+ raise RuntimeError(
+ "Migration arguments must be parsed before generating instructions"
  )
+
+ mutex_path = os.path.join(parsed_args.dir, "additional_recipes/ros-distro-mutex")
+ if os.path.exists(mutex_path):
+ goal_folder = os.path.join(parsed_args.dir, "recipes", "ros-distro-mutex")
  os.makedirs(goal_folder, exist_ok=True)
  copy_tree(mutex_path, goal_folder)
 
  subprocess.check_call(
- ["vinca", "-d", config.parsed_args.dir, "--multiple", "--platform", arch]
+ ["vinca", "-d", parsed_args.dir, "--multiple", "--platform", arch]
  )
 
  # TODO remove hard coded build branch here!
- recipe_dir = os.path.join(config.parsed_args.dir, "recipes")
+ recipe_dir = os.path.join(parsed_args.dir, "recipes")
  subprocess.check_call(
  [
  "vinca-gha",

vinca/mutex.py

@@ -59,6 +59,8 @@ def get_mutex_package_dependency(
  raise ValueError(
  f"Error parsing mutex_package configuration: {error}"
  ) from error
+ if mutex_config is None:
+ return None
 
  version_parts = mutex_config["version"].split(".")
  pin_depth = len(mutex_config["upper_bound"].split("."))

vinca/pinning.py

@@ -19,7 +19,7 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Iterator, Mapping, Optional, Sequence, Union
+from typing import Any, Iterator, Literal, Mapping, Optional, Sequence, Union, cast
 from urllib.parse import quote
 
 import requests
@@ -127,7 +127,9 @@ def get_pinning_distribution(version: str) -> str:
  return url
 
 
-def _read_tar_members(fileobj: Any, *, mode: str) -> dict[str, bytes]:
+def _read_tar_members(
+ fileobj: Any, *, mode: Literal["r|", "r:bz2"]
+) -> dict[str, bytes]:
  """Extract only the base config and migration YAML files from a package tarball."""
  files = {}
  with tarfile.open(fileobj=fileobj, mode=mode) as archive:
@@ -290,7 +292,7 @@ def render_pinning(
  else:
  base_payload, migration_payloads = package
 
- rendered = yaml.load(base_payload.decode("utf-8")) or {}
+ rendered: Any = yaml.load(base_payload.decode("utf-8")) or {}
  migration_names = {_migration_name(name): name for name in selected_migrations}
  missing = sorted(set(migration_names) - set(migration_payloads))
  if missing:
@@ -305,7 +307,7 @@ def render_pinning(
  for migration in ordered_migrations:
  migration_config = yaml.load(migration_payloads[migration].decode("utf-8"))
  try:
- rendered = variant_add(rendered, migration_config or {})
+ rendered = cast(Any, variant_add(rendered, migration_config or {}))
  except (KeyError, RuntimeError, TypeError, ValueError) as exc:
  raise PinningError(f"Could not apply migration {migration}: {exc}") from exc
  _validate_zipped_overrides(rendered, overrides)
@@ -416,6 +418,8 @@ def dependencies_from_vinca(
  raise PinningError(
  "Platform selectors must not change the ROS distro snapshots"
  )
+ if group_packages is None:
+ raise PinningError("No group packages were generated")
  vinca_config["_selected_pkgs"] = get_selected_packages(
  distro, vinca_config
  )
@@ -501,8 +505,9 @@ def _migration_selectors(data: Any) -> Iterator[Optional[str]]:
  key_comment = data.ca.items.get(key, [None, None, None])[2]
  key_selector = _comment_selector(key_comment)
  if isinstance(value, list):
+ value_with_comments: Any = value
  for index in range(len(value)):
- item_comment = value.ca.items.get(index, [None])[0]
+ item_comment = value_with_comments.ca.items.get(index, [None])[0]
  item_selector = _comment_selector(item_comment)
  if key_selector and item_selector:
  yield f"({key_selector}) and ({item_selector})"
@@ -525,6 +530,7 @@ def _migration_applies_to_platforms(payload: bytes, platforms: Sequence[str]) ->
  return any(
  _eval_condition(selector, _platform_flags(platform))
  for selector in selectors
+ if selector is not None
  for platform in platforms
  )
 

vinca/resolve.py

@@ -32,10 +32,15 @@ def get_conda_index(vinca_conf, base_dir):
 def resolve_pkgname_from_indexes(pkg_shortname, conda_index):
  for i in conda_index:
  if pkg_shortname in i:
- sys_platform = map_platform_python_to_conda[config.selected_platform]
+ selected_platform = config.selected_platform
+ if selected_platform is None:
+ raise RuntimeError(
+ "A target platform is required to resolve package names"
+ )
+ sys_platform = map_platform_python_to_conda[selected_platform]
  if "robostack" in i[pkg_shortname].keys():
- if config.selected_platform in i[pkg_shortname]["robostack"]:
- return i[pkg_shortname]["robostack"][config.selected_platform]
+ if selected_platform in i[pkg_shortname]["robostack"]:
+ return i[pkg_shortname]["robostack"][selected_platform]
  elif sys_platform in i[pkg_shortname]["robostack"]:
  return i[pkg_shortname]["robostack"][sys_platform]
  else:

vinca/snapshot.py

@@ -60,8 +60,10 @@ def main():
  deps = distro.get_depends(args.package)
  deps.add(args.package)
 
+ max_len = 0
+
  if not args.quiet:
- max_len = max([len(dep) for dep in deps])
+ max_len = max(len(dep) for dep in deps)
  print("\033[1m{0:{2}} {1}\033[0m".format("Package", "Version", max_len + 2))
 
  output = {}

vinca/sources.py

@@ -89,7 +89,7 @@ def generate_source(
 
  url, ref, ref_type = distro.get_released_repo(shortname)
  package_name = package_names[0]
- entry = source_reference(url=url, ref=ref, ref_type=ref_type)
+ entry: dict[str, Any] = source_reference(url=url, ref=ref, ref_type=ref_type)
  entry["target_directory"] = f"{package_name}/src/work"
 
  patches = _package_patches(package_name, vinca_conf, platform)
@@ -127,7 +127,7 @@ def generate_source_version(
 
  url, ref, ref_type = distro.get_released_repo(shortname)
  package_name = package_names[0]
- entry = source_reference(url=url, ref=ref, ref_type=ref_type)
+ entry: dict[str, Any] = source_reference(url=url, ref=ref, ref_type=ref_type)
  entry["target_directory"] = f"{package_name}/src/work"
  if patches := _package_patches(package_name, vinca_conf, platform):
  entry["patches"] = patches
@@ -150,7 +150,7 @@ def generate_fat_source(
  continue
  url, ref, ref_type = distro.get_released_repo(shortname)
  package_name = package_names[0]
- entry = source_reference(url=url, ref=ref, ref_type=ref_type)
+ entry: dict[str, Any] = source_reference(url=url, ref=ref, ref_type=ref_type)
  entry["target_directory"] = f"src/{package_name}"
 
  patch_path = Path(vinca_conf["_patch_dir"]) / f"{package_name}.patch"

vinca/test_archive_sources.py

@@ -1,6 +1,7 @@
 import io
 import tarfile
 import zipfile
+from typing import Any
 
 import pytest
 import requests
@@ -216,7 +217,7 @@ def test_get_forces_no_credentials_of_its_own(monkeypatch):
  """
  distro = Distro.__new__(Distro)
  monkeypatch.delenv("GITHUB_TOKEN", raising=False)
- calls = {}
+ calls: dict[str, Any] = {}
 
  class FakeResponse:
  content = b"payload"