88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
import vna_system.core.singletons as singletons
|
|
|
|
router = APIRouter(prefix="/api/v1", tags=["acquisition"])
|
|
|
|
|
|
@router.get("/acquisition/status")
|
|
async def get_acquisition_status():
|
|
"""Get current acquisition status."""
|
|
acquisition = singletons.vna_data_acquisition_instance
|
|
|
|
return {
|
|
"running": acquisition.is_running,
|
|
"paused": acquisition.is_paused,
|
|
"continuous_mode": acquisition.is_continuous_mode,
|
|
"sweep_count": acquisition._sweep_buffer._sweep_counter if hasattr(acquisition._sweep_buffer, '_sweep_counter') else 0
|
|
}
|
|
|
|
|
|
@router.post("/acquisition/start")
|
|
async def start_acquisition():
|
|
"""Start data acquisition."""
|
|
try:
|
|
acquisition = singletons.vna_data_acquisition_instance
|
|
|
|
if not acquisition.is_running:
|
|
# Start thread if not running
|
|
acquisition.start()
|
|
|
|
# Set to continuous mode (also resumes if paused)
|
|
acquisition.set_continuous_mode(True)
|
|
return {"success": True, "message": "Acquisition started"}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/acquisition/stop")
|
|
async def stop_acquisition():
|
|
"""Stop/pause data acquisition."""
|
|
try:
|
|
acquisition = singletons.vna_data_acquisition_instance
|
|
if not acquisition.is_running:
|
|
return {"success": True, "message": "Acquisition already stopped"}
|
|
|
|
# Just pause instead of full stop - keeps thread alive for restart
|
|
acquisition.pause()
|
|
return {"success": True, "message": "Acquisition stopped"}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/acquisition/single-sweep")
|
|
async def trigger_single_sweep():
|
|
"""Trigger a single sweep. Automatically starts acquisition if needed."""
|
|
try:
|
|
acquisition = singletons.vna_data_acquisition_instance
|
|
|
|
if not acquisition.is_running:
|
|
# Start acquisition if not running
|
|
acquisition.start()
|
|
|
|
acquisition.trigger_single_sweep()
|
|
return {"success": True, "message": "Single sweep triggered"}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/acquisition/latest-sweep")
|
|
async def get_latest_sweep():
|
|
"""Get the latest sweep data."""
|
|
try:
|
|
acquisition = singletons.vna_data_acquisition_instance
|
|
latest_sweep = acquisition._sweep_buffer.get_latest_sweep()
|
|
|
|
if not latest_sweep:
|
|
return {"sweep": None, "message": "No sweep data available"}
|
|
|
|
return {
|
|
"sweep": {
|
|
"sweep_number": latest_sweep.sweep_number,
|
|
"timestamp": latest_sweep.timestamp,
|
|
"total_points": latest_sweep.total_points,
|
|
"points": latest_sweep.points[:10] if len(latest_sweep.points) > 10 else latest_sweep.points # Limit for API response
|
|
},
|
|
"message": f"Latest sweep #{latest_sweep.sweep_number} with {latest_sweep.total_points} points"
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e)) |