"""Project health calculation service. Provides functionality to calculate and retrieve project health metrics including risk scores, schedule status, and resource status. """ import uuid from datetime import datetime from typing import List, Optional, Dict, Any from sqlalchemy.orm import Session from app.models import Project, Task, TaskStatus, Blocker, ProjectHealth from app.schemas.project_health import ( RiskLevel, ScheduleStatus, ResourceStatus, ProjectHealthResponse, ProjectHealthWithDetails, ProjectHealthSummary, ProjectHealthDashboardResponse, ) # Constants for health score calculation BLOCKER_PENALTY_PER_ITEM = 10 BLOCKER_PENALTY_MAX = 30 OVERDUE_PENALTY_PER_ITEM = 5 OVERDUE_PENALTY_MAX = 30 COMPLETION_PENALTY_THRESHOLD = 50 COMPLETION_PENALTY_FACTOR = 0.4 COMPLETION_PENALTY_MAX = 20 # Risk level thresholds RISK_LOW_THRESHOLD = 80 RISK_MEDIUM_THRESHOLD = 60 RISK_HIGH_THRESHOLD = 40 # Schedule status thresholds SCHEDULE_AT_RISK_THRESHOLD = 2 # Resource status thresholds RESOURCE_CONSTRAINED_THRESHOLD = 2 def calculate_health_metrics(db: Session, project: Project) -> Dict[str, Any]: """ Calculate health metrics for a project. Args: db: Database session project: Project object to calculate metrics for Returns: Dictionary containing: - health_score: 0-100 integer - risk_level: low/medium/high/critical - schedule_status: on_track/at_risk/delayed - resource_status: adequate/constrained/overloaded - task_count: Total number of active tasks - completed_task_count: Number of completed tasks - blocker_count: Number of unresolved blockers - overdue_task_count: Number of overdue incomplete tasks """ # Fetch active tasks for this project tasks = db.query(Task).filter( Task.project_id == project.id, Task.is_deleted == False ).all() task_count = len(tasks) # Count completed tasks completed_task_count = sum( 1 for task in tasks if task.status and task.status.is_done ) # Count overdue tasks (incomplete with past due date) now = datetime.utcnow() overdue_task_count = sum( 1 for task in tasks if task.due_date and task.due_date < now and not (task.status and task.status.is_done) ) # Count unresolved blockers task_ids = [t.id for t in tasks] blocker_count = 0 if task_ids: blocker_count = db.query(Blocker).filter( Blocker.task_id.in_(task_ids), Blocker.resolved_at.is_(None) ).count() # Calculate completion rate completion_rate = 0.0 if task_count > 0: completion_rate = (completed_task_count / task_count) * 100 # Calculate health score (start at 100, subtract penalties) health_score = 100 # Apply blocker penalty blocker_penalty = min(blocker_count * BLOCKER_PENALTY_PER_ITEM, BLOCKER_PENALTY_MAX) health_score -= blocker_penalty # Apply overdue penalty overdue_penalty = min(overdue_task_count * OVERDUE_PENALTY_PER_ITEM, OVERDUE_PENALTY_MAX) health_score -= overdue_penalty # Apply completion penalty (if below threshold) if task_count > 0 and completion_rate < COMPLETION_PENALTY_THRESHOLD: completion_penalty = int( (COMPLETION_PENALTY_THRESHOLD - completion_rate) * COMPLETION_PENALTY_FACTOR ) health_score -= min(completion_penalty, COMPLETION_PENALTY_MAX) # Ensure health score stays within bounds health_score = max(0, min(100, health_score)) # Determine risk level based on health score risk_level = _determine_risk_level(health_score) # Determine schedule status based on overdue count schedule_status = _determine_schedule_status(overdue_task_count) # Determine resource status based on blocker count resource_status = _determine_resource_status(blocker_count) return { "health_score": health_score, "risk_level": risk_level, "schedule_status": schedule_status, "resource_status": resource_status, "task_count": task_count, "completed_task_count": completed_task_count, "blocker_count": blocker_count, "overdue_task_count": overdue_task_count, } def _determine_risk_level(health_score: int) -> str: """Determine risk level based on health score.""" if health_score >= RISK_LOW_THRESHOLD: return "low" elif health_score >= RISK_MEDIUM_THRESHOLD: return "medium" elif health_score >= RISK_HIGH_THRESHOLD: return "high" else: return "critical" def _determine_schedule_status(overdue_task_count: int) -> str: """Determine schedule status based on overdue task count.""" if overdue_task_count == 0: return "on_track" elif overdue_task_count <= SCHEDULE_AT_RISK_THRESHOLD: return "at_risk" else: return "delayed" def _determine_resource_status(blocker_count: int) -> str: """Determine resource status based on blocker count.""" if blocker_count == 0: return "adequate" elif blocker_count <= RESOURCE_CONSTRAINED_THRESHOLD: return "constrained" else: return "overloaded" def get_or_create_project_health(db: Session, project: Project) -> ProjectHealth: """ Get existing project health record or create a new one. Args: db: Database session project: Project object Returns: ProjectHealth record """ health = db.query(ProjectHealth).filter( ProjectHealth.project_id == project.id ).first() if not health: health = ProjectHealth( id=str(uuid.uuid4()), project_id=project.id ) db.add(health) return health def update_project_health( db: Session, project: Project, metrics: Dict[str, Any] ) -> ProjectHealth: """ Update project health record with calculated metrics. Args: db: Database session project: Project object metrics: Calculated health metrics Returns: Updated ProjectHealth record """ health = get_or_create_project_health(db, project) health.health_score = metrics["health_score"] health.risk_level = metrics["risk_level"] health.schedule_status = metrics["schedule_status"] health.resource_status = metrics["resource_status"] return health def get_project_health( db: Session, project_id: str ) -> Optional[ProjectHealthWithDetails]: """ Get health information for a single project. Args: db: Database session project_id: Project ID Returns: ProjectHealthWithDetails or None if project not found """ project = db.query(Project).filter(Project.id == project_id).first() if not project: return None metrics = calculate_health_metrics(db, project) health = update_project_health(db, project, metrics) db.commit() db.refresh(health) return _build_health_with_details(project, health, metrics) def get_all_projects_health( db: Session, status_filter: Optional[str] = "active" ) -> ProjectHealthDashboardResponse: """ Get health information for all projects. Args: db: Database session status_filter: Filter projects by status (default: "active") Returns: ProjectHealthDashboardResponse with projects list and summary """ query = db.query(Project) if status_filter: query = query.filter(Project.status == status_filter) projects = query.all() projects_health: List[ProjectHealthWithDetails] = [] for project in projects: metrics = calculate_health_metrics(db, project) health = update_project_health(db, project, metrics) project_health = _build_health_with_details(project, health, metrics) projects_health.append(project_health) db.commit() # Calculate summary statistics summary = _calculate_summary(projects_health) return ProjectHealthDashboardResponse( projects=projects_health, summary=summary ) def _build_health_with_details( project: Project, health: ProjectHealth, metrics: Dict[str, Any] ) -> ProjectHealthWithDetails: """Build ProjectHealthWithDetails from project, health, and metrics.""" return ProjectHealthWithDetails( id=health.id, project_id=project.id, health_score=metrics["health_score"], risk_level=RiskLevel(metrics["risk_level"]), schedule_status=ScheduleStatus(metrics["schedule_status"]), resource_status=ResourceStatus(metrics["resource_status"]), last_updated=health.last_updated or datetime.utcnow(), project_title=project.title, project_status=project.status, owner_name=project.owner.name if project.owner else None, space_name=project.space.name if project.space else None, task_count=metrics["task_count"], completed_task_count=metrics["completed_task_count"], blocker_count=metrics["blocker_count"], overdue_task_count=metrics["overdue_task_count"], ) def _calculate_summary( projects_health: List[ProjectHealthWithDetails] ) -> ProjectHealthSummary: """Calculate summary statistics for health dashboard.""" total_projects = len(projects_health) healthy_count = sum(1 for p in projects_health if p.health_score >= 80) at_risk_count = sum(1 for p in projects_health if 50 <= p.health_score < 80) critical_count = sum(1 for p in projects_health if p.health_score < 50) average_health_score = 0.0 if total_projects > 0: average_health_score = sum(p.health_score for p in projects_health) / total_projects projects_with_blockers = sum(1 for p in projects_health if p.blocker_count > 0) projects_delayed = sum( 1 for p in projects_health if p.schedule_status == ScheduleStatus.DELAYED ) return ProjectHealthSummary( total_projects=total_projects, healthy_count=healthy_count, at_risk_count=at_risk_count, critical_count=critical_count, average_health_score=round(average_health_score, 1), projects_with_blockers=projects_with_blockers, projects_delayed=projects_delayed, ) class HealthService: """ Service class for project health operations. Provides a class-based interface for health calculations, following the service pattern used in the codebase. """ def __init__(self, db: Session): """Initialize HealthService with database session.""" self.db = db def calculate_metrics(self, project: Project) -> Dict[str, Any]: """Calculate health metrics for a project.""" return calculate_health_metrics(self.db, project) def get_project_health(self, project_id: str) -> Optional[ProjectHealthWithDetails]: """Get health information for a single project.""" return get_project_health(self.db, project_id) def get_dashboard( self, status_filter: Optional[str] = "active" ) -> ProjectHealthDashboardResponse: """Get health dashboard for all projects.""" return get_all_projects_health(self.db, status_filter) def refresh_project_health(self, project: Project) -> ProjectHealth: """Refresh and persist health data for a project.""" metrics = calculate_health_metrics(self.db, project) health = update_project_health(self.db, project, metrics) self.db.commit() self.db.refresh(health) return health