refactor: ruff linting

Author: pmannion2

api.py

@@ -183,9 +183,7 @@ def get_collection():
         try:
             collection = client.get_collection(COLLECTION_NAME)
             doc_count = collection.count()
-            logger.info(
-                f"Connected to collection '{COLLECTION_NAME}' with {doc_count} documents"
-            )
+            logger.info(f"Connected to collection '{COLLECTION_NAME}' with {doc_count} documents")
         except ValueError:
             logger.info(f"Creating new collection '{COLLECTION_NAME}'")
             collection = client.create_collection(COLLECTION_NAME)
@@ -361,9 +359,7 @@ async def query(request: QueryRequest):
             "metadata": {
                 "total_chunks": collection.count(),
                 "query": request.query,
-                "embedding_type": (
-                    "mock" if MOCK_EMBEDDINGS or not openai_client else "openai"
-                ),
+                "embedding_type": ("mock" if MOCK_EMBEDDINGS or not openai_client else "openai"),
             },
         }
 
@@ -401,14 +397,10 @@ class AgentQueryResponse(BaseModel):
 
 
 @app.post("/agent/query", response_model=AgentQueryResponse)
-async def agent_query(
-    req: AgentQueryRequest, deps: None = Depends(verify_dependencies)
-):
+async def agent_query(req: AgentQueryRequest, deps: None = Depends(verify_dependencies)):
     """Agent-optimized query endpoint for Cursor integration."""
     start_time = time.time()
-    logger.info(
-        f"Agent query received: '{req.query}', top_k={req.top_k}, context={req.context}"
-    )
+    logger.info(f"Agent query received: '{req.query}', top_k={req.top_k}, context={req.context}")
 
     # Check if mock mode is requested for this query
     use_mock = MOCK_EMBEDDINGS
@@ -523,9 +515,7 @@ async def agent_query(
                 suggested_prompt += f"\n--- Context {i} ({chunk['source']}) ---\n"
                 suggested_prompt += f"{chunk['content']}\n"
 
-            suggested_prompt += (
-                "\nBased on the above context, please help with the query."
-            )
+            suggested_prompt += "\nBased on the above context, please help with the query."
             logger.debug(f"Generated prompt in {time.time() - prompt_start:.2f}s")
 
         response_time = time.time() - start_time
@@ -623,9 +613,7 @@ class IndexRequest(BaseModel):
     project_path: str = Field(
         "./whk-ignition-scada", description="Path to the Ignition project directory"
     )
-    rebuild: bool = Field(
-        False, description="Whether to rebuild the index from scratch"
-    )
+    rebuild: bool = Field(False, description="Whether to rebuild the index from scratch")
     skip_rate_limiting: bool = Field(
         False, description="Skip rate limiting for faster processing (use with caution)"
     )
@@ -713,9 +701,7 @@ async def generate_embedding_with_backoff(text, max_retries=5, initial_backoff=1
                 ):
                     retries += 1
                     if retries > max_retries:
-                        logger.error(
-                            f"Max retries reached for rate limit. Final error: {e!s}"
-                        )
+                        logger.error(f"Max retries reached for rate limit. Final error: {e!s}")
                         raise
 
                     logger.info(
@@ -781,7 +767,7 @@ def chunk_by_characters(text, max_chunk_size):
     for file_index, file_path in enumerate(json_files):
         file_start_time = time.time()
         try:
-            logger.info(f"Processing {file_path}... [{file_index+1}/{total_files}]")
+            logger.info(f"Processing {file_path}... [{file_index + 1}/{total_files}]")
             with open(file_path, encoding="utf-8") as f:
                 content = f.read()
 
@@ -844,10 +830,7 @@ def chunk_by_characters(text, max_chunk_size):
                                 )
 
                                 # For array-type JSONs, split at the top level
-                                if (
-                                    isinstance(json_content, list)
-                                    and len(json_content) > 1
-                                ):
+                                if isinstance(json_content, list) and len(json_content) > 1:
                                     logger.info(
                                         f"Using array-level chunking for JSON array with {len(json_content)} items"
                                     )
@@ -876,10 +859,7 @@ def chunk_by_characters(text, max_chunk_size):
                                             )
                                             sub_chunks.extend(item_chunks)
                                         # If adding this would exceed limit, create a new chunk
-                                        elif (
-                                            current_tokens + item_tokens
-                                            > hard_token_limit
-                                        ):
+                                        elif current_tokens + item_tokens > hard_token_limit:
                                             array_str = json.dumps(current_array)
                                             sub_chunks.append(array_str)
                                             current_array = [item]
@@ -901,9 +881,7 @@ def chunk_by_characters(text, max_chunk_size):
                                         content,
                                         int(hard_token_limit / 1.2),
                                     )
-                                    chunks = [
-                                        (chunk, metadata) for chunk in text_chunks
-                                    ]
+                                    chunks = [(chunk, metadata) for chunk in text_chunks]
                             except json.JSONDecodeError:
                                 # If JSON parsing fails, use character-level chunking
                                 text_chunks = chunk_by_characters(
@@ -928,14 +906,10 @@ def chunk_by_characters(text, max_chunk_size):
                     for i, (chunk_text, chunk_metadata) in enumerate(chunks):
                         try:
                             # Generate embedding with backoff
-                            embedding = await generate_embedding_with_backoff(
-                                chunk_text
-                            )
+                            embedding = await generate_embedding_with_backoff(chunk_text)
 
                             # Create a unique ID for this chunk
-                            file_path_replaced = file_path.replace("/", "_").replace(
-                                "\\", "_"
-                            )
+                            file_path_replaced = file_path.replace("/", "_").replace("\\", "_")
                             chunk_id = f"{file_path_replaced}_chunk_{i}"
 
                             # Add to collection
@@ -948,9 +922,7 @@ def chunk_by_characters(text, max_chunk_size):
 
                             chunk_count += 1
                         except Exception as e:
-                            logger.error(
-                                f"Error processing chunk {i} of {file_path}: {e!s}"
-                            )
+                            logger.error(f"Error processing chunk {i} of {file_path}: {e!s}")
 
                     doc_count += 1
                     logger.info(f"Indexed {file_path} into {len(chunks)} chunks")

cursor_agent.py

@@ -146,7 +146,9 @@ def get_cursor_context(
             context_str += f"--- Context {i}: {source} ---\n"
             context_str += f"{content}\n\n"
 
-        context_str += "Use the above context to help answer the query or generate appropriate code.\n"
+        context_str += (
+            "Use the above context to help answer the query or generate appropriate code.\n"
+        )
         return context_str
 
     # No context available
@@ -182,9 +184,7 @@ def get_ignition_tag_info(tag_name: str) -> dict:
         }
 
     # Get tag information from the RAG system
-    rag_results = query_rag(
-        query=f"Tag configuration for {tag_name}", top_k=1, filter_type="tag"
-    )
+    rag_results = query_rag(query=f"Tag configuration for {tag_name}", top_k=1, filter_type="tag")
 
     # Extract tag info from the context
     context_chunks = rag_results.get("context_chunks", [])
@@ -208,9 +208,7 @@ def get_ignition_tag_info(tag_name: str) -> dict:
     return {"error": f"Could not parse tag information for {tag_name}"}
 
 
-def get_ignition_view_component(
-    view_name: str, component_name: Optional[str] = None
-) -> dict:
+def get_ignition_view_component(view_name: str, component_name: Optional[str] = None) -> dict:
     """
     Get information about a specific view or component in an Ignition project.
 
@@ -223,9 +221,7 @@ def get_ignition_view_component(
     """
     # Check if mock mode is enabled
     if USE_MOCK_EMBEDDINGS:
-        logger.info(
-            f"Using mock data for view: {view_name}, component: {component_name}"
-        )
+        logger.info(f"Using mock data for view: {view_name}, component: {component_name}")
         # Create a mock response
         if component_name:
             return {
@@ -238,9 +234,7 @@ def get_ignition_view_component(
                     "width": 200,
                     "height": 150,
                     "text": (
-                        f"Mock {component_name}"
-                        if "label" in component_name.lower()
-                        else None
+                        f"Mock {component_name}" if "label" in component_name.lower() else None
                     ),
                 },
                 "mock_used": True,
@@ -281,11 +275,7 @@ def get_ignition_view_component(
             content_obj = json.loads(content.strip())
 
             # For component search
-            if (
-                component_name
-                and "name" in content_obj
-                and content_obj["name"] == component_name
-            ):
+            if component_name and "name" in content_obj and content_obj["name"] == component_name:
                 return content_obj
 
             # For view search

cursor_client.py

@@ -68,7 +68,7 @@ def get_rag_context(query, current_file=None, top_k=3, filter_type=None):
             metadata = chunk.get("metadata", {})
             file_path = metadata.get("filepath", "Unknown file")
 
-            context_text += f"## Context {i+1}: {source} ({file_path})\n"
+            context_text += f"## Context {i + 1}: {source} ({file_path})\n"
             context_text += "```json\n"
             context_text += content + "\n"
             context_text += "```\n\n"
@@ -101,13 +101,9 @@ def main():
     parser = argparse.ArgumentParser(description="Cursor Client for Ignition RAG")
     parser.add_argument("query", help="The query to search for")
     parser.add_argument("--file", "-f", help="Path to the current file")
-    parser.add_argument(
-        "--top-k", "-k", type=int, default=3, help="Number of results to return"
-    )
+    parser.add_argument("--top-k", "-k", type=int, default=3, help="Number of results to return")
     parser.add_argument("--filter", help="Filter by document type (perspective or tag)")
-    parser.add_argument(
-        "--output", "-o", help="Output format (text or json)", default="text"
-    )
+    parser.add_argument("--output", "-o", help="Output format (text or json)", default="text")
 
     args = parser.parse_args()
 

indexer.py

@@ -167,9 +167,7 @@ def load_json_files(file_paths: List[str]) -> List[Dict[str, Any]]:
     return documents
 
 
-def chunk_perspective_view(
-    view_json: Dict[str, Any], view_meta: Dict[str, str]
-) -> List[tuple]:
+def chunk_perspective_view(view_json: Dict[str, Any], view_meta: Dict[str, str]) -> List[tuple]:
     """Split a Perspective view JSON into semantically meaningful chunks."""
     chunks = []
 
@@ -230,9 +228,7 @@ def process_component(comp, view_meta, chunks, parent_path=""):
             comp_copy = {k: v for k, v in comp.items() if k != "children"}
             comp_without_children = json.dumps(comp_copy, ensure_ascii=False)
             if len(enc.encode(comp_without_children)) <= MAX_TOKENS:
-                chunks.append(
-                    (comp_without_children, {**comp_meta, "section": "properties"})
-                )
+                chunks.append((comp_without_children, {**comp_meta, "section": "properties"}))
         else:
             # Split properties if no children but still too large
             props = list(comp.items())
@@ -341,13 +337,11 @@ def generate_embeddings(texts: List[str], batch_size: int = 20) -> List[List[flo
         batch = texts[i : i + batch_size]
         try:
             # Updated for OpenAI v1.0+
-            response = client.embeddings.create(
-                model="text-embedding-ada-002", input=batch
-            )
+            response = client.embeddings.create(model="text-embedding-ada-002", input=batch)
             batch_embeddings = [item.embedding for item in response.data]
             embeddings.extend(batch_embeddings)
             print(
-                f"Generated embeddings for batch {i//batch_size + 1}/{(len(texts) + batch_size - 1) // batch_size}"
+                f"Generated embeddings for batch {i // batch_size + 1}/{(len(texts) + batch_size - 1) // batch_size}"
             )
         except Exception as e:
             print(f"Error generating embeddings for batch starting at index {i}: {e}")
@@ -422,15 +416,11 @@ def load_last_index_time() -> float:
 @app.command()
 def main(
     path: str = typer.Argument(..., help="Path to the Ignition project directory"),
-    rebuild: bool = typer.Option(
-        False, "--rebuild", help="Rebuild the index from scratch"
-    ),
+    rebuild: bool = typer.Option(False, "--rebuild", help="Rebuild the index from scratch"),
     changed_only: bool = typer.Option(
         False, "--changed-only", help="Only index files changed since last run"
     ),
-    file: Optional[str] = typer.Option(
-        None, "--file", help="Index only a specific file"
-    ),
+    file: Optional[str] = typer.Option(None, "--file", help="Index only a specific file"),
     mock: bool = typer.Option(
         False,
         "--mock",
@@ -468,9 +458,7 @@ def main(
     elif changed_only and not rebuild:
         # Index only files changed since last run
         last_index_time = load_last_index_time()
-        json_files = [
-            f for f in all_json_files if os.path.getmtime(f) > last_index_time
-        ]
+        json_files = [f for f in all_json_files if os.path.getmtime(f) > last_index_time]
         print(
             f"Found {len(json_files)} changed files since {datetime.fromtimestamp(last_index_time)}"
         )

logger.py

@@ -26,14 +26,18 @@
 
 # Add file handler for general logs
 file_handler = RotatingFileHandler(
-    os.path.join(LOG_DIR, "app.log"), maxBytes=10485760, backupCount=10  # 10MB
+    os.path.join(LOG_DIR, "app.log"),
+    maxBytes=10485760,
+    backupCount=10,  # 10MB
 )
 file_handler.setFormatter(log_format)
 logger.addHandler(file_handler)
 
 # Add file handler for errors only
 error_handler = RotatingFileHandler(
-    os.path.join(LOG_DIR, "error.log"), maxBytes=10485760, backupCount=10  # 10MB
+    os.path.join(LOG_DIR, "error.log"),
+    maxBytes=10485760,
+    backupCount=10,  # 10MB
 )
 error_handler.setLevel(logging.ERROR)
 error_handler.setFormatter(log_format)
@@ -69,7 +73,5 @@ async def __call__(self, scope, receive, send):
             await self.app(scope, receive, send)
             self.logger.info(f"Response {request_id}: {method} {path} completed")
         except Exception as e:
-            self.logger.error(
-                f"Error {request_id}: {method} {path} - {e!s}", exc_info=True
-            )
+            self.logger.error(f"Error {request_id}: {method} {path} - {e!s}", exc_info=True)
             raise