Compare commits
10
Commits
5af30f8f2d
...
f05a662414
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f05a662414 | ||
|
|
05e70a7fc4 | ||
|
|
504794f9a8 | ||
|
|
03e47bad42 | ||
|
|
a2cc56200c | ||
|
|
77a88c093d | ||
|
|
da5fdc3e4f | ||
|
|
d3d0ab9c63 | ||
|
|
48132aa21c | ||
|
|
d87fcb85be |
@@ -0,0 +1,63 @@
|
||||
# MooseCP Image Server
|
||||
|
||||
A Model Context Protocol (MCP) server designed to provide LLMs with efficient, token-optimized visual access to image directories. Instead of dumping full-resolution images (which waste tokens and cause context overflow), MooseCP provides a hierarchical workflow: **List $\rightarrow$ Scan $\rightarrow$ Preview $\rightarrow$ Inspect**.
|
||||
|
||||
## ⚠️ AI SLOP DISCLAIMER
|
||||
This server is designed to facilitate the interaction between AI models and image datasets. While it provides tools for "previewing" and "inspecting" images, remember that AI models can still hallucinate visual details, especially when working with low-resolution previews. **Always verify critical visual information with the `read_image` tool (full resolution) before drawing final conclusions.**
|
||||
|
||||
## Tools Overview
|
||||
|
||||
### 📁 `list_directory`
|
||||
Provides a simplified file-browser view of a directory.
|
||||
* **Best for**: Getting a sense of the files present and their basic metadata (size, date).
|
||||
* **Features**: Pagination and sorting by name, size, or modification date.
|
||||
|
||||
### 🖼️ `contact_sheet`
|
||||
Generates a high-density grid of thumbnails.
|
||||
* **Best for**: Quickly scanning hundreds of images to find a specific one or get a general "vibe" of a folder.
|
||||
* **Note**: This tool is paginated. You must iterate through pages to see all images in a folder.
|
||||
* **Workflow**: Use the indices shown on the contact sheet to call `preview_image`.
|
||||
|
||||
### 🔍 `preview_image`
|
||||
Provides medium-detail previews optimized for the model's token budget.
|
||||
* **Best for**: Comparing a few candidates, inspecting specific details, or selecting a "favorite" image.
|
||||
* **Efficiency**: Automatically calculates dimensions to fit the model's patch size (e.g., 48px patches for Gemma 4), ensuring maximum detail without wasting tokens on padding.
|
||||
* **Workflow**: Pass indices from the `contact_sheet` or a direct file path.
|
||||
|
||||
### 📖 `read_png_metadata`
|
||||
Extracts AI generation parameters from PNG files.
|
||||
* **Best for**: Retrieving prompts, seeds, and model hashes from AI-generated images.
|
||||
|
||||
### 📸 `read_image`
|
||||
Returns the full-resolution image.
|
||||
* **Best for**: Final confirmation or deep visual analysis where every pixel counts.
|
||||
|
||||
---
|
||||
|
||||
## Installation & Requirements
|
||||
|
||||
### Dependencies
|
||||
This server requires Python 3.10+ and the following packages:
|
||||
* `uvicorn`: ASGI server for the SSE transport.
|
||||
* `starlette`: Lightweight ASGI framework.
|
||||
* `Pillow`: Image processing and thumbnail generation.
|
||||
|
||||
```bash
|
||||
pip install uvicorn starlette Pillow
|
||||
```
|
||||
|
||||
### Setup
|
||||
1. Clone this repository to your server.
|
||||
2. (Optional) Edit `config.py` to adjust the server port, log level, or token budgets if you are using a model other than Gemma 4.
|
||||
3. Run the server:
|
||||
```bash
|
||||
python main.py
|
||||
```
|
||||
|
||||
## Configuration (`config.py`)
|
||||
|
||||
You can tune the server's behavior in `config.py`:
|
||||
- **`PATCH_SIZE`**: Set this to `(clip.vision.patch_size * n_merge)` for your specific model to ensure token-perfect resizing.
|
||||
- **`PREVIEW_TOKEN_BUDGET`**: Controls how many tokens the `preview_image` tool aims for (default: 70).
|
||||
- **`CONTACT_SHEET_COLS/ROWS`**: Adjust the grid size to fit within your model's maximum context window.
|
||||
- **`IMAGE_QUALITY`**: Adjust JPEG compression (1-100).
|
||||
@@ -0,0 +1,47 @@
|
||||
import logging
|
||||
|
||||
# --- Server & Logs ---
|
||||
HOST = "127.0.0.1"
|
||||
PORT = 8000
|
||||
LOG_LEVEL = "WARNING" # Options: "DEBUG", "INFO", "WARNING", "ERROR"
|
||||
LOG_FILE = "debug.log"
|
||||
|
||||
# --- Model Specific Token Tuning (Tuned for Gemma 4) ---
|
||||
# Patch size is typically (clip.vision.patch_size * n_merge)
|
||||
# For Gemma 4, this is 48px.
|
||||
PATCH_SIZE = 48
|
||||
|
||||
# The target token count for a single image preview.
|
||||
# The actual budget will be this or greater, but as small as possible.
|
||||
PREVIEW_TOKEN_BUDGET = 70
|
||||
|
||||
# Contact sheet grid optimized to fit within Gemma 4's 1120 max token budget
|
||||
# (10 cols * 7 rows) = 70 images.
|
||||
# Each thumb (192px) is 4x4 patches.
|
||||
# 70 images * (4*4) = 1120 tokens.
|
||||
CONTACT_SHEET_COLS = 10
|
||||
CONTACT_SHEET_ROWS = 7
|
||||
CONTACT_SHEET_THUMB_SIZE = 192
|
||||
|
||||
# --- Image Quality ---
|
||||
# JPEG image quality (usually max 95).
|
||||
# See: https://pillow.readthedocs.io/en/stable/handbook/image-file-formats.html#jpeg-saving
|
||||
IMAGE_QUALITY = 95
|
||||
|
||||
# --- Font Configuration ---
|
||||
# Generic font names for cross-platform compatibility
|
||||
# Pillow's truetype() can often find these by name in the system path
|
||||
SYSTEM_FONT_NAMES = [
|
||||
"arialbd.ttf",
|
||||
"DejaVuSans-Bold",
|
||||
"LiberationSans-Bold",
|
||||
"Verdana",
|
||||
"Tahoma",
|
||||
]
|
||||
|
||||
# Fallback absolute paths for Linux systems
|
||||
FALLBACK_FONT_PATHS = [
|
||||
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
|
||||
"/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf",
|
||||
"/usr/share/fonts/truetype/freefont/FreeSansBold.ttf",
|
||||
]
|
||||
@@ -10,15 +10,18 @@ from starlette.middleware.cors import CORSMiddleware
|
||||
|
||||
from mcp_logic import MCPServer
|
||||
from tools import TOOL_REGISTRY
|
||||
import config
|
||||
|
||||
|
||||
# --- Configuration & Logging ---
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
filename="debug.log",
|
||||
level=getattr(logging, config.LOG_LEVEL),
|
||||
filename=config.LOG_FILE,
|
||||
|
||||
filemode="a",
|
||||
format="%(asctime)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
logger = logging.getLogger("MattCP")
|
||||
logger = logging.getLogger("MooseCP")
|
||||
|
||||
mcp_logic = MCPServer()
|
||||
|
||||
@@ -90,11 +93,12 @@ app.add_middleware(
|
||||
if __name__ == "__main__":
|
||||
config = uvicorn.Config(
|
||||
app=app,
|
||||
host="127.0.0.1",
|
||||
port=8000,
|
||||
host=config.HOST,
|
||||
port=config.PORT,
|
||||
log_level="info",
|
||||
timeout_graceful_shutdown=2 # Reduced from 5 to 2 seconds
|
||||
)
|
||||
|
||||
server = uvicorn.Server(config)
|
||||
|
||||
def signal_handler(sig, frame):
|
||||
|
||||
+10
-2
@@ -3,6 +3,7 @@ import uuid
|
||||
import json
|
||||
from typing import Any, Dict, Union, List
|
||||
from starlette.responses import Response
|
||||
from tools.utils import ToolError
|
||||
|
||||
class MCPServer:
|
||||
def __init__(self):
|
||||
@@ -23,7 +24,7 @@ class MCPServer:
|
||||
return {
|
||||
"protocolVersion": "2024-11-05",
|
||||
"capabilities": {"tools": {}},
|
||||
"serverInfo": {"name": "MattCP", "version": "0.4.1"}
|
||||
"serverInfo": {"name": "MooseCP", "version": "0.4.1"}
|
||||
}
|
||||
|
||||
if method == "tools/list":
|
||||
@@ -40,7 +41,14 @@ class MCPServer:
|
||||
media_type="application/json"
|
||||
)
|
||||
|
||||
result_data = await tool.handler(args)
|
||||
try:
|
||||
result_data = await tool.handler(args)
|
||||
except ToolError as e:
|
||||
# Convert tool errors into a text response so the LLM can understand and potentially fix the input
|
||||
result_data = {"text": str(e)}
|
||||
except Exception as e:
|
||||
# Catch-all for unexpected crashes to prevent server death
|
||||
result_data = {"text": f"Unexpected internal error: {str(e)}"}
|
||||
|
||||
# Handle results that are already lists of content (for multimodal/multi-part responses)
|
||||
if isinstance(result_data, list):
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
#!/bin/bash
|
||||
# Start the MCP Image Server in the background
|
||||
nohup /home/matt/code/llama.cpp/build/bin/mcp-image-server/venv/bin/python3 /home/matt/code/llama.cpp/build/bin/mcp-image-server/server.py > /home/matt/code/llama.cpp/build/bin/mcp-image-server/server.log 2>&1 &
|
||||
echo "MCP Image Server started in background on port 8000"
|
||||
echo "Log file: /home/matt/code/llama.cpp/build/bin/mcp-image-server/server.log"
|
||||
+39
-9
@@ -18,14 +18,16 @@ class Tool:
|
||||
# Import handlers from separate files
|
||||
from .read_image import handle as read_image_handler
|
||||
from .read_metadata import handle as read_png_metadata_handler
|
||||
from .list_thumbnails import handle as list_thumbnails_handler
|
||||
from .contact_sheet import handle as contact_sheet_handler
|
||||
from .list_directory import handle as list_directory_details_handler
|
||||
from .preview_image import handle as preview_image_handler
|
||||
from .get_text_context import handle as get_text_context_handler
|
||||
|
||||
# Central registry of all available tools
|
||||
TOOL_REGISTRY = [
|
||||
Tool(
|
||||
name="read_image",
|
||||
description="Reads an image from the disk and returns it as an image content object.",
|
||||
description="Reads an image at full resolution for maximum detail. Use this for final confirmation of a selected image or when deep visual analysis of a specific file is required.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {"path": {"type": "string", "description": "Path to the image file"}},
|
||||
@@ -35,7 +37,7 @@ TOOL_REGISTRY = [
|
||||
),
|
||||
Tool(
|
||||
name="read_png_metadata",
|
||||
description="Reads the metadata (tEXt, zTXt, iTXt) from a PNG image to fetch info such as AI prompts.",
|
||||
description="Reads the metadata (tEXt, zTXt, iTXt) from a PNG image to fetch info such as prompts for AI generated images.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {"path": {"type": "string", "description": "Path to the PNG file"}},
|
||||
@@ -44,23 +46,22 @@ TOOL_REGISTRY = [
|
||||
handler=read_png_metadata_handler
|
||||
),
|
||||
Tool(
|
||||
name="list_thumbnails",
|
||||
description="Generates a thumbnail grid of images in a directory. Supports pagination and sorting.",
|
||||
name="contact_sheet",
|
||||
description="Generates a coarse, high-density overview of a directory. This tool is paginated; a single page may not show all images. To perform a comprehensive scan or find specific images, you MUST iterate through multiple pages. DO NOT use this for selecting specific images or making quality judgments; use preview_image for those tasks.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {"type": "string", "description": "Path to the directory containing images"},
|
||||
"page": {"type": "integer", "description": "The page number to retrieve (1-indexed)", "default": 1},
|
||||
"page_size": {"type": "integer", "description": "Number of images per page", "default": 64},
|
||||
"page": {"type": "integer", "description": "The page number to retrieve (1-indexed). Increment this value to navigate through the full list of images in the folder.", "default": 1},
|
||||
"sort_by": {"type": "string", "description": "Sort order: 'mtime' (newest first, default), 'name' (alphabetical), or 'size' (largest first)", "default": "mtime"},
|
||||
},
|
||||
"required": ["path"],
|
||||
},
|
||||
handler=list_thumbnails_handler
|
||||
handler=contact_sheet_handler
|
||||
),
|
||||
Tool(
|
||||
name="list_directory",
|
||||
description="Lists the contents of a directory in a file browser format. Directories first, then files. Supports pagination and sorting.",
|
||||
description="Lists the contents of a directory in a simplified format. Directories first, then files. Supports pagination and sorting.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
@@ -73,4 +74,33 @@ TOOL_REGISTRY = [
|
||||
},
|
||||
handler=list_directory_details_handler
|
||||
),
|
||||
Tool(
|
||||
name="preview_image",
|
||||
description="Provides low-resolution previews and detailed file info for specific images. It uses the minimum viable token budget, so some fine details will be missed. This is the PRIMARY tool for inspecting files, comparing candidates, or selecting images before moving to read_image for maximum detail.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {"type": "string", "description": "Path to an image file or a directory containing images"},
|
||||
"indices": {"type": "string", "description": "1-based indices or ranges (e.g., '1, 3-5, 10') to preview from the directory. Required if path is a directory."},
|
||||
"sort_by": {"type": "string", "description": "Sort order to resolve indices: 'mtime' (default), 'name', or 'size'."},
|
||||
},
|
||||
"required": ["path"],
|
||||
},
|
||||
handler=preview_image_handler
|
||||
),
|
||||
Tool(
|
||||
name="get_text_context",
|
||||
description="Extracts specific sections of a file with absolute line numbers and surrounding context. This tool MUST be called immediately before edit_file to verify the exact line numbers and content of the block being replaced, preventing misalignment errors.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {"type": "string", "description": "Path to the file"},
|
||||
"pattern": {"type": "string", "description": "Regex pattern to search for"},
|
||||
"lines": {"type": "string", "description": "1-based indices or ranges (e.g., '10-20, 45')"},
|
||||
"context": {"type": "integer", "description": "Number of lines of context to show above and below", "default": 1},
|
||||
},
|
||||
"required": ["path"],
|
||||
},
|
||||
handler=get_text_context_handler
|
||||
),
|
||||
]
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
import base64
|
||||
import logging
|
||||
import io
|
||||
import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List
|
||||
from PIL import Image, ImageDraw, ImageFont, ImageOps
|
||||
import config
|
||||
from tools.utils import format_relative_time, ToolError, get_file_info_list, sort_file_list, get_paginated_list
|
||||
|
||||
|
||||
logger = logging.getLogger("MooseCP")
|
||||
|
||||
async def handle(args: Dict[str, Any]):
|
||||
"""
|
||||
Generates a contact sheet of images.
|
||||
Grid: 10 columns x 7 rows (70 images total).
|
||||
Sized for optimal token usage (1120 tokens) on llama.cpp.
|
||||
"""
|
||||
dir_path_str = args.get("path")
|
||||
page = int(args.get("page", 1))
|
||||
sort_by = args.get("sort_by", "mtime")
|
||||
|
||||
if not dir_path_str:
|
||||
raise ToolError("Missing path argument")
|
||||
|
||||
dir_path = Path(dir_path_str)
|
||||
if not dir_path.is_dir():
|
||||
raise ToolError(f"{dir_path_str} is not a directory.")
|
||||
|
||||
exts = ('.png', '.jpg', '.jpeg', '.webp', '.bmp')
|
||||
file_info_list = get_file_info_list(dir_path, extensions=exts)
|
||||
|
||||
if not file_info_list:
|
||||
return {"text": "No supported images found in the directory."}
|
||||
|
||||
# Sorting
|
||||
file_info_list = sort_file_list(file_info_list, sort_by)
|
||||
|
||||
sort_labels = {
|
||||
"mtime": "Date (newest first)",
|
||||
"size": "Size (largest first)",
|
||||
"name": "Name (alphabetical)"
|
||||
}
|
||||
sort_label = sort_labels.get(sort_by, "Name (alphabetical)")
|
||||
|
||||
# Fixed grid dimensions for token optimization
|
||||
COLS = config.CONTACT_SHEET_COLS
|
||||
ROWS = config.CONTACT_SHEET_ROWS
|
||||
PAGE_SIZE = COLS * ROWS
|
||||
|
||||
|
||||
total_files = len(file_info_list)
|
||||
paged_files = get_paginated_list(file_info_list, page, PAGE_SIZE)
|
||||
|
||||
if not paged_files:
|
||||
return {"text": f"No images found on page {page}."}
|
||||
|
||||
thumb_size = 192 # 192 / 48 = 4 patches per side
|
||||
canvas_w = COLS * thumb_size
|
||||
canvas_h = ROWS * thumb_size
|
||||
|
||||
canvas = Image.new('RGB', (canvas_w, canvas_h), (30, 30, 30))
|
||||
draw = ImageDraw.Draw(canvas)
|
||||
|
||||
# Font loading
|
||||
font = None
|
||||
# Try generic system font names first
|
||||
for font_name in config.SYSTEM_FONT_NAMES:
|
||||
try:
|
||||
font = ImageFont.truetype(font_name, 24)
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
# Fallback to absolute paths
|
||||
if font is None:
|
||||
for font_path in config.FALLBACK_FONT_PATHS:
|
||||
try:
|
||||
font = ImageFont.truetype(font_path, 24)
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
if font is None:
|
||||
font = ImageFont.load_default()
|
||||
|
||||
|
||||
start_idx = (page - 1) * PAGE_SIZE
|
||||
for i, info in enumerate(paged_files):
|
||||
full_path = info["path"]
|
||||
row = i // COLS
|
||||
col = i % COLS
|
||||
|
||||
x = col * thumb_size
|
||||
y = row * thumb_size
|
||||
global_index = start_idx + i + 1
|
||||
|
||||
try:
|
||||
with Image.open(full_path) as img:
|
||||
# Apply EXIF orientation to fix sideways images
|
||||
img = ImageOps.exif_transpose(img)
|
||||
img.thumbnail((thumb_size, thumb_size))
|
||||
off_x = (thumb_size - img.width) // 2
|
||||
off_y = (thumb_size - img.height) // 2
|
||||
canvas.paste(img, (x + off_x, y + off_y))
|
||||
|
||||
text = str(global_index)
|
||||
if font:
|
||||
bbox = draw.textbbox((0, 0), text, font=font)
|
||||
text_w, text_h = bbox[2] - bbox[0], bbox[3] - bbox[1]
|
||||
else:
|
||||
text_w, text_h = len(text) * 8, 15
|
||||
|
||||
# Adjusted black box to align perfectly with the bottom of the image (y + thumb_size)
|
||||
draw.rectangle([x, y + thumb_size - text_h - 4, x + text_w + 5, y + thumb_size], fill=(0, 0, 0, 180))
|
||||
draw.text((x + 2, y + thumb_size - text_h - 7), text, fill=(255, 255, 255), font=font)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to process {info['name']}: {e}")
|
||||
draw.text((x + 5, y + thumb_size // 2), "Error", fill=(255, 0, 0), font=font)
|
||||
|
||||
buf = io.BytesIO()
|
||||
# Save as JPEG to save data
|
||||
canvas.save(buf, format='JPEG', quality=config.IMAGE_QUALITY)
|
||||
img_data = base64.b64encode(buf.getvalue()).decode("utf-8")
|
||||
|
||||
total_pages = (total_files + PAGE_SIZE - 1) // PAGE_SIZE
|
||||
|
||||
first_on_page = paged_files[0]
|
||||
last_on_page = paged_files[-1]
|
||||
|
||||
if sort_by == "mtime":
|
||||
first_val = f"{format_relative_time(first_on_page['mtime'])} ({datetime.datetime.fromtimestamp(first_on_page['mtime']).strftime('%Y-%m-%d')})"
|
||||
last_val = f"{format_relative_time(last_on_page['mtime'])} ({datetime.datetime.fromtimestamp(last_on_page['mtime']).strftime('%Y-%m-%d')})"
|
||||
elif sort_by == "size":
|
||||
first_val = f"{first_on_page['size']/1024:.1f}KB"
|
||||
last_val = f"{last_on_page['size']/1024:.1f}KB"
|
||||
else:
|
||||
first_val = first_on_page['name']
|
||||
last_val = last_on_page['name']
|
||||
|
||||
header_text = (
|
||||
f"Directory: {dir_path_str}\n"
|
||||
f"Page {page} of {total_pages} ({total_files} images total). Sorted by: {sort_label}\n"
|
||||
f"Range shown: {first_val} ... {last_val}"
|
||||
)
|
||||
|
||||
return [
|
||||
{"type": "text", "text": header_text},
|
||||
{"type": "image", "data": img_data, "mimeType": "image/jpeg"}
|
||||
]
|
||||
@@ -0,0 +1,86 @@
|
||||
import re
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Set
|
||||
from tools.utils import ToolError, parse_indices
|
||||
|
||||
logger = logging.getLogger("MooseCP")
|
||||
|
||||
async def handle(args: Dict[str, Any]):
|
||||
"""
|
||||
Extracts specific sections of a file with absolute line numbers and surrounding context.
|
||||
This tool MUST be called immediately before edit_file to verify the exact line numbers
|
||||
and content of the block being replaced, preventing misalignment errors.
|
||||
"""
|
||||
path_str = args.get("path")
|
||||
pattern = args.get("pattern")
|
||||
lines_str = args.get("lines")
|
||||
context = int(args.get("context", 1))
|
||||
|
||||
if not path_str:
|
||||
raise ToolError("Missing path argument")
|
||||
|
||||
path = Path(path_str)
|
||||
if not path.is_file():
|
||||
raise ToolError(f"{path_str} is not a file.")
|
||||
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
||||
all_lines = f.readlines()
|
||||
except Exception as e:
|
||||
raise ToolError(f"Error reading file {path_str}: {e}")
|
||||
|
||||
total_lines = len(all_lines)
|
||||
|
||||
# Case 1: No pattern and no lines provided -> Dump whole file
|
||||
if not pattern and not lines_str:
|
||||
output = [f"{i+1}: {line}" for i, line in enumerate(all_lines)]
|
||||
return {"text": "".join(output)}
|
||||
|
||||
# Determine target line numbers (1-based)
|
||||
target_lines: Set[int] = set()
|
||||
|
||||
if pattern:
|
||||
try:
|
||||
regex = re.compile(pattern)
|
||||
for i, line in enumerate(all_lines):
|
||||
if regex.search(line):
|
||||
target_lines.add(i + 1)
|
||||
except re.error as e:
|
||||
raise ToolError(f"Invalid regex pattern: {e}")
|
||||
|
||||
if lines_str:
|
||||
try:
|
||||
indices = parse_indices(lines_str)
|
||||
for idx in indices:
|
||||
if 1 <= idx <= total_lines:
|
||||
target_lines.add(idx)
|
||||
except ToolError as e:
|
||||
raise e
|
||||
except Exception as e:
|
||||
raise ToolError(f"Error parsing line indices: {e}")
|
||||
|
||||
if not target_lines:
|
||||
return {"text": "No matching lines found."}
|
||||
|
||||
# Expand targets with context
|
||||
lines_to_show: Set[int] = set()
|
||||
for line in target_lines:
|
||||
for offset in range(-context, context + 1):
|
||||
ln = line + offset
|
||||
if 1 <= ln <= total_lines:
|
||||
lines_to_show.add(ln)
|
||||
|
||||
# Sort and format output
|
||||
sorted_lines = sorted(list(lines_to_show))
|
||||
output = []
|
||||
last_line = None
|
||||
|
||||
for ln in sorted_lines:
|
||||
if last_line is not None and ln > last_line + 1:
|
||||
output.append("...\n")
|
||||
|
||||
output.append(f"{ln}: {all_lines[ln-1]}")
|
||||
last_line = ln
|
||||
|
||||
return {"text": "".join(output)}
|
||||
+23
-58
@@ -1,10 +1,9 @@
|
||||
import os
|
||||
import logging
|
||||
import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
from tools.utils import format_relative_time
|
||||
from tools.utils import format_relative_time, ToolError, get_file_info_list, sort_file_list
|
||||
|
||||
logger = logging.getLogger("MattCP")
|
||||
logger = logging.getLogger("MooseCP")
|
||||
|
||||
async def handle(args: Dict[str, Any]):
|
||||
"""
|
||||
@@ -12,65 +11,24 @@ async def handle(args: Dict[str, Any]):
|
||||
Directories are listed first, then files.
|
||||
Supports pagination and sorting.
|
||||
"""
|
||||
path = args.get("path")
|
||||
path_str = args.get("path")
|
||||
page = int(args.get("page", 1))
|
||||
page_size = int(args.get("page_size", 64))
|
||||
sort_by = args.get("sort_by", "mtime")
|
||||
|
||||
if not path:
|
||||
return {"text": "Error: Missing path argument"}
|
||||
if not os.path.isdir(path):
|
||||
return {"text": f"Error: {path} is not a directory."}
|
||||
if not path_str:
|
||||
raise ToolError("Missing path argument")
|
||||
|
||||
try:
|
||||
entries = os.listdir(path)
|
||||
except Exception as e:
|
||||
return {"text": f"Error reading directory: {e}"}
|
||||
path = Path(path_str)
|
||||
if not path.is_dir():
|
||||
raise ToolError(f"{path_str} is not a directory.")
|
||||
|
||||
all_items = []
|
||||
for entry in entries:
|
||||
full_path = os.path.join(path, entry)
|
||||
try:
|
||||
stats = os.stat(full_path)
|
||||
mtime_rel = format_relative_time(stats.st_mtime)
|
||||
|
||||
if os.path.isdir(full_path):
|
||||
try:
|
||||
count = len(os.listdir(full_path))
|
||||
except:
|
||||
count = 0
|
||||
all_items.append({
|
||||
"type": "dir",
|
||||
"name": entry,
|
||||
"info": f"{count} items",
|
||||
"mtime": mtime_rel,
|
||||
"mtime_raw": stats.st_mtime,
|
||||
"size": stats.st_size
|
||||
})
|
||||
else:
|
||||
size_kb = stats.st_size / 1024
|
||||
all_items.append({
|
||||
"type": "file",
|
||||
"name": entry,
|
||||
"info": f"{size_kb:.1f}KB",
|
||||
"mtime": mtime_rel,
|
||||
"mtime_raw": stats.st_mtime,
|
||||
"size": stats.st_size
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"Could not stat {entry}: {e}")
|
||||
|
||||
# Sort Logic: Directories always come first, then apply secondary sort
|
||||
if sort_by == "mtime":
|
||||
all_items.sort(key=lambda x: (x["type"] == "file", -x["mtime_raw"]))
|
||||
elif sort_by == "size":
|
||||
all_items.sort(key=lambda x: (x["type"] == "file", -x["size"]))
|
||||
else:
|
||||
all_items.sort(key=lambda x: (x["type"] == "file", x["name"].lower()))
|
||||
all_items = get_file_info_list(path)
|
||||
all_items = sort_file_list(all_items, sort_by)
|
||||
|
||||
total_items = len(all_items)
|
||||
|
||||
# Custom Pagination: show all if <= 100, otherwise paginate by 64
|
||||
# Custom Pagination: show all if <= 100, otherwise paginate by page_size
|
||||
if total_items <= 100:
|
||||
start_idx = 0
|
||||
end_idx = total_items
|
||||
@@ -89,11 +47,18 @@ async def handle(args: Dict[str, Any]):
|
||||
|
||||
output = []
|
||||
for item in paged_items:
|
||||
if item["type"] == "dir":
|
||||
output.append(f"[DIR] {item['name']} | {item['info']} | {item['mtime']}")
|
||||
mtime_rel = format_relative_time(item["mtime"])
|
||||
if item["is_dir"]:
|
||||
try:
|
||||
count = len(list(item["path"].iterdir()))
|
||||
info = f"{count} items"
|
||||
except:
|
||||
info = "0 items"
|
||||
output.append(f"[DIR] {item['name']} | {info} | {mtime_rel}")
|
||||
else:
|
||||
output.append(f" {item['name']} | {item['info']} | {item['mtime']}")
|
||||
size_kb = item["size"] / 1024
|
||||
output.append(f" {item['name']} | {size_kb:.1f}KB | {mtime_rel}")
|
||||
|
||||
header = f"Listing of {path}\nPage {effective_page} of {total_pages} ({total_items} total items). Showing {start_idx+1}-{min(end_idx, total_items)}."
|
||||
header = f"Listing of {path_str}\nPage {effective_page} of {total_pages} ({total_items} total items). Showing {start_idx+1}-{min(end_idx, total_items)}."
|
||||
|
||||
return {"text": f"{header}\n\n" + "\n".join(output)}
|
||||
|
||||
@@ -1,159 +0,0 @@
|
||||
import os
|
||||
import base64
|
||||
import logging
|
||||
import asyncio
|
||||
import io
|
||||
import datetime
|
||||
import time
|
||||
from typing import Any, Dict
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
logger = logging.getLogger("MattCP")
|
||||
|
||||
async def handle(args: Dict[str, Any]):
|
||||
"""
|
||||
Generates a thumbnail grid of images in a directory.
|
||||
Returns a combined text list (with global indices) and a visual contact sheet.
|
||||
"""
|
||||
dir_path = args.get("path")
|
||||
page = int(args.get("page", 1))
|
||||
page_size = int(args.get("page_size", 64))
|
||||
sort_by = args.get("sort_by", "mtime")
|
||||
|
||||
if not dir_path:
|
||||
return {"text": "Error: Missing path argument"}
|
||||
if not os.path.isdir(dir_path):
|
||||
return {"text": f"Error: {dir_path} is not a directory."}
|
||||
|
||||
exts = ('.png', '.jpg', '.jpeg', '.webp', '.bmp')
|
||||
file_info_list = []
|
||||
|
||||
# 1. Gather all valid images and their stats
|
||||
for f in os.listdir(dir_path):
|
||||
if f.lower().endswith(exts):
|
||||
full_path = os.path.join(dir_path, f)
|
||||
try:
|
||||
stats = os.stat(full_path)
|
||||
file_info_list.append({
|
||||
"name": f,
|
||||
"path": full_path,
|
||||
"mtime": stats.st_mtime,
|
||||
"size": stats.st_size
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"Could not stat {f}: {e}")
|
||||
|
||||
if not file_info_list:
|
||||
return {"text": "No supported images found in the directory."}
|
||||
|
||||
# 2. Sort the list based on requested criteria
|
||||
if sort_by == "mtime":
|
||||
file_info_list.sort(key=lambda x: x["mtime"], reverse=True)
|
||||
elif sort_by == "size":
|
||||
file_info_list.sort(key=lambda x: x["size"], reverse=True)
|
||||
else:
|
||||
file_info_list.sort(key=lambda x: x["name"].lower())
|
||||
|
||||
# 3. Apply Pagination
|
||||
total_files = len(file_info_list)
|
||||
start_idx = (page - 1) * page_size
|
||||
end_idx = start_idx + page_size
|
||||
paged_files = file_info_list[start_idx:end_idx]
|
||||
|
||||
if not paged_files:
|
||||
return {"text": f"No images found on page {page}."}
|
||||
|
||||
# Layout constants
|
||||
thumb_size = 160
|
||||
padding = 15
|
||||
label_height = 35
|
||||
|
||||
num_files = len(paged_files)
|
||||
cols = int(num_files**0.5) if num_files > 0 else 1
|
||||
if cols == 0: cols = 1
|
||||
rows = (num_files + cols - 1) // cols
|
||||
|
||||
canvas_w = cols * (thumb_size + padding) + padding
|
||||
canvas_h = rows * (thumb_size + label_height + padding) + padding
|
||||
|
||||
# Create dark gray background
|
||||
canvas = Image.new('RGB', (canvas_w, canvas_h), (30, 30, 30))
|
||||
draw = ImageDraw.Draw(canvas)
|
||||
|
||||
# Attempt to load a bold system font for the indices
|
||||
font = None
|
||||
font_paths = [
|
||||
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
|
||||
"/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf",
|
||||
"/usr/share/fonts/truetype/freefont/FreeSansBold.ttf",
|
||||
]
|
||||
for path in font_paths:
|
||||
if os.path.exists(path):
|
||||
try:
|
||||
font = ImageFont.truetype(path, 20)
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
if font is None:
|
||||
font = ImageFont.load_default()
|
||||
|
||||
file_details = []
|
||||
|
||||
# 4. Paste thumbnails and render indices
|
||||
for i, info in enumerate(paged_files):
|
||||
filename = info["name"]
|
||||
full_path = info["path"]
|
||||
row = i // cols
|
||||
col = i % cols
|
||||
|
||||
x = padding + col * (thumb_size + padding)
|
||||
y = padding + row * (thumb_size + label_height + padding)
|
||||
|
||||
# Continuous indexing across pages
|
||||
global_index = start_idx + i + 1
|
||||
|
||||
try:
|
||||
with Image.open(full_path) as img:
|
||||
width, height = img.size
|
||||
img.thumbnail((thumb_size, thumb_size))
|
||||
# Center image in its slot
|
||||
off_x = (thumb_size - img.width) // 2
|
||||
off_y = (thumb_size - img.height) // 2
|
||||
canvas.paste(img, (x + off_x, y + off_y))
|
||||
|
||||
# Format text metadata
|
||||
from tools.utils import format_relative_time # Import helper from utils
|
||||
mod_time_rel = format_relative_time(info["mtime"])
|
||||
size_kb = info["size"] / 1024
|
||||
file_details.append(f"{global_index}. {filename} | {width}x{height} | {size_kb:.1f}KB | {mod_time_rel}")
|
||||
|
||||
# Draw the index number centered under the image
|
||||
text = str(global_index)
|
||||
if font:
|
||||
bbox = draw.textbbox((0, 0), text, font=font)
|
||||
text_w = bbox[2] - bbox[0]
|
||||
else:
|
||||
text_w = len(text) * 7
|
||||
|
||||
text_x = x + (thumb_size - text_w) // 2
|
||||
draw.text((text_x, y + thumb_size + 2), text, fill=(255, 255, 255), font=font)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to process {filename}: {e}")
|
||||
draw.text((x, y + thumb_size // 2), "Error", fill=(255, 0, 0), font=font)
|
||||
file_details.append(f"{global_index}. {filename} | ERROR")
|
||||
|
||||
# 5. Encode result to base64 PNG
|
||||
buf = io.BytesIO()
|
||||
canvas.save(buf, format='PNG')
|
||||
img_data = base64.b64encode(buf.getvalue()).decode("utf-8")
|
||||
|
||||
total_pages = (total_files + page_size - 1) // page_size
|
||||
header_text = f"Directory: {dir_path}\nPage {page} of {total_pages} ({total_files} total images). Showing {start_idx+1}-{min(end_idx, total_files)}."
|
||||
|
||||
details_text = "\n".join(file_details)
|
||||
|
||||
return [
|
||||
{"type": "text", "text": f"{header_text}\n\nFile List:\n{details_text}"},
|
||||
{"type": "image", "data": img_data, "mimeType": "image/png"}
|
||||
]
|
||||
@@ -0,0 +1,132 @@
|
||||
import math
|
||||
import base64
|
||||
import logging
|
||||
import io
|
||||
import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Tuple
|
||||
from PIL import Image, ImageOps
|
||||
import config
|
||||
from tools.utils import format_relative_time, ToolError, get_file_info_list, sort_file_list, parse_indices
|
||||
|
||||
|
||||
logger = logging.getLogger("MooseCP")
|
||||
|
||||
def calculate_patch_dimensions(orig_w: int, orig_h: int, target_tokens: int = None):
|
||||
"""
|
||||
Calculates optimal pixel dimensions to hit a token budget of at least target_tokens.
|
||||
Validates budget based on actual rounded pixel dimensions to avoid float precision errors.
|
||||
"""
|
||||
if target_tokens is None:
|
||||
target_tokens = config.PREVIEW_TOKEN_BUDGET
|
||||
|
||||
ar = orig_w / orig_h
|
||||
|
||||
# Case A: Width is the anchor (non-padded)
|
||||
w_anchor = 1
|
||||
while True:
|
||||
w_px = w_anchor * config.PATCH_SIZE
|
||||
h_px = round(w_px / ar)
|
||||
# Calculate actual tokens based on resulting pixel dimensions
|
||||
tokens = math.ceil(w_px / config.PATCH_SIZE) * math.ceil(h_px / config.PATCH_SIZE)
|
||||
if tokens >= target_tokens:
|
||||
res_a = {"w": w_px, "h": h_px, "tokens": tokens}
|
||||
break
|
||||
w_anchor += 1
|
||||
|
||||
# Case B: Height is the anchor (non-padded)
|
||||
h_anchor = 1
|
||||
while True:
|
||||
h_px = h_anchor * config.PATCH_SIZE
|
||||
w_px = round(h_px * ar)
|
||||
# Calculate actual tokens based on resulting pixel dimensions
|
||||
tokens = math.ceil(w_px / config.PATCH_SIZE) * math.ceil(h_px / config.PATCH_SIZE)
|
||||
if tokens >= target_tokens:
|
||||
res_b = {"w": w_px, "h": h_px, "tokens": tokens}
|
||||
break
|
||||
h_anchor += 1
|
||||
|
||||
# Pick the one with the smaller token count
|
||||
best = res_a if res_a["tokens"] <= res_b["tokens"] else res_b
|
||||
return max(1, best["w"]), max(1, best["h"])
|
||||
|
||||
async def handle(args: Dict[str, Any]):
|
||||
"""
|
||||
Generates small thumbnails of images with detailed info.
|
||||
Can take a single file path, or a directory with indices/ranges from contact_sheet.
|
||||
Thumbnails are optimized for ~70 token usage.
|
||||
"""
|
||||
path_str = args.get("path")
|
||||
indices_str = args.get("indices")
|
||||
sort_by = args.get("sort_by", "mtime")
|
||||
|
||||
if not path_str:
|
||||
raise ToolError("Missing path argument")
|
||||
|
||||
path = Path(path_str)
|
||||
# Store as (path, index_label) where index_label is None if direct path
|
||||
target_files: List[Tuple[Path, Any]] = []
|
||||
|
||||
if path.is_file():
|
||||
target_files.append((path, None))
|
||||
elif path.is_dir():
|
||||
if not indices_str:
|
||||
raise ToolError("Indices are required when providing a directory path. Example: '1, 5-10'")
|
||||
|
||||
exts = ('.png', '.jpg', '.jpeg', '.webp', '.bmp')
|
||||
file_info_list = get_file_info_list(path, extensions=exts)
|
||||
file_info_list = sort_file_list(file_info_list, sort_by)
|
||||
|
||||
indices = parse_indices(indices_str)
|
||||
for idx in indices:
|
||||
# contact_sheet uses 1-based indexing
|
||||
if 1 <= idx <= len(file_info_list):
|
||||
target_files.append((file_info_list[idx-1]["path"], idx))
|
||||
else:
|
||||
logger.warning(f"Index {idx} out of range for directory {path_str}")
|
||||
else:
|
||||
raise ToolError(f"Path {path_str} is neither a file nor a directory.")
|
||||
|
||||
if not target_files:
|
||||
return {"text": "No valid images found to preview."}
|
||||
|
||||
results = []
|
||||
for file_path, index_label in target_files:
|
||||
try:
|
||||
stats = file_path.stat()
|
||||
with Image.open(file_path) as img:
|
||||
# Apply EXIF orientation to fix sideways images
|
||||
img = ImageOps.exif_transpose(img)
|
||||
orig_w, orig_h = img.size
|
||||
target_w, target_h = calculate_patch_dimensions(orig_w, orig_h)
|
||||
|
||||
# Use thumbnail() to preserve aspect ratio and fit within the target bounding box.
|
||||
thumb = img.convert("RGB")
|
||||
thumb.thumbnail((target_w, target_h), Image.Resampling.LANCZOS)
|
||||
|
||||
buf = io.BytesIO()
|
||||
thumb.save(buf, format='JPEG', quality=config.IMAGE_QUALITY)
|
||||
img_data = base64.b64encode(buf.getvalue()).decode("utf-8")
|
||||
|
||||
# Gather info
|
||||
mtime_rel = format_relative_time(stats.st_mtime)
|
||||
mtime_abs = datetime.datetime.fromtimestamp(stats.st_mtime).strftime('%Y-%m-%d %H:%M')
|
||||
size_kb = stats.st_size / 1024
|
||||
|
||||
# Prepend index if this image was selected via index/range
|
||||
idx_prefix = f"[Index: {index_label}] " if index_label is not None else ""
|
||||
|
||||
info_text = (
|
||||
f"{idx_prefix}File: {file_path.name}\n"
|
||||
f"Resolution: {orig_w}x{orig_h}\n"
|
||||
f"Size: {size_kb:.1f}KB\n"
|
||||
f"Modified: {mtime_rel} ({mtime_abs})"
|
||||
)
|
||||
|
||||
results.append({"type": "text", "text": info_text})
|
||||
results.append({"type": "image", "data": img_data, "mimeType": "image/jpeg"})
|
||||
|
||||
except Exception as e:
|
||||
results.append({"type": "text", "text": f"Error processing {file_path.name}: {e}"})
|
||||
|
||||
return results
|
||||
+13
-12
@@ -1,29 +1,30 @@
|
||||
import os
|
||||
import base64
|
||||
import mimetypes
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
from tools.utils import ToolError
|
||||
|
||||
logger = logging.getLogger("MattCP")
|
||||
logger = logging.getLogger("MooseCP")
|
||||
|
||||
async def handle(args: Dict[str, Any]):
|
||||
"""
|
||||
Reads an image file from disk and returns it as a base64 encoded string
|
||||
within an MCP ImageContent object.
|
||||
"""
|
||||
path = args.get("path")
|
||||
if not path:
|
||||
return {"text": "Error: Missing path argument"}
|
||||
path_str = args.get("path")
|
||||
if not path_str:
|
||||
raise ToolError("Missing path argument")
|
||||
|
||||
path = Path(path_str)
|
||||
|
||||
# Validate that the file has an image-like MIME type
|
||||
mime_type, _ = mimetypes.guess_type(path)
|
||||
mime_type, _ = mimetypes.guess_type(path_str)
|
||||
if not mime_type or not mime_type.startswith("image/"):
|
||||
return {"text": f"Error: File {path} is not a supported image format."}
|
||||
raise ToolError(f"File {path_str} is not a supported image format.")
|
||||
|
||||
try:
|
||||
with open(path, "rb") as f:
|
||||
# Read raw bytes and encode to base64 for transport
|
||||
data = base64.b64encode(f.read()).decode("utf-8")
|
||||
return {"type": "image", "data": data, "mimeType": mime_type}
|
||||
data = base64.b64encode(path.read_bytes()).decode("utf-8")
|
||||
return {"type": "image", "data": data, "mimeType": mime_type}
|
||||
except Exception as e:
|
||||
return {"text": f"Error reading image: {str(e)}"}
|
||||
raise ToolError(f"Error reading image: {str(e)}")
|
||||
|
||||
+12
-7
@@ -1,24 +1,27 @@
|
||||
import os
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
from PIL import Image
|
||||
from tools.utils import ToolError
|
||||
|
||||
logger = logging.getLogger("MattCP")
|
||||
logger = logging.getLogger("MooseCP")
|
||||
|
||||
async def handle(args: Dict[str, Any]):
|
||||
"""
|
||||
Extracts text-based metadata chunks (tEXt, zTXt, iTXt) from a PNG file.
|
||||
These often contain AI generation prompts and parameters.
|
||||
"""
|
||||
path = args.get("path")
|
||||
if not path:
|
||||
return {"text": "Error: Missing path argument"}
|
||||
path_str = args.get("path")
|
||||
if not path_str:
|
||||
raise ToolError("Missing path argument")
|
||||
|
||||
path = Path(path_str)
|
||||
|
||||
try:
|
||||
with Image.open(path) as img:
|
||||
# Metadata extraction is specific to PNG format in this implementation
|
||||
if img.format != 'PNG':
|
||||
return {"text": "Error: File is not a PNG image."}
|
||||
raise ToolError("File is not a PNG image.")
|
||||
|
||||
# Pillow parses PNG text chunks into the .info dictionary
|
||||
metadata = img.info
|
||||
@@ -28,5 +31,7 @@ async def handle(args: Dict[str, Any]):
|
||||
# Format as a simple key: value list
|
||||
output = "\n".join([f"{k}: {v}" for k, v in metadata.items()])
|
||||
return {"text": f"PNG Metadata:\n{output}"}
|
||||
except ToolError:
|
||||
raise
|
||||
except Exception as e:
|
||||
return {"text": f"Error reading PNG metadata: {str(e)}"}
|
||||
raise ToolError(f"Error reading PNG metadata: {str(e)}")
|
||||
|
||||
@@ -1,5 +1,12 @@
|
||||
import time
|
||||
import datetime
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
|
||||
class ToolError(Exception):
|
||||
"""Custom exception for tool-related errors to be caught by the MCP server."""
|
||||
pass
|
||||
|
||||
def format_relative_time(timestamp: float) -> str:
|
||||
"""Converts a timestamp to a human-readable relative format."""
|
||||
@@ -32,3 +39,72 @@ def format_relative_time(timestamp: float) -> str:
|
||||
|
||||
dt = datetime.datetime.fromtimestamp(timestamp)
|
||||
return dt.strftime('%Y-%m-%d')
|
||||
|
||||
|
||||
def get_file_info_list(path: Path, extensions: Optional[tuple] = None) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Gathers basic information about files in a directory.
|
||||
Optional extensions filter (e.g. ('.png', '.jpg')).
|
||||
"""
|
||||
items = []
|
||||
try:
|
||||
for entry in path.iterdir():
|
||||
if extensions and not entry.name.lower().endswith(extensions):
|
||||
continue
|
||||
|
||||
stats = entry.stat()
|
||||
items.append({
|
||||
"name": entry.name,
|
||||
"path": entry,
|
||||
"mtime": stats.st_mtime,
|
||||
"size": stats.st_size,
|
||||
"is_dir": entry.is_dir()
|
||||
})
|
||||
except Exception as e:
|
||||
raise ToolError(f"Error accessing directory {path}: {e}")
|
||||
|
||||
return items
|
||||
|
||||
def sort_file_list(items: List[Dict[str, Any]], sort_by: str) -> List[Dict[str, Any]]:
|
||||
"""Sorts items based on the provided key. Directories always come first."""
|
||||
if sort_by == "mtime":
|
||||
# Newest first
|
||||
items.sort(key=lambda x: (not x["is_dir"], -x["mtime"]))
|
||||
elif sort_by == "size":
|
||||
# Largest first
|
||||
items.sort(key=lambda x: (not x["is_dir"], -x["size"]))
|
||||
else:
|
||||
# Alphabetical
|
||||
items.sort(key=lambda x: (not x["is_dir"], x["name"].lower()))
|
||||
return items
|
||||
|
||||
def get_paginated_list(items: List[Any], page: int, page_size: int) -> List[Any]:
|
||||
"""Returns a slice of the list based on page and page_size."""
|
||||
start_idx = (page - 1) * page_size
|
||||
end_idx = start_idx + page_size
|
||||
return items[start_idx:end_idx]
|
||||
|
||||
def parse_indices(indices_str: str) -> List[int]:
|
||||
"""
|
||||
Parses a string like '11, 31-33, 44' into a list of 1-based indices.
|
||||
Preserves the order specified in the string.
|
||||
"""
|
||||
indices = []
|
||||
parts = [p.strip() for p in indices_str.split(',')]
|
||||
for part in parts:
|
||||
if not part:
|
||||
continue
|
||||
if '-' in part:
|
||||
try:
|
||||
start, end = map(int, part.split('-'))
|
||||
step = 1 if start <= end else -1
|
||||
for i in range(start, end + step, step):
|
||||
indices.append(i)
|
||||
except ValueError:
|
||||
raise ToolError(f"Invalid range format: {part}")
|
||||
else:
|
||||
try:
|
||||
indices.append(int(part))
|
||||
except ValueError:
|
||||
raise ToolError(f"Invalid index format: {part}")
|
||||
return indices
|
||||
|
||||
Reference in New Issue
Block a user