almosther isolating soleprint
This commit is contained in:
buenosairesam
370
station/tools/modelgen/model_generator.py
Normal file
370
station/tools/modelgen/model_generator.py
Normal file
@@ -0,0 +1,370 @@
|
||||
"""
|
||||
Model Generator
|
||||
|
||||
Generic model generation from configuration files.
|
||||
Supports multiple output formats and is extensible for bidirectional conversion.
|
||||
|
||||
Output formats:
|
||||
- pydantic: Pydantic BaseModel classes
|
||||
- django: Django ORM models (planned)
|
||||
- prisma: Prisma schema (planned)
|
||||
- sqlalchemy: SQLAlchemy models (planned)
|
||||
|
||||
Future: Extract models FROM existing codebases (reverse direction)
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
from typing import Dict, Type
|
||||
|
||||
from .config_loader import ConfigLoader
|
||||
|
||||
|
||||
class BaseModelWriter(ABC):
|
||||
"""Abstract base for model output writers."""
|
||||
|
||||
@abstractmethod
|
||||
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
||||
"""Write models to the specified path."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def file_extension(self) -> str:
|
||||
"""Return the file extension for this format."""
|
||||
pass
|
||||
|
||||
|
||||
class PydanticWriter(BaseModelWriter):
|
||||
"""Generates Pydantic model files."""
|
||||
|
||||
def file_extension(self) -> str:
|
||||
return ".py"
|
||||
|
||||
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
||||
"""Write Pydantic models to output_path."""
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
content = self._generate_content(config)
|
||||
output_path.write_text(content)
|
||||
|
||||
def _generate_content(self, config: ConfigLoader) -> str:
|
||||
"""Generate the Pydantic models file content."""
|
||||
|
||||
# Get component names from config
|
||||
config_comp = config.get_shared_component("config")
|
||||
data_comp = config.get_shared_component("data")
|
||||
|
||||
data_flow_sys = config.get_system("data_flow")
|
||||
doc_sys = config.get_system("documentation")
|
||||
exec_sys = config.get_system("execution")
|
||||
|
||||
connector_comp = config.get_component("data_flow", "connector")
|
||||
pulse_comp = config.get_component("data_flow", "composed")
|
||||
|
||||
pattern_comp = config.get_component("documentation", "pattern")
|
||||
doc_composed = config.get_component("documentation", "composed")
|
||||
|
||||
tool_comp = config.get_component("execution", "utility")
|
||||
monitor_comp = config.get_component("execution", "watcher")
|
||||
cabinet_comp = config.get_component("execution", "container")
|
||||
exec_composed = config.get_component("execution", "composed")
|
||||
|
||||
return f'''"""
|
||||
Pydantic models - Generated from {config.framework.name}.config.json
|
||||
|
||||
DO NOT EDIT MANUALLY - Regenerate from config
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
from typing import List, Literal, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class Status(str, Enum):
|
||||
PENDING = "pending"
|
||||
PLANNED = "planned"
|
||||
BUILDING = "building"
|
||||
DEV = "dev"
|
||||
LIVE = "live"
|
||||
READY = "ready"
|
||||
|
||||
|
||||
class System(str, Enum):
|
||||
{data_flow_sys.name.upper()} = "{data_flow_sys.name}"
|
||||
{doc_sys.name.upper()} = "{doc_sys.name}"
|
||||
{exec_sys.name.upper()} = "{exec_sys.name}"
|
||||
|
||||
|
||||
class ToolType(str, Enum):
|
||||
APP = "app"
|
||||
CLI = "cli"
|
||||
|
||||
|
||||
# === Shared Components ===
|
||||
|
||||
|
||||
class {config_comp.title}(BaseModel):
|
||||
"""{config_comp.description}. Shared across {data_flow_sys.name}, {exec_sys.name}."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
config_path: Optional[str] = None
|
||||
|
||||
|
||||
class {data_comp.title}(BaseModel):
|
||||
"""{data_comp.description}. Shared across all systems."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
source_template: Optional[str] = None
|
||||
data_path: Optional[str] = None
|
||||
|
||||
|
||||
# === System-Specific Components ===
|
||||
|
||||
|
||||
class {connector_comp.title}(BaseModel):
|
||||
"""{connector_comp.description} ({data_flow_sys.name})."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
system: Literal["{data_flow_sys.name}"] = "{data_flow_sys.name}"
|
||||
mock: Optional[bool] = None
|
||||
description: Optional[str] = None
|
||||
|
||||
|
||||
class {pattern_comp.title}(BaseModel):
|
||||
"""{pattern_comp.description} ({doc_sys.name})."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
template_path: Optional[str] = None
|
||||
system: Literal["{doc_sys.name}"] = "{doc_sys.name}"
|
||||
|
||||
|
||||
class {tool_comp.title}(BaseModel):
|
||||
"""{tool_comp.description} ({exec_sys.name})."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
||||
type: Optional[ToolType] = None
|
||||
description: Optional[str] = None
|
||||
path: Optional[str] = None
|
||||
url: Optional[str] = None
|
||||
cli: Optional[str] = None
|
||||
|
||||
|
||||
class {monitor_comp.title}(BaseModel):
|
||||
"""{monitor_comp.description} ({exec_sys.name})."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
||||
|
||||
|
||||
class {cabinet_comp.title}(BaseModel):
|
||||
"""{cabinet_comp.description} ({exec_sys.name})."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
tools: List[{tool_comp.title}] = Field(default_factory=list)
|
||||
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
||||
|
||||
|
||||
# === Composed Types ===
|
||||
|
||||
|
||||
class {pulse_comp.title}(BaseModel):
|
||||
"""{pulse_comp.description} ({data_flow_sys.name}). Formula: {pulse_comp.formula}."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
{connector_comp.name}: Optional[{connector_comp.title}] = None
|
||||
{config_comp.name}: Optional[{config_comp.title}] = None
|
||||
{data_comp.name}: Optional[{data_comp.title}] = None
|
||||
system: Literal["{data_flow_sys.name}"] = "{data_flow_sys.name}"
|
||||
|
||||
|
||||
class {doc_composed.title}(BaseModel):
|
||||
"""{doc_composed.description} ({doc_sys.name}). Formula: {doc_composed.formula}."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
template: Optional[{pattern_comp.title}] = None
|
||||
{data_comp.name}: Optional[{data_comp.title}] = None
|
||||
output_{data_comp.name}: Optional[{data_comp.title}] = None
|
||||
system: Literal["{doc_sys.name}"] = "{doc_sys.name}"
|
||||
|
||||
|
||||
class {exec_composed.title}(BaseModel):
|
||||
"""{exec_composed.description} ({exec_sys.name}). Formula: {exec_composed.formula}."""
|
||||
|
||||
name: str # Unique identifier
|
||||
slug: str # URL-friendly identifier
|
||||
title: str # Display title for UI
|
||||
status: Optional[Status] = None
|
||||
cabinet: Optional[{cabinet_comp.title}] = None
|
||||
{config_comp.name}: Optional[{config_comp.title}] = None
|
||||
{data_comp.plural}: List[{data_comp.title}] = Field(default_factory=list)
|
||||
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
||||
|
||||
|
||||
# === Collection wrappers for JSON files ===
|
||||
|
||||
|
||||
class {config_comp.title}Collection(BaseModel):
|
||||
items: List[{config_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {data_comp.title}Collection(BaseModel):
|
||||
items: List[{data_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {connector_comp.title}Collection(BaseModel):
|
||||
items: List[{connector_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {pattern_comp.title}Collection(BaseModel):
|
||||
items: List[{pattern_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {tool_comp.title}Collection(BaseModel):
|
||||
items: List[{tool_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {monitor_comp.title}Collection(BaseModel):
|
||||
items: List[{monitor_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {cabinet_comp.title}Collection(BaseModel):
|
||||
items: List[{cabinet_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {pulse_comp.title}Collection(BaseModel):
|
||||
items: List[{pulse_comp.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {doc_composed.title}Collection(BaseModel):
|
||||
items: List[{doc_composed.title}] = Field(default_factory=list)
|
||||
|
||||
|
||||
class {exec_composed.title}Collection(BaseModel):
|
||||
items: List[{exec_composed.title}] = Field(default_factory=list)
|
||||
'''
|
||||
|
||||
|
||||
class DjangoWriter(BaseModelWriter):
|
||||
"""Generates Django model files (placeholder)."""
|
||||
|
||||
def file_extension(self) -> str:
|
||||
return ".py"
|
||||
|
||||
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
||||
raise NotImplementedError("Django model generation not yet implemented")
|
||||
|
||||
|
||||
class PrismaWriter(BaseModelWriter):
|
||||
"""Generates Prisma schema files (placeholder)."""
|
||||
|
||||
def file_extension(self) -> str:
|
||||
return ".prisma"
|
||||
|
||||
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
||||
raise NotImplementedError("Prisma schema generation not yet implemented")
|
||||
|
||||
|
||||
class SQLAlchemyWriter(BaseModelWriter):
|
||||
"""Generates SQLAlchemy model files (placeholder)."""
|
||||
|
||||
def file_extension(self) -> str:
|
||||
return ".py"
|
||||
|
||||
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
||||
raise NotImplementedError("SQLAlchemy model generation not yet implemented")
|
||||
|
||||
|
||||
# Registry of available writers
|
||||
WRITERS: Dict[str, Type[BaseModelWriter]] = {
|
||||
"pydantic": PydanticWriter,
|
||||
"django": DjangoWriter,
|
||||
"prisma": PrismaWriter,
|
||||
"sqlalchemy": SQLAlchemyWriter,
|
||||
}
|
||||
|
||||
|
||||
class ModelGenerator:
|
||||
"""
|
||||
Generates typed models from configuration.
|
||||
|
||||
This is the main entry point for model generation.
|
||||
Delegates to format-specific writers.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config: ConfigLoader,
|
||||
output_path: Path,
|
||||
output_format: str = "pydantic",
|
||||
):
|
||||
"""
|
||||
Initialize the generator.
|
||||
|
||||
Args:
|
||||
config: Loaded configuration
|
||||
output_path: Exact path where to write (file or directory depending on format)
|
||||
output_format: Output format (pydantic, django, prisma, sqlalchemy)
|
||||
"""
|
||||
self.config = config
|
||||
self.output_path = Path(output_path)
|
||||
self.output_format = output_format
|
||||
|
||||
if output_format not in WRITERS:
|
||||
raise ValueError(
|
||||
f"Unknown output format: {output_format}. "
|
||||
f"Available: {list(WRITERS.keys())}"
|
||||
)
|
||||
|
||||
self.writer = WRITERS[output_format]()
|
||||
|
||||
def generate(self) -> Path:
|
||||
"""
|
||||
Generate models to the specified output path.
|
||||
|
||||
Returns:
|
||||
Path to the generated file/directory
|
||||
"""
|
||||
# Determine output file path
|
||||
if self.output_path.suffix:
|
||||
# User specified a file path
|
||||
output_file = self.output_path
|
||||
else:
|
||||
# User specified a directory, add default filename
|
||||
output_file = self.output_path / f"__init__{self.writer.file_extension()}"
|
||||
|
||||
self.writer.write(self.config, output_file)
|
||||
print(f"Generated {self.output_format} models: {output_file}")
|
||||
return output_file
|
||||
|
||||
@classmethod
|
||||
def available_formats(cls) -> list:
|
||||
"""Return list of available output formats."""
|
||||
return list(WRITERS.keys())
|
||||
Reference in New Issue
Block a user