# cjm-fasthtml-plugins
<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->
## Install
``` bash
pip install cjm_fasthtml_plugins
```
## Project Structure
nbs/
├── core/ (3)
│ ├── execution_mode.ipynb # Enum definitions for plugin execution modes (in-process, subprocess, Docker, cloud, etc.)
│ ├── metadata.ipynb # Plugin metadata structures for tracking plugin information and resources
│ └── registry.ipynb # Unified plugin registry for managing multiple domain-specific plugin systems with configuration persistence
├── protocols/ (2)
│ ├── cloud_aware.ipynb # Protocol for plugins that use cloud or remote computing resources
│ └── lifecycle.ipynb # Protocol for plugins that manage child processes, containers, or other external resources
└── utils/ (1)
└── helpers.ipynb # Utility functions for plugin registry operations
Total: 6 notebooks across 4 directories
## Module Dependencies
``` mermaid
graph LR
core_execution_mode[core.execution_mode<br/>Execution Mode]
core_metadata[core.metadata<br/>Metadata]
core_registry[core.registry<br/>Registry]
protocols_cloud_aware[protocols.cloud_aware<br/>Cloud-Aware Protocol]
protocols_lifecycle[protocols.lifecycle<br/>Lifecycle Protocol]
utils_helpers[utils.helpers<br/>Helpers]
core_metadata --> core_execution_mode
core_registry --> core_metadata
core_registry --> core_execution_mode
protocols_cloud_aware --> core_metadata
protocols_cloud_aware --> core_execution_mode
protocols_lifecycle --> core_execution_mode
utils_helpers --> core_metadata
utils_helpers --> core_execution_mode
```
*8 cross-module dependencies detected*
## CLI Reference
No CLI commands found in this project.
## Module Overview
Detailed documentation for each module in the project:
### Cloud-Aware Protocol (`cloud_aware.ipynb`)
> Protocol for plugins that use cloud or remote computing resources
#### Import
``` python
from cjm_fasthtml_plugins.protocols.cloud_aware import (
CloudAwarePlugin,
is_cloud_aware,
has_active_cloud_resources,
get_total_estimated_cost
)
```
#### Functions
``` python
def is_cloud_aware(plugin: Any) -> bool:
"""Check if a plugin implements the CloudAwarePlugin protocol.
Args:
plugin: Plugin instance to check
Returns:
True if plugin implements the protocol
"""
return isinstance(plugin, CloudAwarePlugin)
def has_active_cloud_resources(plugin: Any) -> bool
"""
Check if a plugin implements the CloudAwarePlugin protocol.
Args:
plugin: Plugin instance to check
Returns:
True if plugin implements the protocol
"""
```
``` python
def has_active_cloud_resources(plugin: Any) -> bool:
"""Check if plugin has active cloud resources.
Args:
plugin: Plugin instance
Returns:
True if plugin has running cloud resources
"""
if not is_cloud_aware(plugin)
"""
Check if plugin has active cloud resources.
Args:
plugin: Plugin instance
Returns:
True if plugin has running cloud resources
"""
```
``` python
def get_total_estimated_cost(plugins: List[Any], duration_hours: float = 1.0) -> float:
"""Get total estimated cost for multiple plugins.
Args:
plugins: List of plugin instances
duration_hours: Duration to estimate for
Returns:
Total estimated cost in USD
"""
total = 0.0
for plugin in plugins
"""
Get total estimated cost for multiple plugins.
Args:
plugins: List of plugin instances
duration_hours: Duration to estimate for
Returns:
Total estimated cost in USD
"""
```
#### Classes
```` python
@runtime_checkable
class CloudAwarePlugin(Protocol):
"""
Protocol for plugins that use cloud/remote resources.
Plugins implementing this protocol provide information about
cloud resources they use, enabling cost tracking, resource
management, and emergency shutdown.
Example:
```python
class CloudFinetunePlugin(FinetuningPlugin, CloudAwarePlugin):
def get_remote_resource_info(self) -> Optional[RemoteResourceInfo]:
if not self.remote_instance:
return None
return RemoteResourceInfo(
provider=CloudProviderType.AWS,
instance_id=self.instance_id,
status="running",
gpu_count=8,
estimated_cost_per_hour=24.50
)
def provision_remote_resource(self, **config) -> RemoteResourceInfo:
# Launch EC2 instance
return self.remote_resource_info
def terminate_remote_resource(self) -> bool:
# Terminate EC2 instance
return True
def estimate_cost(self, duration_hours: float) -> float:
return duration_hours * 24.50
```
"""
def get_remote_resource_info(self) -> Optional[RemoteResourceInfo]:
"""Get information about remote/cloud resources.
Returns:
RemoteResourceInfo if resources are provisioned, None otherwise
"""
...
def provision_remote_resource(self, **config) -> RemoteResourceInfo
"Get information about remote/cloud resources.
Returns:
RemoteResourceInfo if resources are provisioned, None otherwise"
def provision_remote_resource(self, **config) -> RemoteResourceInfo:
"""Provision cloud resources (VM, container, etc.).
Args:
**config: Provider-specific configuration
Returns:
RemoteResourceInfo with details about provisioned resource
"""
...
def check_remote_resource_status(self) -> str
"Provision cloud resources (VM, container, etc.).
Args:
**config: Provider-specific configuration
Returns:
RemoteResourceInfo with details about provisioned resource"
def check_remote_resource_status(self) -> str:
"""Check status of remote resource.
Returns:
Status string (e.g., 'running', 'stopped', 'provisioning')
"""
...
def terminate_remote_resource(self) -> bool
"Check status of remote resource.
Returns:
Status string (e.g., 'running', 'stopped', 'provisioning')"
def terminate_remote_resource(self) -> bool:
"""Terminate/stop cloud resources to avoid costs.
Returns:
True if termination succeeded
"""
...
def estimate_cost(self, duration_hours: float) -> float
"Terminate/stop cloud resources to avoid costs.
Returns:
True if termination succeeded"
def estimate_cost(self, duration_hours: float) -> float
"Estimate cost for running this duration.
Args:
duration_hours: Estimated runtime in hours
Returns:
Estimated cost in USD"
````
### Execution Mode (`execution_mode.ipynb`)
> Enum definitions for plugin execution modes (in-process, subprocess,
> Docker, cloud, etc.)
#### Import
``` python
from cjm_fasthtml_plugins.core.execution_mode import (
PluginExecutionMode,
CloudProviderType
)
```
#### Classes
``` python
class PluginExecutionMode(Enum):
"""
How a plugin executes.
This enum categorizes plugins by their execution environment,
from simple in-process execution to complex cloud deployments.
Examples:
>>> # Simple in-process plugin
>>> mode = PluginExecutionMode.IN_PROCESS
>>>
>>> # Plugin that spawns subprocesses (like vLLM server)
>>> mode = PluginExecutionMode.SUBPROCESS
>>>
>>> # Plugin running on cloud GPU
>>> mode = PluginExecutionMode.CLOUD_GPU
"""
```
``` python
class CloudProviderType(Enum):
"""
Supported cloud providers.
Identifies which cloud provider or GPU rental service is being used
for remote execution.
Examples:
>>> # Major cloud providers
>>> provider = CloudProviderType.AWS
>>> provider = CloudProviderType.GCP
>>>
>>> # GPU rental services
>>> provider = CloudProviderType.LAMBDA_LABS
>>> provider = CloudProviderType.RUNPOD
"""
```
### Helpers (`helpers.ipynb`)
> Utility functions for plugin registry operations
#### Import
``` python
from cjm_fasthtml_plugins.utils.helpers import (
filter_plugins_by_execution_mode,
get_cloud_plugins,
get_local_plugins,
get_configured_plugins,
get_unconfigured_plugins,
get_plugin_stats
)
```
#### Functions
``` python
def filter_plugins_by_execution_mode(
plugins: List[PluginMetadata], # List of plugin metadata
mode: PluginExecutionMode # Execution mode to filter by
) -> List[PluginMetadata]: # Filtered list
"""
Filter plugins by execution mode.
Args:
plugins: List of plugin metadata
mode: Execution mode to filter by
Returns:
List of plugins matching the execution mode
"""
```
``` python
def get_cloud_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]:
"""Get all cloud/remote execution plugins.
Args:
plugins: List of plugin metadata
Returns:
List of cloud/remote plugins
"""
return [p for p in plugins if p.is_cloud_execution()]
def get_local_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]
"""
Get all cloud/remote execution plugins.
Args:
plugins: List of plugin metadata
Returns:
List of cloud/remote plugins
"""
```
``` python
def get_local_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]:
"""Get all local execution plugins.
Args:
plugins: List of plugin metadata
Returns:
List of local plugins
"""
return [p for p in plugins if p.is_local_execution()]
def get_configured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]
"""
Get all local execution plugins.
Args:
plugins: List of plugin metadata
Returns:
List of local plugins
"""
```
``` python
def get_configured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]:
"""Get plugins that have saved configuration.
Args:
plugins: List of plugin metadata
Returns:
List of configured plugins
"""
return [p for p in plugins if p.is_configured]
def get_unconfigured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]
"""
Get plugins that have saved configuration.
Args:
plugins: List of plugin metadata
Returns:
List of configured plugins
"""
```
``` python
def get_unconfigured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]
"""
Get plugins that need configuration.
Args:
plugins: List of plugin metadata
Returns:
List of unconfigured plugins
"""
```
``` python
def get_plugin_stats(plugins: List[PluginMetadata]) -> Dict[str, Any]:
"""Get statistics about a list of plugins.
Args:
plugins: List of plugin metadata
Returns:
Dictionary with plugin statistics
"""
return {
"total": len(plugins),
"""
Get statistics about a list of plugins.
Args:
plugins: List of plugin metadata
Returns:
Dictionary with plugin statistics
"""
```
### Lifecycle Protocol (`lifecycle.ipynb`)
> Protocol for plugins that manage child processes, containers, or other
> external resources
#### Import
``` python
from cjm_fasthtml_plugins.protocols.lifecycle import (
LifecycleAwarePlugin,
is_lifecycle_aware,
get_all_managed_pids
)
```
#### Functions
``` python
def is_lifecycle_aware(plugin: Any) -> bool:
"""Check if a plugin implements the LifecycleAwarePlugin protocol.
Args:
plugin: Plugin instance to check
Returns:
True if plugin implements the protocol
"""
return isinstance(plugin, LifecycleAwarePlugin)
def get_all_managed_pids(plugin: Any) -> List[int]
"""
Check if a plugin implements the LifecycleAwarePlugin protocol.
Args:
plugin: Plugin instance to check
Returns:
True if plugin implements the protocol
"""
```
``` python
def get_all_managed_pids(plugin: Any) -> List[int]:
"""Get all PIDs managed by a plugin (including children).
Args:
plugin: Plugin instance
Returns:
List of all PIDs (empty if plugin not lifecycle-aware)
"""
if not is_lifecycle_aware(plugin)
"""
Get all PIDs managed by a plugin (including children).
Args:
plugin: Plugin instance
Returns:
List of all PIDs (empty if plugin not lifecycle-aware)
"""
```
#### Classes
```` python
@runtime_checkable
class LifecycleAwarePlugin(Protocol):
"""
Protocol for plugins that manage external resources.
Plugins implementing this protocol provide information about
child processes, containers, or other resources they manage.
This enables:
- Resource tracking across the application
- Proper cleanup when stopping plugins
- Conflict detection for GPU/memory usage
- Cost tracking for cloud resources
Example:
```python
class VoxtralVLLMPlugin(TranscriptionPlugin, LifecycleAwarePlugin):
def get_execution_mode(self) -> PluginExecutionMode:
return PluginExecutionMode.SUBPROCESS
def get_child_pids(self) -> List[int]:
if not self.server or not self.server.process:
return []
return [self.server.process.pid]
def get_managed_resources(self) -> Dict[str, Any]:
return {
'server_url': self.server.base_url,
'is_running': self.server.is_running()
}
def force_cleanup(self) -> None:
if self.server:
self.server.stop()
```
"""
def get_execution_mode(self) -> PluginExecutionMode:
"""Get the execution mode of this plugin.
Returns:
PluginExecutionMode indicating how this plugin executes
"""
...
def get_child_pids(self) -> List[int]
"Get the execution mode of this plugin.
Returns:
PluginExecutionMode indicating how this plugin executes"
def get_child_pids(self) -> List[int]:
"""Get PIDs of any child processes managed by this plugin.
For plugins that spawn subprocesses (e.g., vLLM servers), this
should return all child process PIDs for resource tracking.
Returns:
List of process IDs (empty list if no child processes)
"""
...
def get_managed_resources(self) -> Dict[str, Any]
"Get PIDs of any child processes managed by this plugin.
For plugins that spawn subprocesses (e.g., vLLM servers), this
should return all child process PIDs for resource tracking.
Returns:
List of process IDs (empty list if no child processes)"
def get_managed_resources(self) -> Dict[str, Any]:
"""Get information about managed resources.
This can include:
- Server URLs and ports
- Container IDs
- Conda environment names
- Status information
- Any other plugin-specific resource info
Returns:
Dictionary with resource information
"""
...
def force_cleanup(self) -> None
"Get information about managed resources.
This can include:
- Server URLs and ports
- Container IDs
- Conda environment names
- Status information
- Any other plugin-specific resource info
Returns:
Dictionary with resource information"
def force_cleanup(self) -> None
"Force cleanup of all managed resources.
This should be more aggressive than regular cleanup(),
killing processes, stopping containers, etc. Used for
emergency shutdown scenarios."
````
### Metadata (`metadata.ipynb`)
> Plugin metadata structures for tracking plugin information and
> resources
#### Import
``` python
from cjm_fasthtml_plugins.core.metadata import (
RemoteResourceInfo,
PluginMetadata
)
```
#### Classes
``` python
@dataclass
class RemoteResourceInfo:
"""
Information about a remote/cloud resource.
Tracks details about cloud VMs, containers, or other remote resources
that a plugin is using for execution.
Attributes:
provider: Cloud provider or service
region: Cloud region/zone
instance_id: VM/instance identifier
job_id: Job/task identifier on remote system
endpoint_url: HTTP endpoint for API access
ssh_host: SSH host for remote access
ssh_port: SSH port number
status: Current status (provisioning, running, stopping, stopped)
resource_type: Instance type (e.g., 'p3.2xlarge', 'n1-standard-8')
gpu_count: Number of GPUs
gpu_type: GPU model (e.g., 'V100', 'A100', 'H100')
estimated_cost_per_hour: Estimated hourly cost in USD
metadata: Additional provider-specific metadata
"""
provider: CloudProviderType
region: Optional[str]
instance_id: Optional[str]
job_id: Optional[str]
endpoint_url: Optional[str]
ssh_host: Optional[str]
ssh_port: int = 22
status: str = 'unknown' # provisioning, running, stopping, stopped
resource_type: Optional[str]
gpu_count: int = 0
gpu_type: Optional[str]
estimated_cost_per_hour: Optional[float]
metadata: Dict[str, Any] = field(...)
```
```` python
@dataclass
class PluginMetadata:
"""
Metadata describing a plugin.
This dataclass holds information about a plugin that can be displayed
in settings UI and used for resource management without loading the
actual plugin instance.
Categories are simple strings - applications choose their own category names
based on their needs (e.g., 'transcription', 'llm', 'image_generation', etc.).
Attributes:
name: Internal plugin identifier
category: Plugin category string (application-defined)
title: Display title for the plugin
config_schema: JSON Schema for plugin configuration
description: Optional plugin description
version: Optional plugin version
is_configured: Whether the plugin has saved configuration
# Lifecycle metadata
execution_mode: How the plugin executes (in-process, subprocess, cloud, etc.)
manages_child_processes: Whether plugin spawns child processes
manages_external_resources: Whether plugin manages Docker/servers/etc.
# Local resource tracking
spawned_pids: List of child process PIDs
container_id: Docker container ID if applicable
conda_env_name: Conda environment name if applicable
# Cloud/Remote resource tracking
remote_resource: Remote resource information if applicable
Example:
```python
# Simple in-process plugin
metadata = PluginMetadata(
name="whisper_base",
category="transcription",
title="Whisper Base Model",
config_schema={...},
execution_mode=PluginExecutionMode.IN_PROCESS
)
# Plugin with vLLM server (subprocess)
metadata = PluginMetadata(
name="voxtral_vllm",
category="transcription",
title="Voxtral via vLLM",
config_schema={...},
execution_mode=PluginExecutionMode.SUBPROCESS,
manages_child_processes=True,
spawned_pids=[12345, 12346, 12347]
)
# Cloud-based plugin
metadata = PluginMetadata(
name="llm_finetune_cloud",
category="finetuning",
title="Cloud LLM Finetuning",
config_schema={...},
execution_mode=PluginExecutionMode.CLOUD_GPU,
manages_external_resources=True,
remote_resource=RemoteResourceInfo(...)
)
```
"""
name: str
category: str
title: str
config_schema: Dict[str, Any]
description: Optional[str]
version: Optional[str]
is_configured: bool = False
execution_mode: PluginExecutionMode = PluginExecutionMode.IN_PROCESS
manages_child_processes: bool = False
manages_external_resources: bool = False
spawned_pids: List[int] = field(...)
container_id: Optional[str]
conda_env_name: Optional[str]
remote_resource: Optional[RemoteResourceInfo]
def get_unique_id(self) -> str:
"""Generate unique ID for this plugin.
Returns:
String in format 'category_name'
"""
return f"{self.category}_{self.name}"
def is_local_execution(self) -> bool
"Generate unique ID for this plugin.
Returns:
String in format 'category_name'"
def is_local_execution(self) -> bool:
"""Check if plugin executes locally (not cloud/remote).
Returns:
True if execution is local
"""
local_modes = {
PluginExecutionMode.IN_PROCESS,
PluginExecutionMode.SUBPROCESS,
PluginExecutionMode.DOCKER,
PluginExecutionMode.CONDA_ENV,
PluginExecutionMode.EXTERNAL_SERVICE
}
return self.execution_mode in local_modes
def is_cloud_execution(self) -> bool
"Check if plugin executes locally (not cloud/remote).
Returns:
True if execution is local"
def is_cloud_execution(self) -> bool:
"""Check if plugin executes on cloud/remote resources.
Returns:
True if execution is cloud/remote
"""
return not self.is_local_execution()
def has_active_resources(self) -> bool
"Check if plugin executes on cloud/remote resources.
Returns:
True if execution is cloud/remote"
def has_active_resources(self) -> bool
"Check if plugin has active managed resources.
Returns:
True if plugin has child processes, containers, or cloud resources"
````
### Registry (`registry.ipynb`)
> Unified plugin registry for managing multiple domain-specific plugin
> systems with configuration persistence
#### Import
``` python
from cjm_fasthtml_plugins.core.registry import (
T,
UnifiedPluginRegistry
)
```
#### Classes
```` python
class UnifiedPluginRegistry:
def __init__(self, config_dir: Optional[Path] = None):
"""Initialize the unified plugin registry.
Args:
config_dir: Directory for plugin configuration files (default: 'configs')
"""
self._managers: Dict[str, Any] = {} # category -> manager
"""
Unified registry for multiple domain-specific plugin systems.
Manages plugin managers from different domains (transcription, LLM, etc.)
and provides a single interface for plugin discovery, configuration,
and resource management.
Example:
```python
from cjm_plugin_system.core.manager import PluginManager
from cjm_transcription_plugin_system.plugin_interface import TranscriptionPlugin
# Create registry
registry = UnifiedPluginRegistry()
# Register transcription plugins
transcription_mgr = PluginManager(plugin_interface=TranscriptionPlugin)
registry.register_plugin_manager(
category="transcription",
manager=transcription_mgr,
display_name="Transcription"
)
# Get all plugins
all_plugins = registry.get_all_plugins()
# Get plugins by category
transcription_plugins = registry.get_plugins_by_category("transcription")
# Get manager for specific operations
mgr = registry.get_manager("transcription")
```
"""
def __init__(self, config_dir: Optional[Path] = None):
"""Initialize the unified plugin registry.
Args:
config_dir: Directory for plugin configuration files (default: 'configs')
"""
self._managers: Dict[str, Any] = {} # category -> manager
"Initialize the unified plugin registry.
Args:
config_dir: Directory for plugin configuration files (default: 'configs')"
def register_plugin_manager(
self,
category: str, # Category name (e.g., "transcription", "llm")
manager: Any, # Domain-specific plugin manager
display_name: Optional[str] = None, # Display name for UI
auto_discover: bool = True # Automatically discover plugins?
) -> List[PluginMetadata]: # List of discovered plugin metadata
"Register a domain-specific plugin manager.
Args:
category: String category (e.g., 'transcription')
manager: The domain-specific plugin manager instance
display_name: Optional display name for UI
auto_discover: Automatically discover and register plugins
Returns:
List of discovered plugin metadata"
def get_manager(
self,
category: str, # Category name
manager_type: Optional[Type[T]] = None # Optional type hint
) -> Optional[T]: # Plugin manager instance
"Get plugin manager for a specific category.
Args:
category: Category name (e.g., 'transcription')
manager_type: Optional type hint for IDE autocomplete
Returns:
Plugin manager instance if found, None otherwise"
def get_categories(self) -> List[str]:
"""Get all registered categories.
Returns:
Sorted list of category names
"""
return sorted(self._categories.keys())
def get_category_display_name(self, category: str) -> str
"Get all registered categories.
Returns:
Sorted list of category names"
def get_category_display_name(self, category: str) -> str:
"""Get display name for a category.
Args:
category: Category name
Returns:
Display name or category name if not set
"""
return self._categories.get(category, category.title())
def get_plugin(self, unique_id: str) -> Optional[PluginMetadata]
"Get display name for a category.
Args:
category: Category name
Returns:
Display name or category name if not set"
def get_plugin(self, unique_id: str) -> Optional[PluginMetadata]:
"""Get plugin metadata by unique ID.
Args:
unique_id: Plugin unique identifier (format: 'category_name')
Returns:
Plugin metadata if found, None otherwise
"""
return self._plugins.get(unique_id)
def get_plugins_by_category(self, category: str) -> List[PluginMetadata]
"Get plugin metadata by unique ID.
Args:
unique_id: Plugin unique identifier (format: 'category_name')
Returns:
Plugin metadata if found, None otherwise"
def get_plugins_by_category(self, category: str) -> List[PluginMetadata]:
"""Get all plugins in a category.
Args:
category: Category name
Returns:
List of plugin metadata for the category
"""
return [p for p in self._plugins.values() if p.category == category]
def get_all_plugins(self) -> List[PluginMetadata]
"Get all plugins in a category.
Args:
category: Category name
Returns:
List of plugin metadata for the category"
def get_all_plugins(self) -> List[PluginMetadata]:
"""Get all plugins across all categories.
Returns:
List of all plugin metadata
"""
return list(self._plugins.values())
def get_categories_with_plugins(self) -> List[str]
"Get all plugins across all categories.
Returns:
List of all plugin metadata"
def get_categories_with_plugins(self) -> List[str]:
"""Get categories that have registered plugins.
Returns:
Sorted list of categories with plugins
"""
categories = set(p.category for p in self._plugins.values())
return sorted(categories)
def load_plugin_config(self, unique_id: str) -> Dict[str, Any]
"Get categories that have registered plugins.
Returns:
Sorted list of categories with plugins"
def load_plugin_config(self, unique_id: str) -> Dict[str, Any]:
"""Load saved configuration for a plugin.
Args:
unique_id: Plugin unique identifier
Returns:
Configuration dictionary (empty if no config exists)
"""
config_file = self._config_dir / f"{unique_id}.json"
if config_file.exists()
"Load saved configuration for a plugin.
Args:
unique_id: Plugin unique identifier
Returns:
Configuration dictionary (empty if no config exists)"
def save_plugin_config(self, unique_id: str, config: Dict[str, Any]) -> bool:
"""Save configuration for a plugin.
Args:
unique_id: Plugin unique identifier
config: Configuration dictionary to save
Returns:
True if save succeeded, False otherwise
"""
try
"Save configuration for a plugin.
Args:
unique_id: Plugin unique identifier
config: Configuration dictionary to save
Returns:
True if save succeeded, False otherwise"
def delete_plugin_config(self, unique_id: str) -> bool:
"""Delete saved configuration for a plugin.
Args:
unique_id: Plugin unique identifier
Returns:
True if deletion succeeded, False otherwise
"""
try
"Delete saved configuration for a plugin.
Args:
unique_id: Plugin unique identifier
Returns:
True if deletion succeeded, False otherwise"
````
#### Variables
``` python
T
```
Raw data
{
"_id": null,
"home_page": "https://github.com/cj-mills/cjm-fasthtml-plugins",
"name": "cjm-fasthtml-plugins",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "nbdev jupyter notebook python",
"author": "Christian J. Mills",
"author_email": "9126128+cj-mills@users.noreply.github.com",
"download_url": "https://files.pythonhosted.org/packages/f5/d3/570deaf0ab5e769fa16055d46429ec14e5366e81150b3cb9dd87df30bf3c/cjm_fasthtml_plugins-0.0.1.tar.gz",
"platform": null,
"description": "# cjm-fasthtml-plugins\n\n\n<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->\n\n## Install\n\n``` bash\npip install cjm_fasthtml_plugins\n```\n\n## Project Structure\n\n nbs/\n \u251c\u2500\u2500 core/ (3)\n \u2502 \u251c\u2500\u2500 execution_mode.ipynb # Enum definitions for plugin execution modes (in-process, subprocess, Docker, cloud, etc.)\n \u2502 \u251c\u2500\u2500 metadata.ipynb # Plugin metadata structures for tracking plugin information and resources\n \u2502 \u2514\u2500\u2500 registry.ipynb # Unified plugin registry for managing multiple domain-specific plugin systems with configuration persistence\n \u251c\u2500\u2500 protocols/ (2)\n \u2502 \u251c\u2500\u2500 cloud_aware.ipynb # Protocol for plugins that use cloud or remote computing resources\n \u2502 \u2514\u2500\u2500 lifecycle.ipynb # Protocol for plugins that manage child processes, containers, or other external resources\n \u2514\u2500\u2500 utils/ (1)\n \u2514\u2500\u2500 helpers.ipynb # Utility functions for plugin registry operations\n\nTotal: 6 notebooks across 4 directories\n\n## Module Dependencies\n\n``` mermaid\ngraph LR\n core_execution_mode[core.execution_mode<br/>Execution Mode]\n core_metadata[core.metadata<br/>Metadata]\n core_registry[core.registry<br/>Registry]\n protocols_cloud_aware[protocols.cloud_aware<br/>Cloud-Aware Protocol]\n protocols_lifecycle[protocols.lifecycle<br/>Lifecycle Protocol]\n utils_helpers[utils.helpers<br/>Helpers]\n\n core_metadata --> core_execution_mode\n core_registry --> core_metadata\n core_registry --> core_execution_mode\n protocols_cloud_aware --> core_metadata\n protocols_cloud_aware --> core_execution_mode\n protocols_lifecycle --> core_execution_mode\n utils_helpers --> core_metadata\n utils_helpers --> core_execution_mode\n```\n\n*8 cross-module dependencies detected*\n\n## CLI Reference\n\nNo CLI commands found in this project.\n\n## Module Overview\n\nDetailed documentation for each module in the project:\n\n### Cloud-Aware Protocol (`cloud_aware.ipynb`)\n\n> Protocol for plugins that use cloud or remote computing resources\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_plugins.protocols.cloud_aware import (\n CloudAwarePlugin,\n is_cloud_aware,\n has_active_cloud_resources,\n get_total_estimated_cost\n)\n```\n\n#### Functions\n\n``` python\ndef is_cloud_aware(plugin: Any) -> bool:\n \"\"\"Check if a plugin implements the CloudAwarePlugin protocol.\n \n Args:\n plugin: Plugin instance to check\n \n Returns:\n True if plugin implements the protocol\n \"\"\"\n return isinstance(plugin, CloudAwarePlugin)\n\ndef has_active_cloud_resources(plugin: Any) -> bool\n \"\"\"\n Check if a plugin implements the CloudAwarePlugin protocol.\n \n Args:\n plugin: Plugin instance to check\n \n Returns:\n True if plugin implements the protocol\n \"\"\"\n```\n\n``` python\ndef has_active_cloud_resources(plugin: Any) -> bool:\n \"\"\"Check if plugin has active cloud resources.\n \n Args:\n plugin: Plugin instance\n \n Returns:\n True if plugin has running cloud resources\n \"\"\"\n if not is_cloud_aware(plugin)\n \"\"\"\n Check if plugin has active cloud resources.\n \n Args:\n plugin: Plugin instance\n \n Returns:\n True if plugin has running cloud resources\n \"\"\"\n```\n\n``` python\ndef get_total_estimated_cost(plugins: List[Any], duration_hours: float = 1.0) -> float:\n \"\"\"Get total estimated cost for multiple plugins.\n \n Args:\n plugins: List of plugin instances\n duration_hours: Duration to estimate for\n \n Returns:\n Total estimated cost in USD\n \"\"\"\n total = 0.0\n for plugin in plugins\n \"\"\"\n Get total estimated cost for multiple plugins.\n \n Args:\n plugins: List of plugin instances\n duration_hours: Duration to estimate for\n \n Returns:\n Total estimated cost in USD\n \"\"\"\n```\n\n#### Classes\n\n```` python\n@runtime_checkable\nclass CloudAwarePlugin(Protocol):\n \"\"\"\n Protocol for plugins that use cloud/remote resources.\n \n Plugins implementing this protocol provide information about\n cloud resources they use, enabling cost tracking, resource\n management, and emergency shutdown.\n \n Example:\n ```python\n class CloudFinetunePlugin(FinetuningPlugin, CloudAwarePlugin):\n def get_remote_resource_info(self) -> Optional[RemoteResourceInfo]:\n if not self.remote_instance:\n return None\n return RemoteResourceInfo(\n provider=CloudProviderType.AWS,\n instance_id=self.instance_id,\n status=\"running\",\n gpu_count=8,\n estimated_cost_per_hour=24.50\n )\n \n def provision_remote_resource(self, **config) -> RemoteResourceInfo:\n # Launch EC2 instance\n return self.remote_resource_info\n \n def terminate_remote_resource(self) -> bool:\n # Terminate EC2 instance\n return True\n \n def estimate_cost(self, duration_hours: float) -> float:\n return duration_hours * 24.50\n ```\n \"\"\"\n \n def get_remote_resource_info(self) -> Optional[RemoteResourceInfo]:\n \"\"\"Get information about remote/cloud resources.\n \n Returns:\n RemoteResourceInfo if resources are provisioned, None otherwise\n \"\"\"\n ...\n \n def provision_remote_resource(self, **config) -> RemoteResourceInfo\n \"Get information about remote/cloud resources.\n\nReturns:\n RemoteResourceInfo if resources are provisioned, None otherwise\"\n \n def provision_remote_resource(self, **config) -> RemoteResourceInfo:\n \"\"\"Provision cloud resources (VM, container, etc.).\n \n Args:\n **config: Provider-specific configuration\n \n Returns:\n RemoteResourceInfo with details about provisioned resource\n \"\"\"\n ...\n \n def check_remote_resource_status(self) -> str\n \"Provision cloud resources (VM, container, etc.).\n\nArgs:\n **config: Provider-specific configuration\n\nReturns:\n RemoteResourceInfo with details about provisioned resource\"\n \n def check_remote_resource_status(self) -> str:\n \"\"\"Check status of remote resource.\n \n Returns:\n Status string (e.g., 'running', 'stopped', 'provisioning')\n \"\"\"\n ...\n \n def terminate_remote_resource(self) -> bool\n \"Check status of remote resource.\n\nReturns:\n Status string (e.g., 'running', 'stopped', 'provisioning')\"\n \n def terminate_remote_resource(self) -> bool:\n \"\"\"Terminate/stop cloud resources to avoid costs.\n \n Returns:\n True if termination succeeded\n \"\"\"\n ...\n \n def estimate_cost(self, duration_hours: float) -> float\n \"Terminate/stop cloud resources to avoid costs.\n\nReturns:\n True if termination succeeded\"\n \n def estimate_cost(self, duration_hours: float) -> float\n \"Estimate cost for running this duration.\n\nArgs:\n duration_hours: Estimated runtime in hours\n\nReturns:\n Estimated cost in USD\"\n````\n\n### Execution Mode (`execution_mode.ipynb`)\n\n> Enum definitions for plugin execution modes (in-process, subprocess,\n> Docker, cloud, etc.)\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_plugins.core.execution_mode import (\n PluginExecutionMode,\n CloudProviderType\n)\n```\n\n#### Classes\n\n``` python\nclass PluginExecutionMode(Enum):\n \"\"\"\n How a plugin executes.\n \n This enum categorizes plugins by their execution environment,\n from simple in-process execution to complex cloud deployments.\n \n Examples:\n >>> # Simple in-process plugin\n >>> mode = PluginExecutionMode.IN_PROCESS\n >>> \n >>> # Plugin that spawns subprocesses (like vLLM server)\n >>> mode = PluginExecutionMode.SUBPROCESS\n >>> \n >>> # Plugin running on cloud GPU\n >>> mode = PluginExecutionMode.CLOUD_GPU\n \"\"\"\n```\n\n``` python\nclass CloudProviderType(Enum):\n \"\"\"\n Supported cloud providers.\n \n Identifies which cloud provider or GPU rental service is being used\n for remote execution.\n \n Examples:\n >>> # Major cloud providers\n >>> provider = CloudProviderType.AWS\n >>> provider = CloudProviderType.GCP\n >>> \n >>> # GPU rental services\n >>> provider = CloudProviderType.LAMBDA_LABS\n >>> provider = CloudProviderType.RUNPOD\n \"\"\"\n```\n\n### Helpers (`helpers.ipynb`)\n\n> Utility functions for plugin registry operations\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_plugins.utils.helpers import (\n filter_plugins_by_execution_mode,\n get_cloud_plugins,\n get_local_plugins,\n get_configured_plugins,\n get_unconfigured_plugins,\n get_plugin_stats\n)\n```\n\n#### Functions\n\n``` python\ndef filter_plugins_by_execution_mode(\n plugins: List[PluginMetadata], # List of plugin metadata\n mode: PluginExecutionMode # Execution mode to filter by\n) -> List[PluginMetadata]: # Filtered list\n \"\"\"\n Filter plugins by execution mode.\n \n Args:\n plugins: List of plugin metadata\n mode: Execution mode to filter by\n \n Returns:\n List of plugins matching the execution mode\n \"\"\"\n```\n\n``` python\ndef get_cloud_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]:\n \"\"\"Get all cloud/remote execution plugins.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n List of cloud/remote plugins\n \"\"\"\n return [p for p in plugins if p.is_cloud_execution()]\n\ndef get_local_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]\n \"\"\"\n Get all cloud/remote execution plugins.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n List of cloud/remote plugins\n \"\"\"\n```\n\n``` python\ndef get_local_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]:\n \"\"\"Get all local execution plugins.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n List of local plugins\n \"\"\"\n return [p for p in plugins if p.is_local_execution()]\n\ndef get_configured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]\n \"\"\"\n Get all local execution plugins.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n List of local plugins\n \"\"\"\n```\n\n``` python\ndef get_configured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]:\n \"\"\"Get plugins that have saved configuration.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n List of configured plugins\n \"\"\"\n return [p for p in plugins if p.is_configured]\n\ndef get_unconfigured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]\n \"\"\"\n Get plugins that have saved configuration.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n List of configured plugins\n \"\"\"\n```\n\n``` python\ndef get_unconfigured_plugins(plugins: List[PluginMetadata]) -> List[PluginMetadata]\n \"\"\"\n Get plugins that need configuration.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n List of unconfigured plugins\n \"\"\"\n```\n\n``` python\ndef get_plugin_stats(plugins: List[PluginMetadata]) -> Dict[str, Any]:\n \"\"\"Get statistics about a list of plugins.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n Dictionary with plugin statistics\n \"\"\"\n return {\n \"total\": len(plugins),\n \"\"\"\n Get statistics about a list of plugins.\n \n Args:\n plugins: List of plugin metadata\n \n Returns:\n Dictionary with plugin statistics\n \"\"\"\n```\n\n### Lifecycle Protocol (`lifecycle.ipynb`)\n\n> Protocol for plugins that manage child processes, containers, or other\n> external resources\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_plugins.protocols.lifecycle import (\n LifecycleAwarePlugin,\n is_lifecycle_aware,\n get_all_managed_pids\n)\n```\n\n#### Functions\n\n``` python\ndef is_lifecycle_aware(plugin: Any) -> bool:\n \"\"\"Check if a plugin implements the LifecycleAwarePlugin protocol.\n \n Args:\n plugin: Plugin instance to check\n \n Returns:\n True if plugin implements the protocol\n \"\"\"\n return isinstance(plugin, LifecycleAwarePlugin)\n\ndef get_all_managed_pids(plugin: Any) -> List[int]\n \"\"\"\n Check if a plugin implements the LifecycleAwarePlugin protocol.\n \n Args:\n plugin: Plugin instance to check\n \n Returns:\n True if plugin implements the protocol\n \"\"\"\n```\n\n``` python\ndef get_all_managed_pids(plugin: Any) -> List[int]:\n \"\"\"Get all PIDs managed by a plugin (including children).\n \n Args:\n plugin: Plugin instance\n \n Returns:\n List of all PIDs (empty if plugin not lifecycle-aware)\n \"\"\"\n if not is_lifecycle_aware(plugin)\n \"\"\"\n Get all PIDs managed by a plugin (including children).\n \n Args:\n plugin: Plugin instance\n \n Returns:\n List of all PIDs (empty if plugin not lifecycle-aware)\n \"\"\"\n```\n\n#### Classes\n\n```` python\n@runtime_checkable\nclass LifecycleAwarePlugin(Protocol):\n \"\"\"\n Protocol for plugins that manage external resources.\n \n Plugins implementing this protocol provide information about\n child processes, containers, or other resources they manage.\n \n This enables:\n - Resource tracking across the application\n - Proper cleanup when stopping plugins\n - Conflict detection for GPU/memory usage\n - Cost tracking for cloud resources\n \n Example:\n ```python\n class VoxtralVLLMPlugin(TranscriptionPlugin, LifecycleAwarePlugin):\n def get_execution_mode(self) -> PluginExecutionMode:\n return PluginExecutionMode.SUBPROCESS\n \n def get_child_pids(self) -> List[int]:\n if not self.server or not self.server.process:\n return []\n return [self.server.process.pid]\n \n def get_managed_resources(self) -> Dict[str, Any]:\n return {\n 'server_url': self.server.base_url,\n 'is_running': self.server.is_running()\n }\n \n def force_cleanup(self) -> None:\n if self.server:\n self.server.stop()\n ```\n \"\"\"\n \n def get_execution_mode(self) -> PluginExecutionMode:\n \"\"\"Get the execution mode of this plugin.\n \n Returns:\n PluginExecutionMode indicating how this plugin executes\n \"\"\"\n ...\n \n def get_child_pids(self) -> List[int]\n \"Get the execution mode of this plugin.\n\nReturns:\n PluginExecutionMode indicating how this plugin executes\"\n \n def get_child_pids(self) -> List[int]:\n \"\"\"Get PIDs of any child processes managed by this plugin.\n \n For plugins that spawn subprocesses (e.g., vLLM servers), this\n should return all child process PIDs for resource tracking.\n \n Returns:\n List of process IDs (empty list if no child processes)\n \"\"\"\n ...\n \n def get_managed_resources(self) -> Dict[str, Any]\n \"Get PIDs of any child processes managed by this plugin.\n\nFor plugins that spawn subprocesses (e.g., vLLM servers), this\nshould return all child process PIDs for resource tracking.\n\nReturns:\n List of process IDs (empty list if no child processes)\"\n \n def get_managed_resources(self) -> Dict[str, Any]:\n \"\"\"Get information about managed resources.\n \n This can include:\n - Server URLs and ports\n - Container IDs\n - Conda environment names\n - Status information\n - Any other plugin-specific resource info\n \n Returns:\n Dictionary with resource information\n \"\"\"\n ...\n \n def force_cleanup(self) -> None\n \"Get information about managed resources.\n\nThis can include:\n- Server URLs and ports\n- Container IDs\n- Conda environment names\n- Status information\n- Any other plugin-specific resource info\n\nReturns:\n Dictionary with resource information\"\n \n def force_cleanup(self) -> None\n \"Force cleanup of all managed resources.\n\nThis should be more aggressive than regular cleanup(),\nkilling processes, stopping containers, etc. Used for\nemergency shutdown scenarios.\"\n````\n\n### Metadata (`metadata.ipynb`)\n\n> Plugin metadata structures for tracking plugin information and\n> resources\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_plugins.core.metadata import (\n RemoteResourceInfo,\n PluginMetadata\n)\n```\n\n#### Classes\n\n``` python\n@dataclass\nclass RemoteResourceInfo:\n \"\"\"\n Information about a remote/cloud resource.\n \n Tracks details about cloud VMs, containers, or other remote resources\n that a plugin is using for execution.\n \n Attributes:\n provider: Cloud provider or service\n region: Cloud region/zone\n instance_id: VM/instance identifier\n job_id: Job/task identifier on remote system\n endpoint_url: HTTP endpoint for API access\n ssh_host: SSH host for remote access\n ssh_port: SSH port number\n status: Current status (provisioning, running, stopping, stopped)\n resource_type: Instance type (e.g., 'p3.2xlarge', 'n1-standard-8')\n gpu_count: Number of GPUs\n gpu_type: GPU model (e.g., 'V100', 'A100', 'H100')\n estimated_cost_per_hour: Estimated hourly cost in USD\n metadata: Additional provider-specific metadata\n \"\"\"\n \n provider: CloudProviderType\n region: Optional[str]\n instance_id: Optional[str]\n job_id: Optional[str]\n endpoint_url: Optional[str]\n ssh_host: Optional[str]\n ssh_port: int = 22\n status: str = 'unknown' # provisioning, running, stopping, stopped\n resource_type: Optional[str]\n gpu_count: int = 0\n gpu_type: Optional[str]\n estimated_cost_per_hour: Optional[float]\n metadata: Dict[str, Any] = field(...)\n```\n\n```` python\n@dataclass\nclass PluginMetadata:\n \"\"\"\n Metadata describing a plugin.\n \n This dataclass holds information about a plugin that can be displayed\n in settings UI and used for resource management without loading the\n actual plugin instance.\n \n Categories are simple strings - applications choose their own category names\n based on their needs (e.g., 'transcription', 'llm', 'image_generation', etc.).\n \n Attributes:\n name: Internal plugin identifier\n category: Plugin category string (application-defined)\n title: Display title for the plugin\n config_schema: JSON Schema for plugin configuration\n description: Optional plugin description\n version: Optional plugin version\n is_configured: Whether the plugin has saved configuration\n \n # Lifecycle metadata\n execution_mode: How the plugin executes (in-process, subprocess, cloud, etc.)\n manages_child_processes: Whether plugin spawns child processes\n manages_external_resources: Whether plugin manages Docker/servers/etc.\n \n # Local resource tracking\n spawned_pids: List of child process PIDs\n container_id: Docker container ID if applicable\n conda_env_name: Conda environment name if applicable\n \n # Cloud/Remote resource tracking\n remote_resource: Remote resource information if applicable\n \n Example:\n ```python\n # Simple in-process plugin\n metadata = PluginMetadata(\n name=\"whisper_base\",\n category=\"transcription\",\n title=\"Whisper Base Model\",\n config_schema={...},\n execution_mode=PluginExecutionMode.IN_PROCESS\n )\n \n # Plugin with vLLM server (subprocess)\n metadata = PluginMetadata(\n name=\"voxtral_vllm\",\n category=\"transcription\",\n title=\"Voxtral via vLLM\",\n config_schema={...},\n execution_mode=PluginExecutionMode.SUBPROCESS,\n manages_child_processes=True,\n spawned_pids=[12345, 12346, 12347]\n )\n \n # Cloud-based plugin\n metadata = PluginMetadata(\n name=\"llm_finetune_cloud\",\n category=\"finetuning\",\n title=\"Cloud LLM Finetuning\",\n config_schema={...},\n execution_mode=PluginExecutionMode.CLOUD_GPU,\n manages_external_resources=True,\n remote_resource=RemoteResourceInfo(...)\n )\n ```\n \"\"\"\n \n name: str\n category: str\n title: str\n config_schema: Dict[str, Any]\n description: Optional[str]\n version: Optional[str]\n is_configured: bool = False\n execution_mode: PluginExecutionMode = PluginExecutionMode.IN_PROCESS\n manages_child_processes: bool = False\n manages_external_resources: bool = False\n spawned_pids: List[int] = field(...)\n container_id: Optional[str]\n conda_env_name: Optional[str]\n remote_resource: Optional[RemoteResourceInfo]\n \n def get_unique_id(self) -> str:\n \"\"\"Generate unique ID for this plugin.\n \n Returns:\n String in format 'category_name'\n \"\"\"\n return f\"{self.category}_{self.name}\"\n \n def is_local_execution(self) -> bool\n \"Generate unique ID for this plugin.\n\nReturns:\n String in format 'category_name'\"\n \n def is_local_execution(self) -> bool:\n \"\"\"Check if plugin executes locally (not cloud/remote).\n \n Returns:\n True if execution is local\n \"\"\"\n local_modes = {\n PluginExecutionMode.IN_PROCESS,\n PluginExecutionMode.SUBPROCESS,\n PluginExecutionMode.DOCKER,\n PluginExecutionMode.CONDA_ENV,\n PluginExecutionMode.EXTERNAL_SERVICE\n }\n return self.execution_mode in local_modes\n \n def is_cloud_execution(self) -> bool\n \"Check if plugin executes locally (not cloud/remote).\n\nReturns:\n True if execution is local\"\n \n def is_cloud_execution(self) -> bool:\n \"\"\"Check if plugin executes on cloud/remote resources.\n \n Returns:\n True if execution is cloud/remote\n \"\"\"\n return not self.is_local_execution()\n \n def has_active_resources(self) -> bool\n \"Check if plugin executes on cloud/remote resources.\n\nReturns:\n True if execution is cloud/remote\"\n \n def has_active_resources(self) -> bool\n \"Check if plugin has active managed resources.\n\nReturns:\n True if plugin has child processes, containers, or cloud resources\"\n````\n\n### Registry (`registry.ipynb`)\n\n> Unified plugin registry for managing multiple domain-specific plugin\n> systems with configuration persistence\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_plugins.core.registry import (\n T,\n UnifiedPluginRegistry\n)\n```\n\n#### Classes\n\n```` python\nclass UnifiedPluginRegistry:\n def __init__(self, config_dir: Optional[Path] = None):\n \"\"\"Initialize the unified plugin registry.\n \n Args:\n config_dir: Directory for plugin configuration files (default: 'configs')\n \"\"\"\n self._managers: Dict[str, Any] = {} # category -> manager\n \"\"\"\n Unified registry for multiple domain-specific plugin systems.\n \n Manages plugin managers from different domains (transcription, LLM, etc.)\n and provides a single interface for plugin discovery, configuration,\n and resource management.\n \n Example:\n ```python\n from cjm_plugin_system.core.manager import PluginManager\n from cjm_transcription_plugin_system.plugin_interface import TranscriptionPlugin\n \n # Create registry\n registry = UnifiedPluginRegistry()\n \n # Register transcription plugins\n transcription_mgr = PluginManager(plugin_interface=TranscriptionPlugin)\n registry.register_plugin_manager(\n category=\"transcription\",\n manager=transcription_mgr,\n display_name=\"Transcription\"\n )\n \n # Get all plugins\n all_plugins = registry.get_all_plugins()\n \n # Get plugins by category\n transcription_plugins = registry.get_plugins_by_category(\"transcription\")\n \n # Get manager for specific operations\n mgr = registry.get_manager(\"transcription\")\n ```\n \"\"\"\n \n def __init__(self, config_dir: Optional[Path] = None):\n \"\"\"Initialize the unified plugin registry.\n \n Args:\n config_dir: Directory for plugin configuration files (default: 'configs')\n \"\"\"\n self._managers: Dict[str, Any] = {} # category -> manager\n \"Initialize the unified plugin registry.\n\nArgs:\n config_dir: Directory for plugin configuration files (default: 'configs')\"\n \n def register_plugin_manager(\n self,\n category: str, # Category name (e.g., \"transcription\", \"llm\")\n manager: Any, # Domain-specific plugin manager\n display_name: Optional[str] = None, # Display name for UI\n auto_discover: bool = True # Automatically discover plugins?\n ) -> List[PluginMetadata]: # List of discovered plugin metadata\n \"Register a domain-specific plugin manager.\n\nArgs:\n category: String category (e.g., 'transcription')\n manager: The domain-specific plugin manager instance\n display_name: Optional display name for UI\n auto_discover: Automatically discover and register plugins\n\nReturns:\n List of discovered plugin metadata\"\n \n def get_manager(\n self,\n category: str, # Category name\n manager_type: Optional[Type[T]] = None # Optional type hint\n ) -> Optional[T]: # Plugin manager instance\n \"Get plugin manager for a specific category.\n\nArgs:\n category: Category name (e.g., 'transcription')\n manager_type: Optional type hint for IDE autocomplete\n\nReturns:\n Plugin manager instance if found, None otherwise\"\n \n def get_categories(self) -> List[str]:\n \"\"\"Get all registered categories.\n \n Returns:\n Sorted list of category names\n \"\"\"\n return sorted(self._categories.keys())\n \n def get_category_display_name(self, category: str) -> str\n \"Get all registered categories.\n\nReturns:\n Sorted list of category names\"\n \n def get_category_display_name(self, category: str) -> str:\n \"\"\"Get display name for a category.\n \n Args:\n category: Category name\n \n Returns:\n Display name or category name if not set\n \"\"\"\n return self._categories.get(category, category.title())\n \n def get_plugin(self, unique_id: str) -> Optional[PluginMetadata]\n \"Get display name for a category.\n\nArgs:\n category: Category name\n\nReturns:\n Display name or category name if not set\"\n \n def get_plugin(self, unique_id: str) -> Optional[PluginMetadata]:\n \"\"\"Get plugin metadata by unique ID.\n \n Args:\n unique_id: Plugin unique identifier (format: 'category_name')\n \n Returns:\n Plugin metadata if found, None otherwise\n \"\"\"\n return self._plugins.get(unique_id)\n \n def get_plugins_by_category(self, category: str) -> List[PluginMetadata]\n \"Get plugin metadata by unique ID.\n\nArgs:\n unique_id: Plugin unique identifier (format: 'category_name')\n\nReturns:\n Plugin metadata if found, None otherwise\"\n \n def get_plugins_by_category(self, category: str) -> List[PluginMetadata]:\n \"\"\"Get all plugins in a category.\n \n Args:\n category: Category name\n \n Returns:\n List of plugin metadata for the category\n \"\"\"\n return [p for p in self._plugins.values() if p.category == category]\n \n def get_all_plugins(self) -> List[PluginMetadata]\n \"Get all plugins in a category.\n\nArgs:\n category: Category name\n\nReturns:\n List of plugin metadata for the category\"\n \n def get_all_plugins(self) -> List[PluginMetadata]:\n \"\"\"Get all plugins across all categories.\n \n Returns:\n List of all plugin metadata\n \"\"\"\n return list(self._plugins.values())\n \n def get_categories_with_plugins(self) -> List[str]\n \"Get all plugins across all categories.\n\nReturns:\n List of all plugin metadata\"\n \n def get_categories_with_plugins(self) -> List[str]:\n \"\"\"Get categories that have registered plugins.\n \n Returns:\n Sorted list of categories with plugins\n \"\"\"\n categories = set(p.category for p in self._plugins.values())\n return sorted(categories)\n \n def load_plugin_config(self, unique_id: str) -> Dict[str, Any]\n \"Get categories that have registered plugins.\n\nReturns:\n Sorted list of categories with plugins\"\n \n def load_plugin_config(self, unique_id: str) -> Dict[str, Any]:\n \"\"\"Load saved configuration for a plugin.\n \n Args:\n unique_id: Plugin unique identifier\n \n Returns:\n Configuration dictionary (empty if no config exists)\n \"\"\"\n config_file = self._config_dir / f\"{unique_id}.json\"\n if config_file.exists()\n \"Load saved configuration for a plugin.\n\nArgs:\n unique_id: Plugin unique identifier\n\nReturns:\n Configuration dictionary (empty if no config exists)\"\n \n def save_plugin_config(self, unique_id: str, config: Dict[str, Any]) -> bool:\n \"\"\"Save configuration for a plugin.\n \n Args:\n unique_id: Plugin unique identifier\n config: Configuration dictionary to save\n \n Returns:\n True if save succeeded, False otherwise\n \"\"\"\n try\n \"Save configuration for a plugin.\n\nArgs:\n unique_id: Plugin unique identifier\n config: Configuration dictionary to save\n\nReturns:\n True if save succeeded, False otherwise\"\n \n def delete_plugin_config(self, unique_id: str) -> bool:\n \"\"\"Delete saved configuration for a plugin.\n \n Args:\n unique_id: Plugin unique identifier\n \n Returns:\n True if deletion succeeded, False otherwise\n \"\"\"\n try\n \"Delete saved configuration for a plugin.\n\nArgs:\n unique_id: Plugin unique identifier\n\nReturns:\n True if deletion succeeded, False otherwise\"\n````\n\n#### Variables\n\n``` python\nT\n```\n",
"bugtrack_url": null,
"license": "Apache Software License 2.0",
"summary": "FastHTML plugin registry with persistence, categories, and settings UI integration for managing application plugins.",
"version": "0.0.1",
"project_urls": {
"Homepage": "https://github.com/cj-mills/cjm-fasthtml-plugins"
},
"split_keywords": [
"nbdev",
"jupyter",
"notebook",
"python"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "0682c3ce21a2d858281531d8240c8f769d1c2b6364ac0c8e76525c720eedd151",
"md5": "1b1ecd5dac64e2338a4d48868bf0ee7d",
"sha256": "3361fb9b1cea3df3e47badd740f8a03756ed49c6201d3635994d6ffd25270f05"
},
"downloads": -1,
"filename": "cjm_fasthtml_plugins-0.0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1b1ecd5dac64e2338a4d48868bf0ee7d",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 23736,
"upload_time": "2025-10-22T23:20:49",
"upload_time_iso_8601": "2025-10-22T23:20:49.109704Z",
"url": "https://files.pythonhosted.org/packages/06/82/c3ce21a2d858281531d8240c8f769d1c2b6364ac0c8e76525c720eedd151/cjm_fasthtml_plugins-0.0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "f5d3570deaf0ab5e769fa16055d46429ec14e5366e81150b3cb9dd87df30bf3c",
"md5": "14c26443c92c81f876061af7d50972c3",
"sha256": "7927bc0cecedaf8d0b5e9c941eeda83a8c31ac0e9181dc9b5dea5b73a6d435b8"
},
"downloads": -1,
"filename": "cjm_fasthtml_plugins-0.0.1.tar.gz",
"has_sig": false,
"md5_digest": "14c26443c92c81f876061af7d50972c3",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 31245,
"upload_time": "2025-10-22T23:20:50",
"upload_time_iso_8601": "2025-10-22T23:20:50.602449Z",
"url": "https://files.pythonhosted.org/packages/f5/d3/570deaf0ab5e769fa16055d46429ec14e5366e81150b3cb9dd87df30bf3c/cjm_fasthtml_plugins-0.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-22 23:20:50",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cj-mills",
"github_project": "cjm-fasthtml-plugins",
"github_not_found": true,
"lcname": "cjm-fasthtml-plugins"
}
JSON
