Spaces:
Running
Running
| """Standard-library helpers for immutable droid-mind training data.""" | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| from typing import TypedDict | |
| class Example(TypedDict): | |
| user: str | |
| emotion: str | |
| reply: str | |
| def sha256_file(path: Path) -> str: | |
| digest = hashlib.sha256() | |
| with path.open("rb") as source: | |
| for chunk in iter(lambda: source.read(1024 * 1024), b""): | |
| digest.update(chunk) | |
| return digest.hexdigest() | |
| def load_emotion_ids(path: Path) -> list[str]: | |
| payload = json.loads(path.read_text()) | |
| if not isinstance(payload, list): | |
| raise ValueError(f"{path}: expected a JSON list") | |
| ids = [item.get("id") for item in payload if isinstance(item, dict)] | |
| if not ids or any(not isinstance(emotion, str) or not emotion for emotion in ids): | |
| raise ValueError(f"{path}: every emotion must have a non-empty id") | |
| if len(ids) != len(set(ids)): | |
| raise ValueError(f"{path}: emotion ids must be unique") | |
| return ids | |
| def load_examples( | |
| path: Path, | |
| emotion_ids: set[str], | |
| max_rows: int | None = None, | |
| ) -> list[Example]: | |
| if max_rows is not None and max_rows < 1: | |
| raise ValueError("max_rows must be positive") | |
| examples: list[Example] = [] | |
| with path.open() as source: | |
| for line_number, line in enumerate(source, 1): | |
| if not line.strip(): | |
| continue | |
| try: | |
| row = json.loads(line) | |
| except json.JSONDecodeError as error: | |
| raise ValueError(f"{path}:{line_number}: invalid JSON: {error}") from error | |
| if not isinstance(row, dict): | |
| raise ValueError(f"{path}:{line_number}: expected a JSON object") | |
| values: dict[str, str] = {} | |
| for field in ("user", "emotion", "reply"): | |
| value = row.get(field) | |
| if not isinstance(value, str) or not value.strip(): | |
| raise ValueError( | |
| f"{path}:{line_number}: {field} must be a non-empty string" | |
| ) | |
| values[field] = value | |
| if values["emotion"] not in emotion_ids: | |
| raise ValueError( | |
| f"{path}:{line_number}: unknown emotion {values['emotion']!r}" | |
| ) | |
| examples.append( | |
| { | |
| "user": values["user"], | |
| "emotion": values["emotion"], | |
| "reply": values["reply"], | |
| } | |
| ) | |
| if max_rows is not None and len(examples) == max_rows: | |
| break | |
| if not examples: | |
| raise ValueError(f"{path}: no examples") | |
| return examples | |