201 lines
8.4 KiB
Python
201 lines
8.4 KiB
Python
# ProjectUtility/core/tool_discovery.py
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
from typing import Dict, List, Any, Optional
|
|
|
|
# Import the data models from the same package
|
|
from .models import ToolInfo, ToolParameter
|
|
|
|
# Constants
|
|
CONFIG_FILENAME = "tool_config.json"
|
|
|
|
# Get a logger for this module
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _parse_parameters(param_list: List[Dict[str, Any]]) -> List[ToolParameter]:
|
|
"""Parses the raw parameter list from JSON into ToolParameter objects."""
|
|
parsed_params = []
|
|
if not isinstance(param_list, list):
|
|
logger.warning(
|
|
f"Invalid 'parameters' format: Expected a list, got {type(param_list)}. Skipping parameters."
|
|
)
|
|
return []
|
|
|
|
for i, param_dict in enumerate(param_list):
|
|
if not isinstance(param_dict, dict):
|
|
logger.warning(
|
|
f"Skipping invalid parameter entry (not a dict) at index {i}."
|
|
)
|
|
continue
|
|
|
|
try:
|
|
# Basic validation for required fields in each parameter definition
|
|
if not all(k in param_dict for k in ("name", "label", "type", "required")):
|
|
logger.warning(
|
|
f"Skipping parameter definition at index {i} due to missing required keys (name, label, type, required): {param_dict}"
|
|
)
|
|
continue
|
|
|
|
# Create ToolParameter instance
|
|
parameter = ToolParameter(
|
|
name=str(param_dict["name"]),
|
|
label=str(param_dict["label"]),
|
|
type=str(param_dict["type"]).lower(), # Normalize type to lowercase
|
|
required=bool(param_dict["required"]),
|
|
default=param_dict.get("default"), # Optional, use get
|
|
description=param_dict.get("description"), # Optional
|
|
options=param_dict.get("options"), # Optional
|
|
)
|
|
# Basic validation for parameter type
|
|
allowed_types = {"string", "integer", "float", "boolean", "file", "folder"}
|
|
if parameter.type not in allowed_types:
|
|
logger.warning(
|
|
f"Parameter '{parameter.name}': Unknown type '{parameter.type}'. Treating as 'string'."
|
|
)
|
|
# Decide how to handle: treat as string, skip, or raise error? Let's default to string.
|
|
# Or alternatively: parameter = dataclasses.replace(parameter, type="string") # Requires import dataclasses
|
|
|
|
parsed_params.append(parameter)
|
|
except (TypeError, KeyError, ValueError) as e:
|
|
logger.warning(
|
|
f"Skipping parameter definition due to parsing error: {param_dict}. Error: {e}"
|
|
)
|
|
except Exception as e:
|
|
logger.exception(
|
|
f"Unexpected error parsing parameter definition: {param_dict}"
|
|
)
|
|
|
|
return parsed_params
|
|
|
|
|
|
def discover_tools(tools_base_dir: str) -> Dict[str, ToolInfo]:
|
|
"""
|
|
Scans subdirectories of `tools_base_dir` for tool configurations.
|
|
|
|
Each subdirectory is expected to contain a `tool_config.json` file.
|
|
|
|
Args:
|
|
tools_base_dir: The absolute path to the directory containing tool subdirectories.
|
|
|
|
Returns:
|
|
A dictionary where keys are tool IDs (subdirectory names) and
|
|
values are ToolInfo objects. Returns an empty dict if the base
|
|
directory doesn't exist or no valid tools are found.
|
|
"""
|
|
discovered_tools: Dict[str, ToolInfo] = {}
|
|
logger.info(f"Starting tool discovery in directory: {tools_base_dir}")
|
|
|
|
if not os.path.isdir(tools_base_dir):
|
|
logger.error(
|
|
f"Tools base directory not found or is not a directory: {tools_base_dir}"
|
|
)
|
|
return discovered_tools # Return empty dict
|
|
|
|
# Iterate through items in the base directory
|
|
for item_name in os.listdir(tools_base_dir):
|
|
tool_dir_path = os.path.join(tools_base_dir, item_name)
|
|
|
|
# Consider only directories as potential tools
|
|
if os.path.isdir(tool_dir_path):
|
|
tool_id = item_name # Use directory name as the unique ID
|
|
config_file_path = os.path.join(tool_dir_path, CONFIG_FILENAME)
|
|
|
|
logger.debug(f"Checking directory: {tool_dir_path}")
|
|
|
|
if os.path.isfile(config_file_path):
|
|
logger.debug(f"Found configuration file: {config_file_path}")
|
|
try:
|
|
with open(config_file_path, "r", encoding="utf-8") as f:
|
|
config_data = json.load(f)
|
|
|
|
# --- Validate config_data structure ---
|
|
if not isinstance(config_data, dict):
|
|
logger.warning(
|
|
f"Skipping tool '{tool_id}': Configuration file '{CONFIG_FILENAME}' does not contain a JSON object."
|
|
)
|
|
continue
|
|
|
|
# Check for mandatory fields in the JSON config
|
|
required_fields = ["display_name", "description", "command"]
|
|
if not all(field in config_data for field in required_fields):
|
|
missing = [f for f in required_fields if f not in config_data]
|
|
logger.warning(
|
|
f"Skipping tool '{tool_id}': Configuration is missing required fields: {missing}"
|
|
)
|
|
continue
|
|
|
|
# Validate command format (must be a list of strings)
|
|
command_list = config_data["command"]
|
|
if not isinstance(command_list, list) or not all(
|
|
isinstance(arg, str) for arg in command_list
|
|
):
|
|
logger.warning(
|
|
f"Skipping tool '{tool_id}': 'command' field must be a list of strings."
|
|
)
|
|
continue
|
|
if not command_list: # Must not be empty
|
|
logger.warning(
|
|
f"Skipping tool '{tool_id}': 'command' list cannot be empty."
|
|
)
|
|
continue
|
|
|
|
# Parse parameters if present
|
|
raw_parameters = config_data.get(
|
|
"parameters", []
|
|
) # Default to empty list if not present
|
|
parsed_parameters = _parse_parameters(raw_parameters)
|
|
|
|
# Create ToolInfo instance
|
|
tool_info = ToolInfo(
|
|
id=tool_id,
|
|
display_name=str(config_data["display_name"]),
|
|
description=str(config_data["description"]),
|
|
command=command_list,
|
|
working_dir=os.path.abspath(
|
|
tool_dir_path
|
|
), # Store absolute path
|
|
parameters=parsed_parameters,
|
|
version=(
|
|
str(config_data["version"])
|
|
if "version" in config_data
|
|
else None
|
|
),
|
|
has_gui=bool(config_data.get("has_gui", False)),
|
|
)
|
|
|
|
# Add the valid tool to the dictionary
|
|
discovered_tools[tool_id] = tool_info
|
|
logger.info(
|
|
f"Successfully discovered and loaded tool: {tool_info.display_name} (ID: {tool_id})"
|
|
)
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.warning(
|
|
f"Skipping tool '{tool_id}': Invalid JSON in '{config_file_path}'. Error: {e}"
|
|
)
|
|
except (TypeError, KeyError, ValueError) as e:
|
|
logger.warning(
|
|
f"Skipping tool '{tool_id}': Error processing configuration data. Error: {e}"
|
|
)
|
|
except Exception as e:
|
|
# Catch unexpected errors during file processing
|
|
logger.exception(
|
|
f"Unexpected error processing tool '{tool_id}' from '{config_file_path}'"
|
|
)
|
|
else:
|
|
logger.debug(
|
|
f"No '{CONFIG_FILENAME}' found in directory '{tool_dir_path}'. Skipping."
|
|
)
|
|
|
|
if not discovered_tools:
|
|
logger.warning(
|
|
f"Tool discovery finished. No valid tools were found in {tools_base_dir}."
|
|
)
|
|
else:
|
|
logger.info(f"Tool discovery finished. Found {len(discovered_tools)} tools.")
|
|
|
|
return discovered_tools
|