Skill Adapter

{
  "_comment": "Example JSON output for plugin analysis by PI Pathfinder",
  "query": "Summarize the key findings from the latest IPCC report on climate change and suggest potential mitigation strategies.",
  "selected_plugin": {
    "name": "Research Assistant",
    "description": "A powerful research tool for accessing and summarizing information from various online sources.",
    "capabilities": [
      "Web scraping",
      "Document summarization",
      "Academic paper retrieval",
      "Fact checking",
      "Trend analysis"
    ],
    "reasoning": "This plugin is best suited for the task because it can access the IPCC report online, summarize its findings, and identify relevant mitigation strategies from reputable sources.",
    "confidence_score": 0.95
  },
  "extracted_skills": [
    {
      "skill_name": "Document Summarization",
      "skill_description": "Condenses large documents into concise summaries highlighting key information.",
      "parameters": {
        "document_url": "URL of the IPCC report",
        "summary_length": "medium",
        "focus_areas": ["key findings", "mitigation strategies"]
      },
      "implementation_details": "Utilizes advanced NLP techniques to identify and extract the most important information, including sentence scoring and topic modeling.",
      "success_probability": 0.9
    },
    {
      "skill_name": "Web Scraping",
      "skill_description": "Extracts data from web pages, including text, tables, and images.",
      "parameters": {
        "url": "URL of the IPCC report website",
        "elements_to_extract": ["text", "tables"]
      },
      "implementation_details": "Uses a robust web scraping library to handle various website structures and anti-scraping measures.",
      "success_probability": 0.98
    }
  ],
  "execution_plan": [
    {
      "step": 1,
      "action": "Use Web Scraping to extract the text content from the IPCC report webpage.",
      "plugin_skill": "Web Scraping",
      "expected_outcome": "Successful extraction of the IPCC report text."
    },
    {
      "step": 2,
      "action": "Use Document Summarization to generate a summary of the extracted text, focusing on key findings and mitigation strategies.",
      "plugin_skill": "Document Summarization",
      "expected_outcome": "A concise summary of the IPCC report's key findings and mitigation strategies."
    },
     {
      "step": 3,
      "action": "Return the generated summary to the user.",
      "plugin_skill": null,
      "expected_outcome": "User receives a helpful summary of the IPCC report."
    }
  ],
  "alternative_plugins": [
    {
      "name": "Web Search",
      "description": "Performs web searches and retrieves relevant snippets.",
      "suitability_score": 0.7,
      "reason": "Useful for finding general information, but less effective for in-depth document analysis."
    },
    {
      "name": "Document Reader",
      "description": "Reads and analyzes local documents.",
      "suitability_score": 0.3,
      "reason": "Not applicable as the IPCC report is likely online."
    }
  ],
  "overall_assessment": {
    "success_likelihood": 0.9,
    "potential_issues": [
      "Website may have anti-scraping measures.",
      "Summarization may miss subtle nuances in the report."
    ],
    "recommendations": "Review the generated summary for accuracy and consult the original report for a complete understanding."
  }
}

# plugin_selection_rules.yaml
# Configuration file for PI Pathfinder plugin selection logic.

# Global settings for plugin selection behavior.
global:
  # Default preference for prioritizing plugins: "speed", "accuracy", "cost"
  default_priority: "accuracy"
  # Maximum number of plugins to consider for a given task.
  max_plugins_considered: 10
  # Minimum relevance score for a plugin to be considered. (0.0 - 1.0)
  minimum_relevance_score: 0.2
  # Enable/Disable verbose logging for debugging.
  verbose_logging: false

# Rules for selecting plugins based on keywords and task descriptions.
keyword_rules:
  # Define keywords and their associated plugin preferences.
  # The higher the weight, the more preferred the plugin is for that keyword.
  - keywords: ["image", "generate", "picture", "visual"]
    plugin_preferences:
      "DALL-E": 0.9  # Example: Prefer DALL-E for image generation
      "Stable Diffusion": 0.8
      "example-value_IMAGE_PLUGIN": 0.5
  - keywords: ["translate", "language", "multilingual"]
    plugin_preferences:
      "Google Translate": 0.95
      "DeepL Translator": 0.9
      "example-value_TRANSLATION_PLUGIN": 0.6
  - keywords: ["code", "programming", "algorithm", "debug"]
    plugin_preferences:
      "Code Interpreter": 0.9
      "GitHub Copilot": 0.85
      "example-value_CODE_PLUGIN": 0.7
  - keywords: ["data analysis", "statistics", "spreadsheet"]
    plugin_preferences:
      "Wolfram Alpha": 0.9
      "Excel Online": 0.8
      "example-value_DATA_PLUGIN": 0.6

# Rules for selecting plugins based on their stated capabilities.
capability_rules:
  # Define capability patterns and their associated plugin preferences.
  # These rules are based on the plugin's description and advertised skills.
  - capability_pattern: "Generates realistic images from text prompts."
    plugin_preferences:
      "DALL-E": 1.0
      "Stable Diffusion": 0.9
  - capability_pattern: "Translates text between multiple languages."
    plugin_preferences:
      "Google Translate": 1.0
      "DeepL Translator": 0.95
  - capability_pattern: "Executes code and analyzes data."
    plugin_preferences:
      "Code Interpreter": 1.0
      "Wolfram Alpha": 0.8

# Rules for handling user preferences.
user_preferences:
  # Default user preferences (can be overridden by user-specific settings).
  defaults:
    priority: "accuracy"  # Default priority: "speed", "accuracy", "cost"
    preferred_plugins: [] # List of plugins the user prefers (e.g., ["DALL-E", "Google Translate"])
    excluded_plugins: []  # List of plugins the user wants to avoid (e.g., ["Expensive Plugin"])

  # Example: User-specific preferences (loaded from a user profile, for example).
  user_id_123:
    priority: "speed"
    preferred_plugins: ["Google Translate"]
    excluded_plugins: ["example-value_EXPENSIVE_PLUGIN"]

# Fallback plugin to use if no other plugin matches the criteria.
fallback_plugin: "Web Search"

# Advanced configuration (for expert users only).
advanced:
  # Weighting factors for combining different rule types.
  keyword_weight: 0.6
  capability_weight: 0.4
  user_preference_weight: 0.2

  # Threshold for considering a plugin "suitable" after applying all rules.
  suitability_threshold: 0.7

  # Plugin specific configurations (example).
  plugin_configurations:
    "DALL-E":
      api_key: "example-value_DALL_E_API_KEY"
      image_size: "1024x1024"
    "Google Translate":
      target_language: "en" # Default target language

#!/usr/bin/env python3

"""
Template for adapting skills from one plugin to another.

This module provides a template for adapting skills from a source plugin
to a target plugin. It includes placeholders for input parameters,
output variables, and adaptation logic.

"""

import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def adapt_skill(source_plugin_skills, target_plugin_requirements, user_query):
    """
    Adapt skills from the source plugin to meet the target plugin's requirements.

    Args:
        source_plugin_skills (dict): A dictionary representing the skills
                                      provided by the source plugin.
        target_plugin_requirements (dict): A dictionary representing the
                                            requirements of the target plugin.
        user_query (str): The original user query.

    Returns:
        dict: A dictionary containing the adapted input parameters for the
              target plugin.  Returns None if adaptation is not possible.

    Raises:
        TypeError: If input types are incorrect.
        ValueError: If input values are invalid.
        Exception: For any other unexpected error during adaptation.
    """

    try:
        if not isinstance(source_plugin_skills, dict):
            raise TypeError("source_plugin_skills must be a dictionary.")
        if not isinstance(target_plugin_requirements, dict):
            raise TypeError("target_plugin_requirements must be a dictionary.")
        if not isinstance(user_query, str):
            raise TypeError("user_query must be a string.")

        # Example adaptation logic (replace with your actual adaptation)
        adapted_input = {}

        # Check if the target plugin requires a 'text' input and adapt from user query
        if "text" in target_plugin_requirements:
            adapted_input["text"] = user_query

        # Check if the source plugin can provide a 'summary' and the target plugin requires it
        if "summary" in target_plugin_requirements and "summarize" in source_plugin_skills:
            # Assuming source_plugin_skills["summarize"] is a function that returns a summary
            # This is a placeholder, replace with actual logic using the source plugin's skills
            try:
                # Placeholder:  Replace with actual call to source plugin's skill
                # summary = source_plugin_skills["summarize"](user_query)
                summary = "This is a placeholder summary."  # Simulate a summary
                adapted_input["summary"] = summary
            except Exception as e:
                logging.error(f"Error summarizing using source plugin: {e}")
                return None  # Adaptation failed

        # Check if adaptation logic was successful
        if not adapted_input:
            logging.warning("No adaptation logic applied. Adaptation may not be effective.")

        return adapted_input

    except TypeError as e:
        logging.error(f"Type error during skill adaptation: {e}")
        raise
    except ValueError as e:
        logging.error(f"Value error during skill adaptation: {e}")
        raise
    except Exception:
        logging.exception("Unexpected error during skill adaptation.")
        raise


def post_process_output(target_plugin_output):
    """
    Post-processes the output from the target plugin.

    Args:
        target_plugin_output (any): The raw output from the target plugin.

    Returns:
        str: A human-readable string representing the processed output.

    Raises:
        TypeError: If input type is incorrect.
        Exception: For any other error during post-processing.
    """
    try:
        if target_plugin_output is None:
            return "No output from target plugin."

        # Simple example: Convert to string
        processed_output = str(target_plugin_output)

        return processed_output

    except TypeError as e:
        logging.error(f"Type error during output post-processing: {e}")
        raise
    except Exception:
        logging.exception("Unexpected error during output post-processing.")
        raise


if __name__ == "__main__":
    # Example Usage
    source_plugin_skills = {
        "summarize": lambda x: f"Summary of: {x}"  # Placeholder summarize function
    }
    target_plugin_requirements = {"text": "string", "summary": "string"}
    user_query = "This is a long document that needs to be summarized."

    try:
        adapted_input = adapt_skill(source_plugin_skills, target_plugin_requirements, user_query)

        if adapted_input:
            print("Adapted Input:", adapted_input)

            # Simulate target plugin output
            target_plugin_output = f"Target plugin processed: {adapted_input}"

            processed_output = post_process_output(target_plugin_output)
            print("Processed Output:", processed_output)
        else:
            print("Skill adaptation failed.")

    except Exception as e:
        print(f"An error occurred: {e}")

#!/usr/bin/env python3
"""
pi-pathfinder - Analysis Script
Analyzes a given plugin directory, extracts skill descriptions, and returns a structured summary.
Generated: 2025-12-10 03:48:17
"""

import json
import argparse
from pathlib import Path
from typing import Dict
from datetime import datetime


class Analyzer:
    def __init__(self, target_path: str):
        self.target_path = Path(target_path)
        self.stats = {"total_files": 0, "total_size": 0, "file_types": {}, "issues": [], "recommendations": []}

    def analyze_directory(self) -> Dict:
        """Analyze directory structure and contents."""
        if not self.target_path.exists():
            self.stats["issues"].append(f"Path does not exist: {self.target_path}")
            return self.stats

        for file_path in self.target_path.rglob("*"):
            if file_path.is_file():
                self.analyze_file(file_path)

        return self.stats

    def analyze_file(self, file_path: Path):
        """Analyze individual file."""
        self.stats["total_files"] += 1
        self.stats["total_size"] += file_path.stat().st_size

        # Track file types
        ext = file_path.suffix.lower()
        if ext:
            self.stats["file_types"][ext] = self.stats["file_types"].get(ext, 0) + 1

        # Check for potential issues
        if file_path.stat().st_size > 100 * 1024 * 1024:  # 100MB
            self.stats["issues"].append(f"Large file: {file_path} ({file_path.stat().st_size // 1024 // 1024}MB)")

        if file_path.stat().st_size == 0:
            self.stats["issues"].append(f"Empty file: {file_path}")

    def generate_recommendations(self):
        """Generate recommendations based on analysis."""
        if self.stats["total_files"] == 0:
            self.stats["recommendations"].append("No files found - check target path")

        if len(self.stats["file_types"]) > 20:
            self.stats["recommendations"].append("Many file types detected - consider organizing")

        if self.stats["total_size"] > 1024 * 1024 * 1024:  # 1GB
            self.stats["recommendations"].append("Large total size - consider archiving old data")

    def generate_report(self) -> str:
        """Generate analysis report."""
        report = []
        report.append("\n" + "=" * 60)
        report.append("ANALYSIS REPORT - pi-pathfinder")
        report.append("=" * 60)
        report.append(f"Target: {self.target_path}")
        report.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append("")

        # Statistics
        report.append("📊 STATISTICS")
        report.append(f"  Total Files: {self.stats['total_files']:,}")
        report.append(f"  Total Size: {self.stats['total_size'] / 1024 / 1024:.2f} MB")
        report.append(f"  File Types: {len(self.stats['file_types'])}")

        # Top file types
        if self.stats["file_types"]:
            report.append("\n📁 TOP FILE TYPES")
            sorted_types = sorted(self.stats["file_types"].items(), key=lambda x: x[1], reverse=True)[:5]
            for ext, count in sorted_types:
                report.append(f"  {ext or 'no extension'}: {count} files")

        # Issues
        if self.stats["issues"]:
            report.append(f"\n⚠️  ISSUES ({len(self.stats['issues'])})")
            for issue in self.stats["issues"][:10]:
                report.append(f"  - {issue}")
            if len(self.stats["issues"]) > 10:
                report.append(f"  ... and {len(self.stats['issues']) - 10} more")

        # Recommendations
        if self.stats["recommendations"]:
            report.append("\n💡 RECOMMENDATIONS")
            for rec in self.stats["recommendations"]:
                report.append(f"  - {rec}")

        report.append("")
        return "\n".join(report)


def main():
    parser = argparse.ArgumentParser(
        description="Analyzes a given plugin directory, extracts skill descriptions, and returns a structured summary."
    )
    parser.add_argument("target", help="Target directory to analyze")
    parser.add_argument("--output", "-o", help="Output report file")
    parser.add_argument("--json", action="store_true", help="Output as JSON")

    args = parser.parse_args()

    print(f"🔍 Analyzing {args.target}...")
    analyzer = Analyzer(args.target)
    stats = analyzer.analyze_directory()
    analyzer.generate_recommendations()

    if args.json:
        output = json.dumps(stats, indent=2)
    else:
        output = analyzer.generate_report()

    if args.output:
        Path(args.output).write_text(output)
        print(f"✓ Report saved to {args.output}")
    else:
        print(output)

    return 0 if len(stats["issues"]) == 0 else 1


if __name__ == "__main__":
    import sys

    sys.exit(main())