"""
JSON serialization for KST objects.
Provides roundtrip-safe serialization for the core KST types:
KnowledgeStructure, SurmiseRelation, SkillMap, and SurmiseFunction.
"""
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
from knowledgespaces.structures.surmise_function import SurmiseFunction
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))
# ------------------------------------------------------------------
# SurmiseFunction
# ------------------------------------------------------------------
[docs]
def surmise_function_to_dict(sf: SurmiseFunction) -> dict[str, Any]:
"""Serialize a SurmiseFunction to a JSON-compatible dict.
Format::
{
"domain": ["a", "b", ...],
"clauses": {
"a": [["a"]],
"b": [["b", "d"], ["a", "b", "c"]],
...
},
"properties": {
"n_items": 5,
"is_ordinal": false,
"is_discriminative": true,
"is_acyclic": false
}
}
"""
domain = sorted(sf.domain)
clauses: dict[str, list[list[str]]] = {}
for q in domain:
clauses[q] = [
sorted(c) for c in sorted(sf.clauses_for(q), key=lambda c: (len(c), sorted(c)))
]
return {
"domain": domain,
"clauses": clauses,
"properties": {
"n_items": sf.n_items,
"is_ordinal": sf.is_ordinal,
"is_discriminative": sf.is_discriminative,
"is_acyclic": sf.is_acyclic,
},
}
[docs]
def dict_to_surmise_function(data: dict[str, Any]) -> SurmiseFunction:
"""Deserialize a SurmiseFunction from a dict.
Raises
------
ValueError
If required keys are missing or have wrong types.
"""
ctx = "SurmiseFunction JSON"
domain = _require_list(data, "domain", ctx)
clauses_raw = _require_dict(data, "clauses", ctx)
clauses: dict[str, list[frozenset[str]]] = {}
for item, family in clauses_raw.items():
if not isinstance(family, list):
raise ValueError(
f"{ctx}: clauses[{item!r}] must be a list, got {type(family).__name__}."
)
parsed: list[frozenset[str]] = []
for i, c in enumerate(family):
if not isinstance(c, list):
raise ValueError(
f"{ctx}: clauses[{item!r}][{i}] must be a list, got {type(c).__name__}."
)
parsed.append(frozenset(c))
clauses[item] = parsed
return SurmiseFunction(domain=domain, clauses=clauses)
[docs]
def write_surmise_function_json(sf: SurmiseFunction, path: PathLike) -> None:
"""Write a SurmiseFunction to a JSON file."""
with open(path, "w", encoding="utf-8") as f:
json.dump(surmise_function_to_dict(sf), f, indent=2, ensure_ascii=False)
[docs]
def read_surmise_function_json(path: PathLike) -> SurmiseFunction:
"""Read a SurmiseFunction from a JSON file."""
with open(path, encoding="utf-8") as f:
return dict_to_surmise_function(json.load(f))