"""MultiCoder CLI module.

This module provides the command-line interface for interacting

with the MultiCoder system.

"""

import argparse

import asyncio

import json

import logging

import os

import sys

import uuid

from typing import Any, Dict, Optional

# Ajouter le répertoire parent au chemin de recherche de Python

sys.path.insert(0, os.path.abspath(os.path.dirname(os.path.dirname(__file__))))

import redis.asyncio as redis

class MultiCoderCLI:

"""Command-line interface for MultiCoder.

Provides a user-friendly CLI for submitting code generation requests

and displaying results.

Attributes:

redis_url: Redis connection string.

request_channel: Channel to publish requests.

response_channel: Channel to subscribe for responses.

logger: Configured logger instance.

"""

def __init__(

self,

redis_url: str = "redis://localhost:6379/0",

log_level: int = logging.INFO

) -> None:

"""Initialize the CLI client.

Args:

redis_url: Redis connection string.

log_level: Logging level.

"""

self.redis_url = redis_url

self.request_channel = "multicoder:requests"

self.response_channel = "multicoder:responses"

# Setup logging

self.logger = logging.getLogger("multicoder.cli")

self.logger.setLevel(log_level)

if not self.logger.handlers:

handler = logging.StreamHandler()

formatter = logging.Formatter(

'%(asctime)s - %(name)s - %(levelname)s - %(message)s'

)

handler.setFormatter(formatter)

self.logger.addHandler(handler)

# Redis connection will be initialized in the run method

self.redis_client = None

self.pubsub = None

async def send_request(self, prompt: str) -> str:

"""Send a code generation request and wait for response.

Args:

prompt: Natural language prompt describing the code to generate.

Returns:

Request ID for tracking.

"""

request_id = str(uuid.uuid4())

# Create request message

message = {

"sender": "cli",

"action": "process",

"payload": {

"request_id": request_id,

"prompt": prompt

}

}

# Publish request

message_json = json.dumps(message)

await self.redis_client.publish(self.request_channel, message_json)

self.logger.info(f"Sent request {request_id}: {prompt}")

return request_id

async def wait_for_response(self, request_id: str, timeout: int = 60) -> Optional[Dict[str, Any]]:

"""Wait for a response to a specific request.

Args:

request_id: The ID of the request to wait for.

timeout: Maximum time to wait in seconds.

Returns:

Response data or None if timed out.

"""

self.logger.info(f"Waiting for response to request {request_id}")

# Subscribe to responses

await self.pubsub.subscribe(self.response_channel)

# Set timeout

end_time = asyncio.get_event_loop().time() + timeout

try:

while asyncio.get_event_loop().time() < end_time:

message = await self.pubsub.get_message(timeout=1)

if message and message["type"] == "message":

data = json.loads(message["data"].decode("utf-8"))

response_id = data.get("payload", {}).get("request_id")

if response_id == request_id:

self.logger.info(f"Received response for request {request_id}")

return data.get("payload", {})

# Small delay to prevent CPU spinning

await asyncio.sleep(0.1)

self.logger.warning(f"Timed out waiting for response to {request_id}")

return None

finally:

# Unsubscribe from responses

await self.pubsub.unsubscribe(self.response_channel)

async def process_prompt(self, prompt: str, timeout: int = 60) -> None:

"""Process a prompt and display the results.

Args:

prompt: The prompt to process.

timeout: Maximum time to wait for response.

"""

try:

# Send request

request_id = await self.send_request(prompt)

# Wait for and display progress

print(f"Processing request: {prompt}")

print("Waiting for response...", end="", flush=True)

# Wait for response

response = await self.wait_for_response(request_id, timeout)

# Display results

print("\r" + " " * 50 + "\r", end="") # Clear waiting message

if response:

result = response.get("result", {})

status = response.get("status", "unknown")

if status == "completed":

print("\n✅ Code generation successful!\n")

code = result.get("code", "")

print("-------- Generated Code --------")

print(code)

print("--------------------------------")

else:

print("\n❌ Code generation failed!\n")

issues = result.get("issues", [])

if issues:

print("Issues:")

for issue in issues:

print(f" - {issue}")

code = result.get("code", "")

if code:

print("\nPartial code:")

print("-------- Generated Code --------")

print(code)

print("--------------------------------")

else:

print("\n⚠️ Request timed out. The system might be busy or experiencing issues.")

except Exception as e:

self.logger.error(f"Error processing prompt: {e}", exc_info=True)

print(f"\n❌ Error: {e}")

async def interactive_mode(self) -> None:

"""Run the CLI in interactive mode."""

print("\n🤖 MultiCoder Interactive Mode")

print("Type 'exit' or 'quit' to exit\n")

while True:

try:

prompt = input("🔍 Enter your code request: ")

if prompt.lower() in ["exit", "quit"]:

break

if prompt:

await self.process_prompt(prompt)

print("\n")

except KeyboardInterrupt:

print("\nExiting...")

break

except Exception as e:

self.logger.error(f"Error in interactive mode: {e}", exc_info=True)

print(f"❌ Error: {e}")

async def run(self, prompt: Optional[str] = None) -> None:

"""Run the CLI client.

Args:

prompt: Optional prompt to process in non-interactive mode.

"""

try:

# Initialize Redis connection

self.redis_client = redis.Redis.from_url(self.redis_url)

self.pubsub = self.redis_client.pubsub()

if prompt:

# Process a single prompt

await self.process_prompt(prompt)

else:

# Interactive mode

await self.interactive_mode()

finally:

# Clean up

if self.pubsub:

await self.pubsub.aclose()

if self.redis_client:

await self.redis_client.aclose()

def parse_args() -> argparse.Namespace:

"""Parse command line arguments.

Returns:

Parsed arguments.

"""

parser = argparse.ArgumentParser(description="MultiCoder: AI-powered code generation")

parser.add_argument(

"--prompt", "-p",

type=str,

help="Prompt for code generation (non-interactive mode)"

)

parser.add_argument(

"--redis-url",

type=str,

default="redis://localhost:6379/0",

help="Redis connection URL (default: redis://localhost:6379/0)"

)

parser.add_argument(

"--log-level",

choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],

default="INFO",

help="Set logging level (default: INFO)"

)

return parser.parse_args()

async def main() -> None:

"""Main entry point for the CLI."""

args = parse_args()

# Configure logging

log_level = getattr(logging, args.log_level)

logging.basicConfig(

level=log_level,

format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"

)

# Run CLI

cli = MultiCoderCLI(redis_url=args.redis_url, log_level=log_level)

await cli.run(args.prompt)

if __name__ == "__main__":

asyncio.run(main())