Results

def __post_init__(self):
    """Initialize the Result instance."""
    self._contributed_stats = {}
    if not hasattr(self, "_preloaded_stats"):
        self._preloaded_stats = None

def get_dimension(
    self,
    dimension: str,
    filter_dimension: str | None = None,
    filter_value: Any = None,
) -> list:
    """
    Get the values of a specific dimension from the responses.

    Args:
        dimension (str): The name of the dimension to retrieve.
        filter_dimension (str, optional): Name of dimension to filter on. Defaults to None.
        filter_value (any, optional): Value to match for the filter dimension. Defaults to None.

    Returns:
        list: A list of values for the specified dimension across all responses.

    Raises:
        ValueError: If the specified dimension is not found in any response.
    """
    if filter_dimension is not None:
        values = [
            getattr(response, dimension)
            for response in self.responses
            if getattr(response, filter_dimension) == filter_value
        ]
    else:
        values = [getattr(response, dimension) for response in self.responses]

    if not any(values):
        # raise ValueError(f"Dimension {dimension} not found in any response")
        logger.warning(f"Dimension {dimension} not found in any response")
    return values

@classmethod
def load(
    cls, result_path: ReadablePathLike, load_responses: bool = True
) -> "Result":
    """
    Load run results from disk or cloud storage.

    Reads previously saved run results from the specified path. Handles
    both complete runs (with ``summary.json``) and interrupted runs where
    only ``responses.jsonl``, ``run_config.json``, or ``stats.json`` are
    available.

    Args:
        result_path (UPath | str): The path to the directory containing the
            result files. Can be a string or a UPath object.
        load_responses (bool): Whether to load individual invocation responses
            from 'responses.jsonl'. Defaults to True. When False, only the
            summary and pre-computed stats are loaded, which is significantly
            faster for large result sets. Use ``result.load_responses()`` to
            load them on demand later.

    Returns:
        Result: An instance of the Result class containing the loaded
        responses and summary data.

    Raises:
        FileNotFoundError: If the directory does not exist or contains no
            recognizable result files (``summary.json``, ``responses.jsonl``,
            or ``stats.json``).
        JSONDecodeError: If ``summary.json`` cannot be parsed.
    """
    result_path = ensure_path(result_path)
    summary_path = result_path / "summary.json"

    # 1. Resolve metadata
    if summary_path.exists():
        metadata = cls._load_summary(result_path)
    else:
        metadata = cls._recover_metadata(result_path)

    # 2. Load responses (if requested and available)
    responses = cls._read_responses(result_path) if load_responses else []
    if not load_responses:
        logger.info(
            "Loaded summary only (responses not loaded). "
            "Call result.load_responses() to load them on demand.",
        )

    # 3. Construct and resolve stats
    result = cls(responses=responses, **metadata)
    cls._resolve_stats(result, result_path / "stats.json")
    return result

def load_responses(self) -> list[InvocationResponse]:
    """
    Load individual invocation responses from disk or cloud storage.

    Reads the 'responses.jsonl' file from the result's output_path directory.
    This is useful when the Result was loaded with ``load_responses=False`` and
    you need to access the individual responses on demand.

    Returns:
        list[InvocationResponse]: The loaded responses. Also updates ``self.responses``
        in place.

    Raises:
        ValueError: If no output_path is set on this Result.
        FileNotFoundError: If 'responses.jsonl' is not found at the output_path.
    """
    if not self.output_path:
        raise ValueError(
            "No output_path set on this Result. Cannot locate responses file."
        )
    responses_path = ensure_path(self.output_path) / "responses.jsonl"
    with responses_path.open("r") as f:
        self.responses = [InvocationResponse.from_json(line) for line in f if line]
    logger.info("Loaded %d responses from %s", len(self.responses), responses_path)
    # Recompute stats from the freshly loaded responses
    self._preloaded_stats = self._compute_stats(self)
    return self.responses

def save(self, output_path: WritablePathLike | None = None) -> None:
    """
    Save the results to disk or cloud storage.

    Saves the run results to the specified output path or the
    instance's default output path. It creates three files:
    1. 'summary.json': Contains the overall summary of the results.
    2. 'stats.json': Contains detailed statistics of the run.
    3. 'responses.jsonl': Contains individual invocation responses
        - Only if the responses are not already saved at the indicated path.


    Args:
        output_path (UPath | str | None, optional): The path where the result
            files will be saved. If None, the instance's default output_path
            will be used. Defaults to None.

    Raises:
        ValueError: If no output path is provided and the instance doesn't
            have a default output_path set.
        TypeError: If the provided output_path is not a valid type.
        IOError: If there's an error writing to the output files.

    Note:
        The method uses the Universal Path (UPath) library for file operations,
        which provides a unified interface for working with different file systems.
    """

    output_path = ensure_path(self.output_path or output_path)
    if output_path is None:
        raise ValueError("No output path provided")

    output_path.mkdir(parents=True, exist_ok=True)

    summary_path = output_path / "summary.json"
    stats_path = output_path / "stats.json"
    with summary_path.open("w") as f, stats_path.open("w") as s:
        f.write(self.to_json(indent=4))
        s.write(
            json.dumps(self.stats, indent=4, default=llmeter_default_serializer)
        )

    responses_path = output_path / "responses.jsonl"
    if not responses_path.exists():
        with responses_path.open("w") as f:
            for response in self.responses:
                f.write(
                    json.dumps(asdict(response), default=llmeter_default_serializer)
                    + "\n"
                )

def to_dict(self, include_responses: bool = False) -> dict:
    """Return a dictionary representation of this result.

    Returns a plain ``dict`` produced by :func:`dataclasses.asdict`,
    preserving native Python types (``datetime``, ``UPath``, etc.).
    This is suitable for programmatic access and internal data
    processing.

    For JSON output, use :meth:`to_json` which delegates to
    :func:`~llmeter.json_utils.llmeter_default_serializer` for
    non-serializable types, or pass the dict through
    ``json.dumps(result.to_dict(), default=llmeter_default_serializer)``.

    Args:
        include_responses: If ``True``, include the full list of
            :class:`~llmeter.endpoints.base.InvocationResponse` dicts
            and the ``stats`` key.  Defaults to ``False``.

    Returns:
        dict: A dictionary of result fields with native Python types.
    """
    data = asdict(self)
    if include_responses:
        return data
    return {k: v for k, v in data.items() if k not in ["responses", "stats"]}

def to_json(self, default=llmeter_default_serializer, **kwargs) -> str:
    """Return the results as a JSON string.

    Args:
        default: Fallback serializer. Defaults to
            :func:`~llmeter.json_utils.llmeter_default_serializer`.
        **kwargs: Extra keyword arguments passed to :func:`json.dumps`.
    """
    summary = {
        k: o for k, o in asdict(self).items() if k not in ["responses", "stats"]
    }
    return json.dumps(summary, default=default, **kwargs)