Skip to content

[cpackget] enhancement : Python Library for cpackget #560

@s-celles

Description

@s-celles

Hi,

This proposal outlines the importance of creating a Python library wrapper for cpackget to enhance the Open-CMSIS-Pack ecosystem's integration with Python-based development workflows, CI/CD pipelines, and automation tools.

Why a Python Library is Critical

1. Ecosystem Integration

The embedded development landscape increasingly relies on Python for:

  • Build automation and CI/CD pipelines (Jenkins, GitHub Actions, GitLab CI)
  • Test automation frameworks (pytest, Robot Framework)
  • Data analysis and reporting (pandas, matplotlib for test results)
  • Development tooling (SCons, PlatformIO, West/Zephyr build systems)

Currently, projects using Python-based workflows must resort to subprocess calls to interact with cpackget, leading to:

  • Complex error handling and parsing
  • Platform-specific binary management
  • Inconsistent integration patterns
  • Limited access to structured data from cpackget operations

2. Developer Experience Enhancement

A native Python library would provide:

# Instead of this complex subprocess approach:
import subprocess
import json

def install_pack_current():
    try:
        result = subprocess.run(['cpackget', 'add', 'ARM::[email protected]'], 
                              capture_output=True, text=True, check=True)
        # Parse stdout manually, handle errors inconsistently
    except subprocess.CalledProcessError as e:
        # Limited error context, no structured error data
        pass
# Developers could use this clean, Pythonic API:
import cpackget

def install_pack_improved():
    try:
        pack_manager = cpackget.PackManager()
        pack = pack_manager.add_pack("ARM::[email protected]")
        return pack.metadata  # Structured pack information
    except cpackget.PackNotFoundError as e:
        # Rich error context with suggestions
        print(f"Pack not found: {e.suggestions}")

3. Advanced Workflow Support

Python library enables sophisticated use cases:

  • Dynamic pack dependency resolution in complex projects
  • Automated pack validation in CI pipelines with detailed reporting
  • Pack metadata analysis for license compliance and security auditing
  • Integration with package managers like pip, conda, or custom corporate package systems
  • Cross-platform automation without binary distribution concerns

Implementation Approaches

(Beaware it's ai generated.. so it may be buggy)

Option 1: HTTP API Gateway (Recommended for MVP)

Create a lightweight HTTP service wrapper around cpackget with a Python client library.

Go HTTP Service

package main

import (
    "encoding/json"
    "net/http"
    "os/exec"
    "strings"
)

type PackRequest struct {
    Action    string            `json:"action"`
    PackName  string            `json:"pack_name,omitempty"`
    Version   string            `json:"version,omitempty"`
    Options   map[string]string `json:"options,omitempty"`
}

type PackResponse struct {
    Success bool                   `json:"success"`
    Data    map[string]interface{} `json:"data,omitempty"`
    Error   string                 `json:"error,omitempty"`
}

func handlePackOperation(w http.ResponseWriter, r *http.Request) {
    var req PackRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    // Build cpackget command
    args := buildCpackgetArgs(req)
    
    cmd := exec.Command("cpackget", args...)
    output, err := cmd.CombinedOutput()
    
    response := PackResponse{
        Success: err == nil,
        Data:    parseOutput(output, req.Action),
    }
    
    if err != nil {
        response.Error = err.Error()
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(response)
}

func main() {
    http.HandleFunc("/api/pack", handlePackOperation)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

Python Client Library

"""
cpackget: Python wrapper for Open-CMSIS-Pack management
"""

import requests
import json
from typing import List, Dict, Optional, Union
from dataclasses import dataclass
from enum import Enum

class PackState(Enum):
    INSTALLED = "installed"
    CACHED = "cached"
    AVAILABLE = "available"
    UNKNOWN = "unknown"

@dataclass
class PackInfo:
    name: str
    version: str
    vendor: str
    state: PackState
    description: Optional[str] = None
    license: Optional[str] = None
    url: Optional[str] = None
    dependencies: List[str] = None

class CpackgetError(Exception):
    """Base exception for cpackget operations"""
    pass

class PackNotFoundError(CpackgetError):
    """Pack not found in repositories"""
    def __init__(self, pack_name: str, suggestions: List[str] = None):
        self.pack_name = pack_name
        self.suggestions = suggestions or []
        super().__init__(f"Pack '{pack_name}' not found")

class PackManager:
    """Main interface for CMSIS pack management"""
    
    def __init__(self, pack_root: str = None, api_base: str = "http://localhost:8080"):
        self.pack_root = pack_root
        self.api_base = api_base
        self._session = requests.Session()
    
    def add_pack(self, pack_spec: str, agree_license: bool = False) -> PackInfo:
        """
        Install a CMSIS pack
        
        Args:
            pack_spec: Pack specification (e.g., "ARM::[email protected]" or "ARM::CMSIS")
            agree_license: Automatically agree to embedded licenses
            
        Returns:
            PackInfo object with installation details
            
        Raises:
            PackNotFoundError: If pack is not found
            CpackgetError: For other installation errors
        """
        payload = {
            "action": "add",
            "pack_name": pack_spec,
            "options": {
                "agree_license": str(agree_license).lower(),
                "pack_root": self.pack_root
            }
        }
        
        response = self._make_request(payload)
        if not response["success"]:
            if "not found" in response.get("error", "").lower():
                raise PackNotFoundError(pack_spec, self._get_suggestions(pack_spec))
            raise CpackgetError(response.get("error", "Unknown error"))
        
        return self._parse_pack_info(response["data"])
    
    def remove_pack(self, pack_spec: str, purge: bool = False) -> bool:
        """
        Remove a CMSIS pack
        
        Args:
            pack_spec: Pack specification to remove
            purge: Also remove cached files
            
        Returns:
            True if removal was successful
        """
        payload = {
            "action": "remove",
            "pack_name": pack_spec,
            "options": {
                "purge": str(purge).lower(),
                "pack_root": self.pack_root
            }
        }
        
        response = self._make_request(payload)
        return response["success"]
    
    def list_packs(self, include_cached: bool = False, include_public: bool = False) -> List[PackInfo]:
        """
        List installed packs
        
        Args:
            include_cached: Include cached (downloaded) packs
            include_public: Include packs available in public index
            
        Returns:
            List of PackInfo objects
        """
        payload = {
            "action": "list",
            "options": {
                "cached": str(include_cached).lower(),
                "public": str(include_public).lower(),
                "pack_root": self.pack_root
            }
        }
        
        response = self._make_request(payload)
        if not response["success"]:
            raise CpackgetError(response.get("error", "Failed to list packs"))
        
        return [self._parse_pack_info(pack_data) for pack_data in response["data"]["packs"]]
    
    def update_index(self, sparse: bool = False) -> bool:
        """
        Update the pack index
        
        Args:
            sparse: Only update index.pidx, skip PDSC files
            
        Returns:
            True if update was successful
        """
        payload = {
            "action": "update_index",
            "options": {
                "sparse": str(sparse).lower(),
                "pack_root": self.pack_root
            }
        }
        
        response = self._make_request(payload)
        return response["success"]
    
    def verify_pack(self, pack_path: str, public_key: str = None) -> Dict[str, bool]:
        """
        Verify pack signature and checksum
        
        Args:
            pack_path: Path to pack file
            public_key: Optional public key for signature verification
            
        Returns:
            Dictionary with verification results
        """
        payload = {
            "action": "verify",
            "pack_name": pack_path,
            "options": {
                "public_key": public_key,
                "pack_root": self.pack_root
            }
        }
        
        response = self._make_request(payload)
        if not response["success"]:
            raise CpackgetError(response.get("error", "Verification failed"))
        
        return response["data"]["verification_results"]
    
    def search_packs(self, query: str) -> List[PackInfo]:
        """
        Search for packs in the public index
        
        Args:
            query: Search term (vendor, pack name, or description)
            
        Returns:
            List of matching PackInfo objects
        """
        # This would need to be implemented to parse the public index
        # and provide search functionality
        pass
    
    def get_pack_dependencies(self, pack_spec: str) -> List[str]:
        """
        Get dependencies for a specific pack
        
        Args:
            pack_spec: Pack specification
            
        Returns:
            List of dependency pack specifications
        """
        # This would parse the PDSC file to extract dependencies
        pass
    
    def _make_request(self, payload: Dict) -> Dict:
        """Make HTTP request to cpackget service"""
        try:
            response = self._session.post(
                f"{self.api_base}/api/pack",
                json=payload,
                timeout=60
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            raise CpackgetError(f"API request failed: {e}")
    
    def _parse_pack_info(self, data: Dict) -> PackInfo:
        """Parse pack information from API response"""
        return PackInfo(
            name=data.get("name", ""),
            version=data.get("version", ""),
            vendor=data.get("vendor", ""),
            state=PackState(data.get("state", "unknown")),
            description=data.get("description"),
            license=data.get("license"),
            url=data.get("url"),
            dependencies=data.get("dependencies", [])
        )
    
    def _get_suggestions(self, pack_name: str) -> List[str]:
        """Get pack name suggestions for failed searches"""
        # This could implement fuzzy matching against the public index
        return []

# Convenience functions for common operations
def install_pack(pack_spec: str, pack_root: str = None) -> PackInfo:
    """Quick pack installation"""
    manager = PackManager(pack_root=pack_root)
    return manager.add_pack(pack_spec)

def list_installed_packs(pack_root: str = None) -> List[PackInfo]:
    """Quick installed pack listing"""
    manager = PackManager(pack_root=pack_root)
    return manager.list_packs()

# CLI integration for familiar interface
def main():
    """CLI interface that mirrors cpackget commands"""
    import argparse
    
    parser = argparse.ArgumentParser(description="Python wrapper for cpackget")
    subparsers = parser.add_subparsers(dest="command", help="Available commands")
    
    # Add command
    add_parser = subparsers.add_parser("add", help="Add a pack")
    add_parser.add_argument("pack_spec", help="Pack specification")
    add_parser.add_argument("--agree-license", action="store_true")
    
    # List command
    list_parser = subparsers.add_parser("list", help="List packs")
    list_parser.add_argument("--cached", action="store_true")
    list_parser.add_argument("--public", action="store_true")
    
    # Parse and execute
    args = parser.parse_args()
    manager = PackManager()
    
    if args.command == "add":
        pack = manager.add_pack(args.pack_spec, args.agree_license)
        print(f"Installed: {pack.vendor}::{pack.name}@{pack.version}")
    elif args.command == "list":
        packs = manager.list_packs(args.cached, args.public)
        for pack in packs:
            print(f"{pack.vendor}::{pack.name}@{pack.version} [{pack.state.value}]")

if __name__ == "__main__":
    main()

Option 2: Shared Library Approach

Compile cpackget functionality into a shared library for maximum performance.

// cpackget_lib.go
package main

import "C"
import (
    "encoding/json"
    "github.com/your-org/cpackget/internal/packmanager"
)

//export AddPack
func AddPack(packSpec *C.char, options *C.char) *C.char {
    goPackSpec := C.GoString(packSpec)
    goOptions := C.GoString(options)
    
    var opts map[string]interface{}
    json.Unmarshal([]byte(goOptions), &opts)
    
    result, err := packmanager.AddPack(goPackSpec, opts)
    if err != nil {
        return C.CString(fmt.Sprintf(`{"success": false, "error": "%s"}`, err.Error()))
    }
    
    resultJSON, _ := json.Marshal(map[string]interface{}{
        "success": true,
        "data": result,
    })
    
    return C.CString(string(resultJSON))
}

//export ListPacks
func ListPacks(options *C.char) *C.char {
    // Implementation for listing packs
}

func main() {} // Required for shared library
# Python ctypes wrapper
import ctypes
import json
import os
from typing import Dict, Any

class CpackgetNative:
    def __init__(self):
        # Load shared library
        lib_path = self._find_library()
        self._lib = ctypes.CDLL(lib_path)
        
        # Configure function signatures
        self._lib.AddPack.argtypes = (ctypes.c_char_p, ctypes.c_char_p)
        self._lib.AddPack.restype = ctypes.c_char_p
        
        self._lib.ListPacks.argtypes = (ctypes.c_char_p,)
        self._lib.ListPacks.restype = ctypes.c_char_p
    
    def add_pack(self, pack_spec: str, **options) -> Dict[str, Any]:
        options_json = json.dumps(options)
        result_ptr = self._lib.AddPack(
            pack_spec.encode('utf-8'),
            options_json.encode('utf-8')
        )
        result_json = ctypes.string_at(result_ptr).decode('utf-8')
        return json.loads(result_json)
    
    def list_packs(self, **options) -> Dict[str, Any]:
        options_json = json.dumps(options)
        result_ptr = self._lib.ListPacks(options_json.encode('utf-8'))
        result_json = ctypes.string_at(result_ptr).decode('utf-8')
        return json.loads(result_json)
    
    def _find_library(self) -> str:
        # Library discovery logic
        possible_paths = [
            "./libcpackget.so",
            "/usr/local/lib/libcpackget.so",
            os.path.expanduser("~/.local/lib/libcpackget.so")
        ]
        
        for path in possible_paths:
            if os.path.exists(path):
                return path
        
        raise RuntimeError("libcpackget shared library not found")

Option 3: gRPC Service

For high-performance, strongly-typed communication.

// cpackget.proto
syntax = "proto3";
package cpackget;

service PackService {
    rpc AddPack(AddPackRequest) returns (AddPackResponse);
    rpc RemovePack(RemovePackRequest) returns (RemovePackResponse);
    rpc ListPacks(ListPacksRequest) returns (ListPacksResponse);
    rpc UpdateIndex(UpdateIndexRequest) returns (UpdateIndexResponse);
    rpc VerifyPack(VerifyPackRequest) returns (VerifyPackResponse);
}

message AddPackRequest {
    string pack_spec = 1;
    bool agree_license = 2;
    string pack_root = 3;
    uint32 timeout = 4;
}

message AddPackResponse {
    bool success = 1;
    string error = 2;
    PackInfo pack_info = 3;
}

message PackInfo {
    string name = 1;
    string version = 2;
    string vendor = 3;
    string state = 4;
    string description = 5;
    string license = 6;
    repeated string dependencies = 7;
}

Conclusion

A Python library for cpackget represents a critical infrastructure investment for the Open-CMSIS-Pack ecosystem but it would greatly help my project platformio/platformio-core#704 (comment)

Metadata

Metadata

Assignees

No one assigned

    Labels

    DiscussionIssue to be discussed before further action is takenenhancementNew feature or request

    Type

    Projects

    Status

    Backlog

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions