Spaces:
Runtime error
Runtime error
File size: 4,249 Bytes
eb5a9e1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 |
"""
Base Plugin Class
Defines the interface that all plugins must implement.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Any, Optional, List
from pathlib import Path
from PIL import Image
import numpy as np
from loguru import logger
@dataclass
class PluginMetadata:
"""Metadata for a plugin."""
name: str
version: str
description: str
author: str
requires: List[str] = None # Required dependencies
category: str = "general" # Plugin category
enabled: bool = True
priority: int = 50 # Execution priority (lower = earlier)
def __post_init__(self):
if self.requires is None:
self.requires = []
class BasePlugin(ABC):
"""
Base class for all DeepVision plugins.
All plugins must inherit from this class and implement
the analyze() method.
"""
def __init__(self):
"""Initialize plugin."""
self._metadata: Optional[PluginMetadata] = None
self._initialized = False
self._enabled = True
logger.debug(f"Plugin {self.__class__.__name__} created")
@property
@abstractmethod
def metadata(self) -> PluginMetadata:
"""
Return plugin metadata.
Returns:
PluginMetadata instance
"""
pass
@abstractmethod
def initialize(self) -> None:
"""
Initialize the plugin.
This method is called when the plugin is loaded.
Use it to load models, initialize resources, etc.
"""
pass
@abstractmethod
def analyze(
self,
media: Any,
media_path: Path
) -> Dict[str, Any]:
"""
Analyze media and return results.
Args:
media: Processed media (PIL Image or numpy array)
media_path: Path to the media file
Returns:
Dictionary with analysis results
"""
pass
def cleanup(self) -> None:
"""
Clean up resources when plugin is unloaded.
Override this method to release resources like
model memory, file handles, etc.
"""
logger.debug(f"Cleaning up plugin {self.metadata.name}")
def validate_input(self, media: Any) -> bool:
"""
Validate input media.
Args:
media: Media to validate
Returns:
True if valid, False otherwise
"""
if isinstance(media, Image.Image):
return True
elif isinstance(media, np.ndarray):
return True
else:
logger.warning(
f"Plugin {self.metadata.name} received unsupported media type: "
f"{type(media)}"
)
return False
def get_config(self) -> Dict[str, Any]:
"""
Get plugin configuration.
Returns:
Configuration dictionary
"""
return {
"name": self.metadata.name,
"version": self.metadata.version,
"enabled": self._enabled,
"initialized": self._initialized,
}
def set_enabled(self, enabled: bool) -> None:
"""
Enable or disable the plugin.
Args:
enabled: True to enable, False to disable
"""
self._enabled = enabled
logger.info(
f"Plugin {self.metadata.name} "
f"{'enabled' if enabled else 'disabled'}"
)
def is_enabled(self) -> bool:
"""
Check if plugin is enabled.
Returns:
True if enabled
"""
return self._enabled
def is_initialized(self) -> bool:
"""
Check if plugin is initialized.
Returns:
True if initialized
"""
return self._initialized
def __repr__(self) -> str:
"""String representation."""
return (
f"{self.__class__.__name__}("
f"name={self.metadata.name}, "
f"version={self.metadata.version}, "
f"enabled={self._enabled})"
)
|