Hardware Domain¶

def calculate_boosted_scores(self, profile: HardwareProfile) -> Dict[str, Any]:
    """Calculate UI-friendly scores with +20 boost for frontend display.

    Returns both raw scores (actual hardware capability) and boosted scores
    (UI-friendly display with +20 point boost, capped at 100). This makes
    the boost transparent while maintaining user-friendly presentation.

    Args:
        profile: HardwareProfile entity with original scores.

    Returns:
        Dict[str, Any]: Dictionary with both raw and boosted scores/labels.

    Example:
        >>> profile = service.get_or_create_profile()
        >>> scores = service.calculate_boosted_scores(profile)
        >>> print(scores)
        {
            "raw_inference_score": 65.0,
            "boosted_inference_score": 85.0,  # min(65 + 20, 100)
            "global_inference_label": "Excellent",  # Based on boosted
            ...
        }
    """
    logger.debug("Calculating raw and boosted scores")

    # Raw score (actual hardware capability)
    raw_inf = profile.global_inference_score

    # Boosted score for UI (+ 20 points, capped at 100)
    boosted_inf = min(100.0, raw_inf + 20.0)

    result = {
        # Raw score (engine output, no modification)
        "raw_inference_score": raw_inf,
        # Boosted score for UI display
        "boosted_inference_score": boosted_inf,
        "global_inference_score": boosted_inf,  # Alias for backward compat
        # Label based on boosted score
        "global_inference_label": self._get_label(boosted_inf),
        # Component scores (raw, no boost needed for internal metrics)
        "cpu_score": profile.cpu_score,
        "memory_score": profile.memory_score,
        "gpu_score": profile.gpu_score if profile.gpu_score is not None else 0.0,
    }

    # Hardware-fit model size window (billions of params) for the UI's
    # "Models For You" recommendations (#86), derived from real usable memory.
    param_min, param_max = recommended_param_range(profile)
    result["recommended_param_min"] = param_min
    result["recommended_param_max"] = param_max

    logger.debug(f"Scores calculated: raw_inf={raw_inf:.1f}, boosted_inf={boosted_inf:.1f}")
    return result

def get_or_create_profile(self) -> HardwareProfile:
    """Get cached hardware profile or detect new one.

    Retrieves existing hardware profile from database. If none exists or
    if the cached profile backend doesn't match the current engine,
    performs hardware detection and creates new profile.

    Returns:
        HardwareProfile: Existing or newly created profile.

    Raises:
        HardwareException: If hardware detection or profile creation fails.

    Note:
        Does NOT commit. Caller must commit transaction.

    Example:
        >>> service = Hardware_Service(repo)
        >>> profile = service.get_or_create_profile()
        >>> print(profile.backend_type)  # "mlx", "cuda", or "cpu"
    """
    try:
        logger.info("Getting or creating hardware profile")

        # Try to get existing profile
        profile = self.repository.get_profile()

        # Extract backend type from current engine name
        current_backend = config.LLM_Engine.__name__.lower().replace("_engine", "")

        if profile:
            stored_version = getattr(profile, "profiling_version", None)
            if profile.backend_type != current_backend:
                # No profile exists or backend mismatch, detect hardware
                logger.info(
                    f"Backend mismatch: cached={profile.backend_type}, current={current_backend}. Re-detecting."
                )
                self.repository.delete_profile(profile)
            elif stored_version != PROFILING_LOGIC_VERSION:
                # Same machine, but the numbers were produced by profiling
                # logic we have since corrected (#365). Redo them, otherwise
                # the fix never reaches anyone who already ran the app.
                logger.info(
                    f"Profiling logic changed: cached=v{stored_version}, "
                    f"current=v{PROFILING_LOGIC_VERSION}. Re-detecting."
                )
                self.repository.delete_profile(profile)
            else:
                logger.info(f"Using cached hardware profile: backend={profile.backend_type}")
                return profile
        else:
            logger.info("No cached profile found, detecting hardware")

        hardware_data = self._detect_hardware()

        # Create new profile
        profile = self.repository.create_profile(hardware_data)
        logger.info(f"Hardware profile created: backend={profile.backend_type}")

        return profile

    except HardwareException:
        raise  # logged where it was raised
    except Exception as e:
        logger.exception(f"Failed to get or create hardware profile: {e}")
        raise HardwareException("Failed to retrieve hardware information", trace=str(e))

def refresh_profile(self) -> HardwareProfile:
    """Re-detect hardware and update existing profile.

    Performs fresh hardware detection and updates existing profile with
    new data. Useful for detecting hardware changes or refreshing dynamic
    fields (available_memory_gb, disk_available_gb).

    Returns:
        HardwareProfile: Updated profile entity.

    Raises:
        HardwareException: If refresh fails.

    Note:
        Does NOT commit. Caller must commit transaction.

    Example:
        >>> service = Hardware_Service(repo)
        >>> profile = service.refresh_profile()
        >>> db.commit()
    """
    try:
        logger.info("Refreshing hardware profile")

        # Detect current hardware state
        hardware_data = self._detect_hardware()

        # Get existing profile
        profile = self.repository.get_profile()
        if not profile:
            # No profile exists, create new
            logger.info("No profile to refresh, creating new")
            profile = self.repository.create_profile(hardware_data)
        else:
            # Update existing profile
            logger.info(f"Updating existing profile: id={profile.id}")
            profile = self.repository.update_profile(profile, hardware_data)

        logger.info("Hardware profile refreshed successfully")
        return profile

    except HardwareException:
        raise  # logged where it was raised
    except Exception as e:
        logger.exception(f"Failed to refresh hardware profile: {e}")
        raise HardwareException("Failed to refresh hardware profile", trace=str(e))

def warm_up(self, duration_seconds: int = 5) -> bool:
    """Warm up hardware accelerator (GPU/MPS/CUDA).

    Delegates to LLM_Engine's warm_up_accelerator() method. Useful before
    benchmarking or first inference to ensure accurate performance.

    Args:
        duration_seconds: How long to run warm-up routine.

    Returns:
        bool: True if warm-up succeeded, False otherwise.

    Raises:
        HardwareException: If warm-up fails critically.

    Example:
        >>> service = Hardware_Service(repo)
        >>> success = service.warm_up(duration_seconds=5)
        >>> if success:
        ...     print("Accelerator ready")
    """
    try:
        logger.info(f"Starting hardware warm-up: duration={duration_seconds}s")

        if not config.LLM_Engine:
            logger.warning("LLM_Engine not initialized, skipping warm-up")
            return False

        success = config.LLM_Engine.warm_up_accelerator(duration_seconds)

        if success:
            logger.info("Hardware warm-up completed successfully")
        else:
            logger.warning("Hardware warm-up failed or not available")

        return success

    except Exception as e:
        logger.exception(f"Hardware warm-up failed: {e}")
        raise HardwareException("Failed to warm up hardware accelerator", trace=str(e))