Coverage for routers/predictive.py: 100%

1from fastapi import APIRouter, Depends, HTTPException, Response 

2import duckdb 

3from typing import List 

4 

5from database import get_db 

6import schemas 

7 

8router = APIRouter( 

9 prefix="/api/v1", 

10 tags=["Predictive Reliability"] 

11) 

12 

13@router.get("/generations/{generation_id}/reliability", response_model=schemas.ReliabilitySummaryResponse) 

14def get_reliability_summary(generation_id: int, response: Response, db: duckdb.DuckDBPyConnection = Depends(get_db)): 

15 # Highly static data (7 days cache) 

16 response.headers["Cache-Control"] = "public, max-age=604800" 

17 

18 query = """ 

19 SELECT category, incidence_rate 

20 FROM predictive_models 

21 WHERE generation_id = ? 

22 ORDER BY incidence_rate DESC 

23 LIMIT 3 

24 """ 

25 db.execute(query, [generation_id]) 

26 categories = db.fetchall() 

27 columns = [desc[0] for desc in db.description] 

28 

29 return { 

30 "generation_id": generation_id, 

31 "top_categories": [ 

32 { 

33 "category": dict(zip(columns, row))["category"], 

34 "incidence_rate": dict(zip(columns, row))["incidence_rate"] 

35 } 

36 for row in categories 

37 ] 

38 } 

39 

40@router.get("/generations/{generation_id}/reliability/granular", response_model=List[schemas.DefectIncidenceResponse]) 

41def get_reliability_granular(generation_id: int, response: Response, db: duckdb.DuckDBPyConnection = Depends(get_db)): 

42 # Highly static data (7 days cache) 

43 response.headers["Cache-Control"] = "public, max-age=604800" 

44 

45 query = """ 

46 SELECT category, defect_description, incidence_rate 

47 FROM predictive_model_defects 

48 WHERE generation_id = ? 

49 ORDER BY incidence_rate DESC 

50 """ 

51 db.execute(query, [generation_id]) 

52 defects = db.fetchall() 

53 columns = [desc[0] for desc in db.description] 

54 

55 return [ 

56 { 

57 "category": dict(zip(columns, row))["category"], 

58 "defect_description": dict(zip(columns, row))["defect_description"], 

59 "incidence_rate": dict(zip(columns, row))["incidence_rate"] 

60 } 

61 for row in defects 

62 ] 

63 

64@router.get("/generations/{generation_id}/population", response_model=schemas.PopulationResponse) 

65def get_population(generation_id: int, response: Response, db: duckdb.DuckDBPyConnection = Depends(get_db)): 

66 # Highly static data (7 days cache) 

67 response.headers["Cache-Control"] = "public, max-age=604800" 

68 

69 query = """ 

70 SELECT total_on_road, total_sorn 

71 FROM vehicle_population 

72 WHERE generation_id = ? 

73 """ 

74 db.execute(query, [generation_id]) 

75 result = db.fetchone() 

76 

77 if not result: 

78 raise HTTPException(status_code=404, detail="Population data not found for this generation") 

79 

80 columns = [desc[0] for desc in db.description] 

81 pop = dict(zip(columns, result)) 

82 

83 return { 

84 "generation_id": generation_id, 

85 "total_on_road": pop["total_on_road"], 

86 "total_sorn": pop["total_sorn"] 

87 }