generated from coulomb/repo-seed
306 lines
9.7 KiB
Python
306 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScoreEntry:
|
|
name: str
|
|
value: float
|
|
max_value: float = 5.0
|
|
rationale: str = ""
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data: dict[str, Any] = {
|
|
"name": self.name,
|
|
"value": self.value,
|
|
"max_value": self.max_value,
|
|
}
|
|
if self.rationale:
|
|
data["rationale"] = self.rationale
|
|
return data
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ScoreEntry":
|
|
return cls(
|
|
name=str(data["name"]),
|
|
value=float(data["value"]),
|
|
max_value=float(data.get("max_value", 5.0)),
|
|
rationale=str(data.get("rationale") or ""),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EntityEvaluation:
|
|
artifact_id: str
|
|
evaluator: str
|
|
scores: list[ScoreEntry]
|
|
evaluated_at: datetime
|
|
notes: list[str] = field(default_factory=list)
|
|
|
|
@property
|
|
def entity_slug(self) -> str:
|
|
"""Legacy alias for readers moving from entity-oriented history files."""
|
|
return self.artifact_id
|
|
|
|
@property
|
|
def overall_score(self) -> float:
|
|
if not self.scores:
|
|
return 0.0
|
|
return sum(score.value for score in self.scores) / len(self.scores)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"artifact_id": self.artifact_id,
|
|
"evaluator": self.evaluator,
|
|
"evaluated_at": self.evaluated_at.isoformat(),
|
|
"overall_score": round(self.overall_score, 4),
|
|
"scores": [score.to_dict() for score in self.scores],
|
|
"notes": self.notes,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "EntityEvaluation":
|
|
return cls(
|
|
artifact_id=str(data["artifact_id"]),
|
|
evaluator=str(data["evaluator"]),
|
|
scores=[ScoreEntry.from_dict(item) for item in data.get("scores", [])],
|
|
evaluated_at=datetime.fromisoformat(str(data["evaluated_at"])),
|
|
notes=list(data.get("notes") or []),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MetricValue:
|
|
name: str
|
|
value: float
|
|
concern: str = ""
|
|
details: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data: dict[str, Any] = {"name": self.name, "value": self.value}
|
|
if self.concern:
|
|
data["concern"] = self.concern
|
|
if self.details:
|
|
data["details"] = self.details
|
|
return data
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "MetricValue":
|
|
return cls(
|
|
name=str(data["name"]),
|
|
value=float(data["value"]),
|
|
concern=str(data.get("concern") or ""),
|
|
details=dict(data.get("details") or {}),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EvaluationSnapshot:
|
|
snapshot_id: str
|
|
created_at: datetime
|
|
schema_name: str
|
|
artifact_count: int
|
|
artifact_evaluations: list[EntityEvaluation] = field(default_factory=list)
|
|
collection_metrics: list[MetricValue] = field(default_factory=list)
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
@property
|
|
def entity_count(self) -> int:
|
|
"""Legacy alias retained for old infospace history readers."""
|
|
return self.artifact_count
|
|
|
|
@property
|
|
def entity_evaluations(self) -> list[EntityEvaluation]:
|
|
"""Legacy alias retained for old infospace history readers."""
|
|
return self.artifact_evaluations
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"snapshot_id": self.snapshot_id,
|
|
"created_at": self.created_at.isoformat(),
|
|
"schema_name": self.schema_name,
|
|
"artifact_count": self.artifact_count,
|
|
"artifact_evaluations": [
|
|
evaluation.to_dict() for evaluation in self.artifact_evaluations
|
|
],
|
|
"collection_metrics": [
|
|
metric.to_dict() for metric in self.collection_metrics
|
|
],
|
|
"metadata": self.metadata,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "EvaluationSnapshot":
|
|
return cls(
|
|
snapshot_id=str(data["snapshot_id"]),
|
|
created_at=datetime.fromisoformat(str(data["created_at"])),
|
|
schema_name=str(data.get("schema_name") or "default"),
|
|
artifact_count=int(data.get("artifact_count", data.get("entity_count", 0))),
|
|
artifact_evaluations=[
|
|
EntityEvaluation.from_dict(item)
|
|
for item in data.get(
|
|
"artifact_evaluations",
|
|
data.get("entity_evaluations", []),
|
|
)
|
|
],
|
|
collection_metrics=[
|
|
MetricValue.from_dict(item) for item in data.get("collection_metrics", [])
|
|
],
|
|
metadata=dict(data.get("metadata") or {}),
|
|
)
|
|
|
|
def diff(self, after: "EvaluationSnapshot") -> "SnapshotDiff":
|
|
return diff_snapshots(self, after)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScoreChange:
|
|
artifact_id: str
|
|
dimension: str
|
|
before: float
|
|
after: float
|
|
|
|
@property
|
|
def delta(self) -> float:
|
|
return self.after - self.before
|
|
|
|
@property
|
|
def entity_slug(self) -> str:
|
|
"""Legacy alias for old diff consumers."""
|
|
return self.artifact_id
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"artifact_id": self.artifact_id,
|
|
"dimension": self.dimension,
|
|
"before": self.before,
|
|
"after": self.after,
|
|
"delta": self.delta,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MetricChange:
|
|
name: str
|
|
before: float
|
|
after: float
|
|
|
|
@property
|
|
def delta(self) -> float:
|
|
return self.after - self.before
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"before": self.before,
|
|
"after": self.after,
|
|
"delta": self.delta,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SnapshotDiff:
|
|
before_id: str
|
|
after_id: str
|
|
added_artifacts: list[str] = field(default_factory=list)
|
|
removed_artifacts: list[str] = field(default_factory=list)
|
|
score_changes: list[ScoreChange] = field(default_factory=list)
|
|
metric_changes: list[MetricChange] = field(default_factory=list)
|
|
|
|
@property
|
|
def added_entities(self) -> list[str]:
|
|
"""Legacy alias for old history diff output."""
|
|
return self.added_artifacts
|
|
|
|
@property
|
|
def removed_entities(self) -> list[str]:
|
|
"""Legacy alias for old history diff output."""
|
|
return self.removed_artifacts
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"before_id": self.before_id,
|
|
"after_id": self.after_id,
|
|
"added_artifacts": self.added_artifacts,
|
|
"removed_artifacts": self.removed_artifacts,
|
|
"score_changes": [change.to_dict() for change in self.score_changes],
|
|
"metric_changes": [change.to_dict() for change in self.metric_changes],
|
|
}
|
|
|
|
def summary(self) -> str:
|
|
lines = [f"Snapshot diff: {self.before_id} -> {self.after_id}"]
|
|
if not (
|
|
self.added_artifacts
|
|
or self.removed_artifacts
|
|
or self.score_changes
|
|
or self.metric_changes
|
|
):
|
|
return "\n".join([*lines, "No changes."])
|
|
for artifact_id in self.added_artifacts:
|
|
lines.append(f"Added artifact: {artifact_id}")
|
|
for artifact_id in self.removed_artifacts:
|
|
lines.append(f"Removed artifact: {artifact_id}")
|
|
for change in self.score_changes:
|
|
lines.append(
|
|
f"Score {change.artifact_id} {change.dimension}: "
|
|
f"{change.before} -> {change.after} ({change.delta:+.4f})"
|
|
)
|
|
for change in self.metric_changes:
|
|
lines.append(
|
|
f"Metric {change.name}: "
|
|
f"{change.before} -> {change.after} ({change.delta:+.4f})"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def diff_snapshots(
|
|
before: EvaluationSnapshot,
|
|
after: EvaluationSnapshot,
|
|
) -> SnapshotDiff:
|
|
before_scores = _score_index(before)
|
|
after_scores = _score_index(after)
|
|
before_artifacts = {
|
|
evaluation.artifact_id for evaluation in before.artifact_evaluations
|
|
}
|
|
after_artifacts = {evaluation.artifact_id for evaluation in after.artifact_evaluations}
|
|
|
|
score_changes = [
|
|
ScoreChange(
|
|
artifact_id,
|
|
dimension,
|
|
before_scores.get(key, 0.0),
|
|
after_scores.get(key, 0.0),
|
|
)
|
|
for key in sorted(before_scores.keys() | after_scores.keys())
|
|
for artifact_id, dimension in [key]
|
|
if before_scores.get(key) != after_scores.get(key)
|
|
]
|
|
|
|
before_metrics = {metric.name: metric.value for metric in before.collection_metrics}
|
|
after_metrics = {metric.name: metric.value for metric in after.collection_metrics}
|
|
metric_changes = [
|
|
MetricChange(name, before_metrics.get(name, 0.0), after_metrics.get(name, 0.0))
|
|
for name in sorted(before_metrics.keys() | after_metrics.keys())
|
|
if before_metrics.get(name) != after_metrics.get(name)
|
|
]
|
|
|
|
return SnapshotDiff(
|
|
before_id=before.snapshot_id,
|
|
after_id=after.snapshot_id,
|
|
added_artifacts=sorted(after_artifacts - before_artifacts),
|
|
removed_artifacts=sorted(before_artifacts - after_artifacts),
|
|
score_changes=score_changes,
|
|
metric_changes=metric_changes,
|
|
)
|
|
|
|
|
|
def _score_index(snapshot: EvaluationSnapshot) -> dict[tuple[str, str], float]:
|
|
return {
|
|
(evaluation.artifact_id, score.name): score.value
|
|
for evaluation in snapshot.artifact_evaluations
|
|
for score in evaluation.scores
|
|
}
|