Clean up module and package display names
This commit is contained in:
@@ -1,8 +1,6 @@
|
|||||||
import importlib
|
|
||||||
import logging
|
import logging
|
||||||
import sys
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, List, Optional
|
|
||||||
|
|
||||||
# Configure logging
|
# Configure logging
|
||||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
|
||||||
@@ -24,7 +22,6 @@ extensions = [
|
|||||||
"sphinx.ext.napoleon",
|
"sphinx.ext.napoleon",
|
||||||
"sphinx.ext.viewcode",
|
"sphinx.ext.viewcode",
|
||||||
"sphinx.ext.intersphinx",
|
"sphinx.ext.intersphinx",
|
||||||
"sphinx.ext.coverage",
|
|
||||||
]
|
]
|
||||||
|
|
||||||
# Napoleon settings
|
# Napoleon settings
|
||||||
@@ -50,62 +47,26 @@ autodoc_typehints = "description"
|
|||||||
html_show_sphinx = False
|
html_show_sphinx = False
|
||||||
|
|
||||||
|
|
||||||
def get_installed_services() -> Dict[str, Optional[str]]:
|
def clean_title(title: str) -> str:
|
||||||
"""Scan for installed pipecat services and return their status.
|
"""Automatically clean module titles."""
|
||||||
Returns a dictionary of service names and their import status/error message.
|
# Remove everything after space (like 'module', 'processor', etc.)
|
||||||
"""
|
title = title.split(" ")[0]
|
||||||
services_dir = project_root / "src" / "pipecat" / "services"
|
|
||||||
services_status = {}
|
|
||||||
|
|
||||||
if not services_dir.exists():
|
# Get the last part of the dot-separated path
|
||||||
logger.warning(f"Services directory not found: {services_dir}")
|
parts = title.split(".")
|
||||||
return services_status
|
title = parts[-1]
|
||||||
|
|
||||||
for item in services_dir.iterdir():
|
# Handle special cases for common acronyms
|
||||||
if item.is_dir() and not item.name.startswith("_") and not item.name == "to_be_updated":
|
acronyms = ["ai", "aws", "api", "vad"]
|
||||||
service_name = item.name
|
words = title.split("_")
|
||||||
try:
|
cleaned_words = []
|
||||||
module = importlib.import_module(f"pipecat.services.{service_name}")
|
for word in words:
|
||||||
services_status[service_name] = None # None indicates success
|
if word.lower() in acronyms:
|
||||||
logger.info(f"Found service: {service_name} at {module.__file__}")
|
cleaned_words.append(word.upper())
|
||||||
except ImportError as e:
|
else:
|
||||||
services_status[service_name] = str(e)
|
cleaned_words.append(word.capitalize())
|
||||||
logger.warning(f"Failed to import {service_name}: {e}")
|
|
||||||
|
|
||||||
return services_status
|
return " ".join(cleaned_words)
|
||||||
|
|
||||||
|
|
||||||
def generate_services_rst() -> str:
|
|
||||||
"""Generate RST content for services section."""
|
|
||||||
services = get_installed_services()
|
|
||||||
|
|
||||||
# Sort services into successful and failed imports
|
|
||||||
successful = [name for name, status in services.items() if status is None]
|
|
||||||
failed = [(name, status) for name, status in services.items() if status is not None]
|
|
||||||
|
|
||||||
rst_content = [
|
|
||||||
"Services",
|
|
||||||
"~~~~~~~~",
|
|
||||||
"",
|
|
||||||
"Successfully Detected Services:",
|
|
||||||
"",
|
|
||||||
]
|
|
||||||
|
|
||||||
for service in sorted(successful):
|
|
||||||
rst_content.append(f"* :mod:`pipecat.services.{service}`")
|
|
||||||
|
|
||||||
if failed:
|
|
||||||
rst_content.extend(
|
|
||||||
[
|
|
||||||
"",
|
|
||||||
"Services with Import Issues:",
|
|
||||||
"",
|
|
||||||
]
|
|
||||||
)
|
|
||||||
for service, error in sorted(failed):
|
|
||||||
rst_content.append(f"* {service} (Import failed: {error})")
|
|
||||||
|
|
||||||
return "\n".join(rst_content)
|
|
||||||
|
|
||||||
|
|
||||||
def setup(app):
|
def setup(app):
|
||||||
@@ -128,15 +89,6 @@ def setup(app):
|
|||||||
logger.info(f"Output directory: {output_dir}")
|
logger.info(f"Output directory: {output_dir}")
|
||||||
logger.info(f"Source directory: {source_dir}")
|
logger.info(f"Source directory: {source_dir}")
|
||||||
|
|
||||||
# Get installed services
|
|
||||||
services = get_installed_services()
|
|
||||||
logger.info(f"Found {len(services)} services")
|
|
||||||
for service, status in services.items():
|
|
||||||
if status is None:
|
|
||||||
logger.info(f"Service available: {service}")
|
|
||||||
else:
|
|
||||||
logger.warning(f"Service import failed: {service} - {status}")
|
|
||||||
|
|
||||||
excludes = [
|
excludes = [
|
||||||
str(project_root / "src/pipecat/processors/gstreamer"),
|
str(project_root / "src/pipecat/processors/gstreamer"),
|
||||||
str(project_root / "src/pipecat/transports/network"),
|
str(project_root / "src/pipecat/transports/network"),
|
||||||
@@ -165,10 +117,18 @@ def setup(app):
|
|||||||
|
|
||||||
logger.info("API documentation generated successfully!")
|
logger.info("API documentation generated successfully!")
|
||||||
|
|
||||||
# Generate services index file
|
# Process generated RST files to update titles
|
||||||
services_index = Path(output_dir) / "services_index.rst"
|
for rst_file in Path(output_dir).glob("*.rst"):
|
||||||
services_index.write_text(generate_services_rst())
|
content = rst_file.read_text()
|
||||||
logger.info(f"Generated services index at {services_index}")
|
lines = content.split("\n")
|
||||||
|
|
||||||
|
# Find and clean up the title
|
||||||
|
if lines and "=" in lines[1]: # Title is typically the first line
|
||||||
|
old_title = lines[0]
|
||||||
|
new_title = clean_title(old_title)
|
||||||
|
content = content.replace(old_title, new_title)
|
||||||
|
rst_file.write_text(content)
|
||||||
|
logger.info(f"Updated title: {old_title} -> {new_title}")
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error generating API documentation: {e}", exc_info=True)
|
logger.error(f"Error generating API documentation: {e}", exc_info=True)
|
||||||
|
|||||||
Reference in New Issue
Block a user