diff --git a/conda_lock/conda_lock.py b/conda_lock/conda_lock.py index f1643261..5016b303 100644 --- a/conda_lock/conda_lock.py +++ b/conda_lock/conda_lock.py @@ -325,6 +325,11 @@ def make_lock_files( # noqa: C901 platform_overrides=platform_overrides, required_categories=required_categories if filter_categories else None, ) + from conda_lock.models.lock_spec import render_pixi_toml + + pixi_toml = render_pixi_toml(lock_spec=lock_spec) + print("\n".join(pixi_toml)) + sys.exit(0) # Load existing lockfile if it exists original_lock_content: Optional[Lockfile] = None @@ -1619,6 +1624,7 @@ def render( # bail out if we do not encounter the lockfile lock_file = pathlib.Path(lock_file) if not lock_file.exists(): + print(f"ERROR: Lockfile {lock_file} does not exist.\n\n", file=sys.stderr) print(ctx.get_help()) sys.exit(1) diff --git a/conda_lock/models/lock_spec.py b/conda_lock/models/lock_spec.py index db69960d..299702ce 100644 --- a/conda_lock/models/lock_spec.py +++ b/conda_lock/models/lock_spec.py @@ -2,8 +2,10 @@ import json import pathlib import typing +import warnings -from typing import Dict, List, Optional, Union +from collections import defaultdict +from typing import Any, Dict, List, NamedTuple, Optional, Set, Union from pydantic import BaseModel, Field, validator from typing_extensions import Literal @@ -120,3 +122,206 @@ def validate_pip_repositories( if isinstance(repository, str): value[index] = PipRepository.from_string(repository) return typing.cast(List[PipRepository], value) + + +class DepKey1(NamedTuple): + name: str + category: str + platform: str + manager: str + + +class DepKey2(NamedTuple): + name: str + category: str + manager: str + + +class DepWithPlatform(NamedTuple): + dep: Dependency + platform: str + + +class DepWithSubdir(NamedTuple): + dep: Dependency + subdir: Optional[str] + + +def render_pixi_toml( # noqa: C901 + *, + lock_spec: LockSpecification, + project_name: str = "project-name-placeholder", +) -> List[str]: + all_platforms = lock_spec.dependencies.keys() + + all_categories: Set[str] = set() + for platform in all_platforms: + for dep in lock_spec.dependencies[platform]: + all_categories.add(dep.category) + if {"main", "default"} <= all_categories: + raise ValueError("Cannot have both 'main' and 'default' as categories/extras") + + indexed_deps: Dict[DepKey1, Dependency] = {} + for platform in all_platforms: + deps = lock_spec.dependencies[platform] + for dep in deps: + category = dep.category + key1 = DepKey1( + name=dep.name, category=category, platform=platform, manager=dep.manager + ) + if key1 in indexed_deps: + raise ValueError( + f"Duplicate dependency {key1}: {dep}, {indexed_deps[key1]}" + ) + indexed_deps[key1] = dep + + # Collect by platform + aggregated_deps: Dict[DepKey2, List[DepWithPlatform]] = defaultdict(list) + for key1, dep in indexed_deps.items(): + key2 = DepKey2(name=key1.name, category=key1.category, manager=key1.manager) + aggregated_deps[key2].append(DepWithPlatform(dep=dep, platform=key1.platform)) + + # Reduce by platform + reduced_deps: Dict[DepKey2, DepWithSubdir] = {} + for key2, deps_with_platforms in aggregated_deps.items(): + for curr, next in zip(deps_with_platforms, deps_with_platforms[1:]): + if curr.dep != next.dep: + raise ValueError(f"Conflicting dependencies {curr} and {next}") + dep = deps_with_platforms[0].dep + dep_platforms = {dep.platform for dep in deps_with_platforms} + if len(dep_platforms) < len(all_platforms): + if len(dep_platforms) != 1: + raise ValueError( + f"Dependency {dep} is specified for more than one platform but not all: {dep_platforms=}, {all_platforms=}" + ) + reduced_deps[key2] = DepWithSubdir(dep=dep, subdir=dep_platforms.pop()) + elif len(dep_platforms) == len(all_platforms): + reduced_deps[key2] = DepWithSubdir(dep=dep, subdir=None) + else: + raise RuntimeError(f"Impossible: {dep_platforms=}, {all_platforms=}") + + result: List[str] = [ + "# This file was generated by conda-lock for the pixi environment manager.", + "# For more information, see ." "# Source files:", + ] + for source in lock_spec.sources: + result.append(f"# - {source}") + result.extend( + [ + "", + "[project]", + f'name = "{project_name}"', + f"platforms = {list(all_platforms)}", + ] + ) + channels: List[str] = [channel.url for channel in lock_spec.channels] + for channel in lock_spec.channels: + if channel.used_env_vars: + warnings.warn( + f"Channel {channel.url} uses environment variables " + "which are not implemented" + ) + result.append(f"channels = {channels}") + result.append("") + + sorted_categories = sorted(all_categories) + if "main" in sorted_categories: + sorted_categories.remove("main") + sorted_categories.insert(0, "main") + if "default" in sorted_categories: + sorted_categories.remove("default") + sorted_categories.insert(0, "default") + + for category in sorted_categories: + feature = category if category != "main" else "default" + + conda_deps: Dict[str, DepWithSubdir] = {} + pip_deps: Dict[str, Dependency] = {} + for key2, dep_with_subdir in sorted( + reduced_deps.items(), key=lambda x: x[0].name + ): + if key2.category == category: + name = key2.name + manager = dep_with_subdir.dep.manager + if manager == "conda": + conda_deps[name] = dep_with_subdir + elif manager == "pip": + if dep_with_subdir.subdir is not None: + raise ValueError( + f"Subdir specified for pip dependency {dep_with_subdir}" + ) + else: + pip_deps[name] = dep_with_subdir.dep + else: + raise ValueError(f"Unknown manager {manager}") + + if conda_deps: + if feature == "default": + result.append("[dependencies]") + else: + result.append(f"[feature.{feature}.dependencies]") + for name, dep_with_subdir in conda_deps.items(): + pixi_spec = make_pixi_conda_spec(dep_with_subdir) + result.append(f"{name} = {pixi_spec}") + result.append("") + + if pip_deps: + if feature == "default": + result.append("[pypi-dependencies]") + else: + result.append(f"[feature.{feature}.pypi-dependencies]") + for name, dep in pip_deps.items(): + pixi_spec = make_pixi_pip_spec(dep) + result.append(f"{name} = {pixi_spec}") + result.append("") + return result + + +def make_pixi_conda_spec(dep_with_subdir: DepWithSubdir) -> str: + dep = dep_with_subdir.dep + subdir = dep_with_subdir.subdir + matchspec = {} + if dep.extras: + warnings.warn(f"Extras not supported in Conda dep {dep}") + if isinstance(dep, VersionedDependency): + matchspec["version"] = dep.version or "*" + if subdir is not None: + matchspec["subdir"] = subdir + if dep.build is not None: + matchspec["build"] = dep.build + if dep.hash is not None: + raise NotImplementedError(f"Hash not yet supported in {dep}") + if dep.conda_channel is not None: + matchspec["channel"] = dep.conda_channel + if len(matchspec) == 1: + return f'"{matchspec["version"]}"' + else: + return json.dumps(matchspec) + elif isinstance(dep, URLDependency): + raise NotImplementedError(f"URL not yet supported in {dep}") + elif isinstance(dep, VCSDependency): + raise NotImplementedError(f"VCS not yet supported in {dep}") + else: + raise ValueError(f"Unknown dependency type {dep}") + + +def make_pixi_pip_spec(dep: Dependency) -> str: + matchspec: Dict[str, Any] = {} + if isinstance(dep, VersionedDependency): + matchspec["version"] = dep.version or "*" + if dep.hash is not None: + raise NotImplementedError(f"Hash not yet supported in {dep}") + if dep.conda_channel is not None: + matchspec["channel"] = dep.conda_channel + if dep.extras: + matchspec["extras"] = dep.extras + if len(matchspec) == 1: + return f'"{matchspec["version"]}"' + else: + return json.dumps(matchspec) + elif isinstance(dep, URLDependency): + raise NotImplementedError(f"URL not yet supported in {dep}") + elif isinstance(dep, VCSDependency): + raise NotImplementedError(f"VCS not yet supported in {dep}") + else: + raise ValueError(f"Unknown dependency type {dep}")