"""
JSON serialization for KST objects.
Provides roundtrip-safe serialization for the core KST types:
KnowledgeStructure, SurmiseRelation, and SkillMap.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Union
from knowledgespaces.derivation.skill_map import SkillMap
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
from knowledgespaces.structures.relations import SurmiseRelation
PathLike = Union[str, Path]
def _require_key(data: dict, key: str, context: str) -> Any:
"""Get a required key from a dict, raising ValueError on missing."""
if key not in data:
raise ValueError(f"{context}: missing required key '{key}'.")
return data[key]
def _require_list(data: dict, key: str, context: str) -> list:
"""Get a required key that must be a list."""
val = _require_key(data, key, context)
if not isinstance(val, list):
raise ValueError(f"{context}: '{key}' must be a list, got {type(val).__name__}.")
return val
def _require_dict(data: dict, key: str, context: str) -> dict:
"""Get a required key that must be a dict."""
val = _require_key(data, key, context)
if not isinstance(val, dict):
raise ValueError(f"{context}: '{key}' must be a dict, got {type(val).__name__}.")
return val
# ------------------------------------------------------------------
# KnowledgeStructure
# ------------------------------------------------------------------
[docs]
def structure_to_dict(structure: KnowledgeStructure) -> dict[str, Any]:
"""Serialize a KnowledgeStructure to a JSON-compatible dict."""
return {
"domain": sorted(structure.domain),
"states": [sorted(s) for s in sorted(structure.states, key=lambda s: (len(s), sorted(s)))],
"properties": {
"n_states": structure.n_states,
"is_knowledge_space": structure.is_knowledge_space,
"is_closure_space": structure.is_closure_space,
"is_well_graded": structure.is_well_graded,
"is_learning_space": structure.is_learning_space,
},
}
[docs]
def dict_to_structure(data: dict[str, Any]) -> KnowledgeStructure:
"""Deserialize a KnowledgeStructure from a dict.
Raises
------
ValueError
If required keys are missing or have wrong types.
"""
ctx = "KnowledgeStructure JSON"
domain = _require_list(data, "domain", ctx)
states_raw = _require_list(data, "states", ctx)
states = []
for i, s in enumerate(states_raw):
if not isinstance(s, list):
raise ValueError(f"{ctx}: states[{i}] must be a list, got {type(s).__name__}.")
states.append(frozenset(s))
return KnowledgeStructure(domain=domain, states=states)
[docs]
def write_structure_json(structure: KnowledgeStructure, path: PathLike) -> None:
"""Write a KnowledgeStructure to a JSON file."""
with open(path, "w", encoding="utf-8") as f:
json.dump(structure_to_dict(structure), f, indent=2, ensure_ascii=False)
[docs]
def read_structure_json(path: PathLike) -> KnowledgeStructure:
"""Read a KnowledgeStructure from a JSON file."""
with open(path, encoding="utf-8") as f:
return dict_to_structure(json.load(f))
# ------------------------------------------------------------------
# SurmiseRelation
# ------------------------------------------------------------------
[docs]
def relation_to_dict(relation: SurmiseRelation) -> dict[str, Any]:
"""Serialize a SurmiseRelation to a JSON-compatible dict."""
return {
"items": sorted(relation.items),
"relations": sorted([list(pair) for pair in relation.relations]),
}
[docs]
def dict_to_relation(data: dict[str, Any]) -> SurmiseRelation:
"""Deserialize a SurmiseRelation from a dict.
Raises
------
ValueError
If required keys are missing or have wrong types.
"""
ctx = "SurmiseRelation JSON"
items = _require_list(data, "items", ctx)
relations_raw = _require_list(data, "relations", ctx)
relations = []
for i, pair in enumerate(relations_raw):
if not isinstance(pair, list) or len(pair) != 2:
raise ValueError(f"{ctx}: relations[{i}] must be a [a, b] pair.")
relations.append((pair[0], pair[1]))
return SurmiseRelation(items=items, relations=relations)
[docs]
def write_relation_json(relation: SurmiseRelation, path: PathLike) -> None:
"""Write a SurmiseRelation to a JSON file."""
with open(path, "w", encoding="utf-8") as f:
json.dump(relation_to_dict(relation), f, indent=2, ensure_ascii=False)
[docs]
def read_relation_json(path: PathLike) -> SurmiseRelation:
"""Read a SurmiseRelation from a JSON file."""
with open(path, encoding="utf-8") as f:
return dict_to_relation(json.load(f))
# ------------------------------------------------------------------
# SkillMap
# ------------------------------------------------------------------
[docs]
def skill_map_to_dict(skill_map: SkillMap) -> dict[str, Any]:
"""Serialize a SkillMap to a JSON-compatible dict."""
return {
"items": list(skill_map.items),
"skills": sorted(skill_map.skills),
"mapping": {item: sorted(skill_map.skills_for(item)) for item in skill_map.items},
}
[docs]
def dict_to_skill_map(data: dict[str, Any]) -> SkillMap:
"""Deserialize a SkillMap from a dict.
Raises
------
ValueError
If required keys are missing or have wrong types.
"""
ctx = "SkillMap JSON"
items = _require_list(data, "items", ctx)
skills = _require_list(data, "skills", ctx)
mapping_raw = _require_dict(data, "mapping", ctx)
mapping = {item: frozenset(v) for item, v in mapping_raw.items()}
return SkillMap(items=items, skills=skills, mapping=mapping)
[docs]
def write_skill_map_json(skill_map: SkillMap, path: PathLike) -> None:
"""Write a SkillMap to a JSON file."""
with open(path, "w", encoding="utf-8") as f:
json.dump(skill_map_to_dict(skill_map), f, indent=2, ensure_ascii=False)
[docs]
def read_skill_map_json(path: PathLike) -> SkillMap:
"""Read a SkillMap from a JSON file."""
with open(path, encoding="utf-8") as f:
return dict_to_skill_map(json.load(f))