Add initial implementation of CalMiner with project structure, environment setup, and core features

- Create .env.example for environment variables
- Update README with project structure and development setup instructions
- Implement FastAPI application with API routes for scenarios and parameters
- Add database models for scenarios, parameters, and simulation results
- Introduce validation middleware for JSON requests
- Create services for running simulations and generating reports
- Add testing strategy and directory structure in documentation
This commit is contained in:
2025-10-20 18:37:57 +02:00
parent cb9749010f
commit 39c45e720c
20 changed files with 604 additions and 64 deletions

50
routes/parameters.py Normal file
View File

@@ -0,0 +1,50 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from config.database import SessionLocal
from models.parameters import Parameter
from models.scenario import Scenario
from pydantic import BaseModel
from typing import Optional, List
router = APIRouter(prefix="/api/parameters", tags=["parameters"])
class ParameterCreate(BaseModel):
scenario_id: int
name: str
value: float
class ParameterRead(ParameterCreate):
id: int
class Config:
orm_mode = True
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/", response_model=ParameterRead)
def create_parameter(param: ParameterCreate, db: Session = Depends(get_db)):
scen = db.query(Scenario).filter(Scenario.id == param.scenario_id).first()
if not scen:
raise HTTPException(status_code=404, detail="Scenario not found")
new_param = Parameter(scenario_id=param.scenario_id,
name=param.name, value=param.value)
db.add(new_param)
db.commit()
db.refresh(new_param)
return new_param
@router.get("/", response_model=List[ParameterRead])
def list_parameters(db: Session = Depends(get_db)):
return db.query(Parameter).all()

23
routes/reporting.py Normal file
View File

@@ -0,0 +1,23 @@
from fastapi import APIRouter, HTTPException, Depends
from typing import Dict, Any
from services.reporting import generate_report
from sqlalchemy.orm import Session
from config.database import SessionLocal
router = APIRouter(prefix="/api/reporting", tags=["Reporting"])
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/summary", response_model=Dict[str, float])
async def summary_report(results: Any):
# Expect a list of simulation result dicts
if not isinstance(results, list):
raise HTTPException(status_code=400, detail="Invalid input format")
report = generate_report(results)
return report

52
routes/scenarios.py Normal file
View File

@@ -0,0 +1,52 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from config.database import SessionLocal
from models.scenario import Scenario
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
router = APIRouter(prefix="/api/scenarios", tags=["scenarios"])
# Pydantic schemas
class ScenarioCreate(BaseModel):
name: str
description: Optional[str] = None
class ScenarioRead(ScenarioCreate):
id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/", response_model=ScenarioRead)
def create_scenario(scenario: ScenarioCreate, db: Session = Depends(get_db)):
db_s = db.query(Scenario).filter(Scenario.name == scenario.name).first()
if db_s:
raise HTTPException(status_code=400, detail="Scenario already exists")
new_s = Scenario(name=scenario.name, description=scenario.description)
db.add(new_s)
db.commit()
db.refresh(new_s)
return new_s
@router.get("/", response_model=list[ScenarioRead])
def list_scenarios(db: Session = Depends(get_db)):
return db.query(Scenario).all()

25
routes/simulations.py Normal file
View File

@@ -0,0 +1,25 @@
from fastapi import APIRouter, HTTPException, Depends
from typing import List
from services.simulation import run_simulation
from sqlalchemy.orm import Session
from config.database import SessionLocal
router = APIRouter(prefix="/api/simulations", tags=["Simulations"])
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/run", response_model=List[dict])
async def simulate(params: List[dict], iterations: int = 1000, db: Session = Depends(get_db)):
if not params:
raise HTTPException(status_code=400, detail="No parameters provided")
# TODO: you might use db to fetch scenario info or persist results
results = run_simulation(params, iterations)
return results