simulation_plan.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738
  1. """Simulation plan model."""
  2. import json
  3. from datetime import datetime
  4. from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
  5. from sqlalchemy.orm import relationship, backref
  6. from ..database import Base
  7. class SimulationPlan(Base):
  8. """A simulation plan (simulation_plan.json format, compatible with local executor)."""
  9. __tablename__ = "simulation_plans"
  10. id = Column(Integer, primary_key=True, index=True)
  11. project_id = Column(Integer, ForeignKey("projects.id"), nullable=False, index=True)
  12. name = Column(String(200), nullable=False)
  13. plan_id = Column(String(100), unique=True, index=True) # e.g. SP-20260827-001
  14. status = Column(String(20), default="draft") # draft / confirmed / executing / completed / failed
  15. plan_json = Column(Text, nullable=False) # Full simulation_plan.json content
  16. variables_summary = Column(Text, default="") # JSON: variable names and ranges for display
  17. estimated_points = Column(Integer, default=0)
  18. estimated_time_min = Column(Integer, default=0)
  19. notes = Column(Text, default="")
  20. created_at = Column(DateTime, default=datetime.utcnow)
  21. updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
  22. # F2 fix: cascade delete from project -> plans -> results
  23. project = relationship("Project", backref=backref("plans", cascade="all, delete-orphan"))
  24. results = relationship("SimulationResult", backref="plan", cascade="all, delete-orphan")
  25. def get_plan_dict(self) -> dict:
  26. try:
  27. return json.loads(self.plan_json) if self.plan_json else {}
  28. except (json.JSONDecodeError, TypeError):
  29. return {}
  30. def set_plan_dict(self, data: dict) -> None:
  31. self.plan_json = json.dumps(data, ensure_ascii=False, indent=2)