[Penify]: Documentation for commit - ea03621

agi-pipeline.py

@@ -1,51 +1,51 @@
 # pylint: disable=import-error, wrong-import-position, wrong-import-order, missing-function-docstring, missing-class-docstring, broad-exception-caught, logging-fstring-interpolation, too-few-public-methods, no-member, unused-import, unused-variable, unused-argument, invalid-name, unnecessary-lambda, useless-parent-delegation, too-many-instance-attributes
 """
 AGI Pipeline Legacy Module.
 """
 
 from google.colab import drive
 
 # Mount Google Drive
 drive.mount("/content/drive")
 
 import logging
 import os
 
 import albumentations as A
 import cv2
 import matplotlib.pyplot as plt
 import numpy as np
 import plotly.express as px
 import pyttsx3
 import seaborn as sns
 import speech_recognition as sr
 import torch
 import uvicorn
 from celery import Celery
 from fastapi import Depends, FastAPI, File, UploadFile
 from fastapi.security import OAuth2PasswordBearer
 from gym import Env
 from gym.spaces import Box, Discrete
 from PIL import Image
 from stable_baselines3 import PPO
 from stable_baselines3.common.vec_env import DummyVecEnv
 from torchvision import models, transforms
 from transformers import AutoModelForSeq2SeqLM, AutoTokenizer, CLIPModel, CLIPProcessor
 
 # Hugging Face Authentication (Optional)
 HF_TOKEN = os.environ.get("HF_TOKEN", None)
 
 # Setting up logging
 logging.basicConfig(level=logging.INFO)
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
 class NLPModule:
     """
     Class NLPModule.
     """
 
     def __init__(self, model_name="facebook/bart-large-cnn"):
         """
         Method __init__.
@@ -58,130 +58,124 @@
         )  # nosec B615
 
     def process_text(self, text, max_length=25, num_beams=5):
-        """
-        Method process_text.
-        """
+        """Process and summarize the given text using a model."""
         logging.info("Processing text for summarization")
         try:
             inputs = self.tokenizer(
                 text, return_tensors="pt", max_length=512, truncation=True
             )
             outputs = self.model.generate(
                 inputs["input_ids"],
                 max_length=max_length,
                 min_length=10,
                 num_beams=num_beams,
             )
             return self.tokenizer.decode(outputs[0], skip_special_tokens=True)
         except Exception as e:
             logging.error(f"Error in NLPModule: {e}")
             return "NLP processing error"
 
 
 class CVModule:
     """
     Class CVModule.
     """
 
     def __init__(self):
         """
         Method __init__.
         """
         self.model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1)
         self.model.eval()
         self.transform = transforms.Compose(
             [
                 transforms.Resize((224, 224)),
                 transforms.RandomHorizontalFlip(),
                 transforms.ColorJitter(
                     brightness=0.5, contrast=0.5, saturation=0.5, hue=0.5
                 ),
                 transforms.ToTensor(),
                 transforms.Normalize(
                     mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]
                 ),
             ]
         )
 
     @staticmethod
     def preprocess_large_image(image_path, max_size=(2000, 2000)):
         """
         Method preprocess_large_image.
         """
         try:
             with Image.open(image_path) as img:
                 img.thumbnail(max_size)
                 resized_path = "resized_image.jpg"
                 img.save(resized_path)
             return resized_path
         except Exception as e:
             logging.error(f"Error in preprocessing image: {e}")
             return None
 
     def process_image(self, image_path):
-        """
-        Method process_image.
-        """
+        """Process an image for classification."""
         logging.info("Processing image for classification")
         try:
             image_path = self.preprocess_large_image(
                 image_path
             )  # Ensure the image is manageable
             image = Image.open(image_path).convert("RGB")
             tensor = self.transform(image).unsqueeze(0)
             with torch.no_grad():
                 outputs = self.model(tensor)
             return outputs.argmax().item()
         except Exception as e:
             logging.error(f"Error in CVModule: {e}")
             return "CV processing error"
 
 
 class AdvancedDataAugmentation(CVModule):
     """
     Class AdvancedDataAugmentation.
     """
 
     def __init__(self):
         """
         Method __init__.
         """
         super().__init__()
         self.aug = A.Compose(
             [
                 A.HorizontalFlip(p=0.5),
                 A.RandomBrightnessContrast(p=0.5),
                 A.Rotate(limit=40, p=0.5),
             ]
         )
 
     def process_image(self, image_path):
-        """
-        Method process_image.
-        """
+        """Process an image for classification with augmentation."""
         logging.info("Processing image with augmentation for classification")
         try:
             image_path = self.preprocess_large_image(
                 image_path
             )  # Ensure the image is manageable
             image = Image.open(image_path).convert("RGB")
             image = np.array(image)
             augmented = self.aug(image=image)
             image = augmented["image"]
             tensor = self.transform(image).unsqueeze(0)
             with torch.no_grad():
                 outputs = self.model(tensor)
             return outputs.argmax().item()
         except Exception as e:
             logging.error(f"Error in AdvancedDataAugmentation: {e}")
             return "CV processing error"
 
 
 class MultiModalModule:
     """
     Class MultiModalModule.
     """
 
     def __init__(self, model_name="openai/clip-vit-base-patch32"):
         """
         Method __init__.
@@ -197,26 +191,26 @@
         """
         Method process_text_image.
         """
         logging.info("Processing text and image for multi-modal integration")
         try:
             image_path = CVModule.preprocess_large_image(image_path)
             image = Image.open(image_path)
             inputs = self.processor(
                 text=[text], images=[image], return_tensors="pt", padding=True
             )
             outputs = self.model(**inputs)
             logits_per_image = outputs.logits_per_image
             return logits_per_image.softmax(dim=1)
         except Exception as e:
             logging.error(f"Error in MultiModalModule: {e}")
             return "Multi-modal processing error"
 
 
 class CustomEnv(Env):
     """
     Class CustomEnv.
     """
 
     def __init__(self):
         """
         Method __init__.
@@ -227,16 +221,12 @@
         self.state = 50
 
     def reset(self):
-        """
-        Method reset.
-        """
+        """Resets the state to 50 and returns it as a numpy array."""
         self.state = 50
         return np.array([self.state], dtype=np.float32)
 
     def step(self, action):
-        """
-        Method step.
-        """
+        """Executes a step in the environment based on the given action."""
         reward = -abs(self.state - (50 + action * 10))
         self.state += action - 2
         done = self.state <= 0 or self.state >= 100
@@ -256,9 +246,7 @@
         self.model = PPO("MlpPolicy", self.env, verbose=1)
 
     def train(self, timesteps=10000):
-        """
-        Method train.
-        """
+        """Trains the RL model for a specified number of timesteps."""
         logging.info("Training RL model")
         try:
             self.model.learn(total_timesteps=timesteps)
@@ -267,9 +255,7 @@
             logging.error(f"Error in RLModule training: {e}")
 
     def save_model(self, path):
-        """
-        Method save_model.
-        """
+        """Saves the model to the specified path."""
         try:
             self.model.save(path)
             logging.info(f"Model saved to {path}")
@@ -341,9 +327,7 @@
         return frame_count
 
     def process_frame(self, frame_path):
-        """
-        Method process_frame.
-        """
+        """Processes an image frame and returns a tensor."""
         try:
             image = Image.open(frame_path).convert("RGB")
             tensor = self.transform(image).unsqueeze(0)
@@ -365,8 +349,17 @@
         super().__init__()
 
     def process_real_time_video(self, source=0):
-        """
-        Method process_real_time_video.
+        """Process real-time video from a specified source.
+
+        This method captures video from the given source and processes each frame  in
+        real-time. It checks if the video source is opened successfully, and if  not,
+        logs an error. The frames are resized and transformed before being  displayed
+        in a window. The processing continues until the video ends or  the user presses
+        the 'q' key to quit. Finally, it releases the video  capture and closes all
+        OpenCV windows.
+
+        Args:
+            source (int or str): The video source, which can be an integer for
         """
         cap = cv2.VideoCapture(source)
         if not cap.isOpened():
@@ -401,9 +394,7 @@
         self.engine = pyttsx3.init()
 
     def speech_to_text(self, audio_file):
-        """
-        Method speech_to_text.
-        """
+        """Converts speech from an audio file to text."""
         try:
             with sr.AudioFile(audio_file) as source:
                 audio = self.recognizer.record(source)
@@ -443,9 +434,7 @@
         self.voice_processor = VoiceProcessor()
 
     def process_input(self, text=None, image_path=None):
-        """
-        Method process_input.
-        """
+        """Processes text and image input and returns the results."""
         results = {}
         if text:
             results["nlp"] = self.nlp.process_text(text)
@@ -454,25 +443,19 @@
         return results
 
     def process_multi_modal(self, text, image_path):
-        """
-        Method process_multi_modal.
-        """
+        """Processes text and image using multi-modal processing."""
         return self.multi_modal.process_text_image(text, image_path)
 
     def process_video(self, video_path, frame_output_dir):
-        """
-        Method process_video.
-        """
+        """Process a video and extract its frames."""
         frame_count = self.video_processor.extract_frames(video_path, frame_output_dir)
         if frame_count == 0:
             logging.error("No frames were saved. Please check the video file and path.")
             return
         logging.info(f"Video frames processed and saved to {frame_output_dir}")
 
     def process_real_time_video(self, source=0):
-        """
-        Method process_real_time_video.
-        """
+        """Processes real-time video from the specified source."""
         self.real_time_video_processor.process_real_time_video(source)
 
     def train_rl(self, timesteps=10000):
@@ -482,15 +465,11 @@
         self.rl.train(timesteps)
 
     def choose_action(self, state):
-        """
-        Method choose_action.
-        """
+        """Selects an action based on the given state."""
         return self.rl.choose_action(state)
 
     def visualize_data(self, data):
-        """
-        Method visualize_data.
-        """
+        """Visualizes the given data using a bar chart."""
         try:
             fig = px.bar(
                 x=list(data.keys()), y=list(data.values()), title="Data Visualization"
@@ -500,15 +479,11 @@
             logging.error(f"Error in data visualization: {e}")
 
     def speech_to_text(self, audio_file):
-        """
-        Method speech_to_text.
-        """
+        """Converts speech from an audio file to text."""
         return self.voice_processor.speech_to_text(audio_file)
 
     def text_to_speech(self, text):
-        """
-        Method text_to_speech.
-        """
+        """Converts text to speech using the voice processor."""
         self.voice_processor.text_to_speech(text)
 
 

cv_module.py

@@ -22,15 +22,14 @@ def __init__(self):
         logger.info("CV model loaded successfully.")
 
     def detect_objects(self, image: Image.Image) -> str:
-        """
-        Detects objects in the provided image.
-
+        """Detects objects in the provided image.
+
         Args:
             image (Image.Image): The input image.
-
+        
         Returns:
             str: JSON string containing detection results.
-
+        
         Raises:
             ValueError: If the image is None.
         """

fix_agi.py

@@ -1,8 +1,9 @@
 import re
 import sys
 
 def add_docstrings(content):
     # Add module docstring if missing
+    """Add missing module and class/method/function docstrings to the content."""
     if not content.startswith('"""'):
         content = '"""\nAGI Pipeline Legacy Module.\n"""\n' + content
 
@@ -17,7 +18,7 @@
 
     return content
 
 with open('agi-pipeline.py', 'r') as f:
     lines = f.readlines()
 
 # Remove my previous disable line
@@ -27,5 +28,5 @@
 content = ''.join(lines)
 content = add_docstrings(content)
 
 with open('agi-pipeline.py', 'w') as f:
     f.write(content)

main.py

@@ -33,21 +33,15 @@ def __init__(self):
         self.speech = SpeechProcessor()
 
     def process_nlp(self, prompt: str) -> str:
-        """
-        Processes text using the NLP module.
-        """
+        """Processes text using the NLP module."""
         return self.nlp.generate_text(prompt)
 
     def process_cv(self, image: Image.Image) -> str:
-        """
-        Processes an image using the CV module.
-        """
+        """Processes an image to detect objects using the CV module."""
         return self.cv.detect_objects(image)
 
     def process_stt(self, file: UploadFile) -> str:
-        """
-        Processes audio using the STT module.
-        """
+        """Processes audio and converts it to text using the STT module."""
         return self.speech.speech_to_text(file)
 
     def process_tts(self, text: str) -> None:
@@ -62,18 +56,14 @@ def process_tts(self, text: str) -> None:
 
 
 def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
-    """
-    Verifies the Bearer token in the Authorization header.
-    """
+    """Verifies the Bearer token in the Authorization header."""
     if credentials.credentials != VALID_API_KEY:
         raise HTTPException(status_code=403, detail="Forbidden")
 
 
 @app.post("/process-nlp/", dependencies=[Depends(verify_token)])
 async def process_nlp(data: dict):
-    """
-    Endpoint for text generation.
-    """
+    """Processes natural language input and generates a response."""
     try:
         prompt = data.get("text", "")
         return {"response": agi.process_nlp(prompt)}
@@ -84,9 +74,7 @@ async def process_nlp(data: dict):
 
 @app.post("/process-cv-detection/", dependencies=[Depends(verify_token)])
 async def process_cv_detection(file: UploadFile = File(...)):
-    """
-    Endpoint for object detection in images.
-    """
+    """Handles object detection in uploaded images."""
     try:
         image_data = await file.read()
         image = Image.open(BytesIO(image_data))
@@ -98,9 +86,7 @@ async def process_cv_detection(file: UploadFile = File(...)):
 
 @app.post("/speech-to-text/", dependencies=[Depends(verify_token)])
 async def speech_to_text(file: UploadFile = File(...)):
-    """
-    Endpoint for Speech-to-Text conversion.
-    """
+    """Converts speech from an uploaded file to text."""
     try:
         return {"response": agi.process_stt(file)}
     except Exception as e:
@@ -110,9 +96,7 @@ async def speech_to_text(file: UploadFile = File(...)):
 
 @app.post("/text-to-speech/", dependencies=[Depends(verify_token)])
 async def text_to_speech(data: dict):
-    """
-    Endpoint for Text-to-Speech conversion.
-    """
+    """Converts text to speech."""
     try:
         text = data.get("text", "")
         agi.process_tts(text)

nlp_module.py

@@ -27,15 +27,11 @@ def __init__(self):
         logger.info("NLP model loaded successfully.")
 
     def generate_text(self, prompt: str) -> str:
-        """
-        Generates text based on the provided prompt.
-
+        """Generates text based on the provided prompt.
+
         Args:
             prompt (str): The input text to process.
-
-        Returns:
-            str: The generated response.
-
+
         Raises:
             ValueError: If the prompt is empty.
         """