"""
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, SkillMultiMap
from knowledgespaces.structures.attribution import Attribution
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
from knowledgespaces.structures.relations import SurmiseRelation
from knowledgespaces.structures.set_family import SetFamily
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 not isinstance(data, dict):
raise ValueError(f"{context}: expected a JSON object.")
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))
def _labels(value: Any) -> list[str]:
if not isinstance(value, list) or any(not isinstance(q, str) for q in value):
raise ValueError("Expected a JSON list of string labels.")
if len(set(value)) != len(value):
raise ValueError("JSON label lists must not contain duplicates.")
return value
def _families(value: dict) -> dict[str, list[list[str]]]:
result = {}
for q, family in value.items():
if not isinstance(q, str) or not isinstance(family, list):
raise ValueError("Each family must be a list indexed by a string label.")
result[q] = [_labels(c) for c in family]
return result
[docs]
def skill_multimap_to_dict(mapping: SkillMultiMap) -> dict[str, Any]:
"""Serialize items, all skills and alternative competency families."""
return {
"items": list(mapping.items),
"skills": sorted(mapping.skills),
"mapping": {q: [sorted(c) for c in mapping.competencies_for(q)] for q in mapping.items},
}
[docs]
def dict_to_skill_multimap(data: dict[str, Any]) -> SkillMultiMap:
"""Deserialize alternative competencies, retaining item/alternative order."""
context = "SkillMultiMap JSON"
return SkillMultiMap(
_labels(_require_list(data, "items", context)),
_labels(_require_list(data, "skills", context)),
_families(_require_dict(data, "mapping", context)),
)
[docs]
def write_skill_multimap_json(mapping: SkillMultiMap, path: PathLike) -> None:
"""Write a skill multimap with labels and empty/unused skills preserved."""
Path(path).write_text(
json.dumps(skill_multimap_to_dict(mapping), indent=2, ensure_ascii=False), encoding="utf-8"
)
[docs]
def read_skill_multimap_json(path: PathLike) -> SkillMultiMap:
"""Read a labelled skill multimap JSON document."""
return dict_to_skill_multimap(json.loads(Path(path).read_text(encoding="utf-8")))
[docs]
def attribution_to_dict(attribution: Attribution) -> dict[str, Any]:
"""Serialize an attribution without replacing its clauses by canonical ones."""
return {
"domain": sorted(attribution.domain),
"clauses": {
q: [
sorted(c)
for c in sorted(attribution.clauses_for(q), key=lambda s: (len(s), sorted(s)))
]
for q in sorted(attribution.domain)
},
}
[docs]
def dict_to_attribution(data: dict[str, Any]) -> Attribution:
"""Deserialize general clauses; missing clause families are errors."""
context = "Attribution JSON"
return Attribution(
_labels(_require_list(data, "domain", context)),
_families(_require_dict(data, "clauses", context)),
)
[docs]
def write_attribution_json(attribution: Attribution, path: PathLike) -> None:
"""Write general clauses and their domain to JSON."""
Path(path).write_text(
json.dumps(attribution_to_dict(attribution), indent=2, ensure_ascii=False), encoding="utf-8"
)
[docs]
def read_attribution_json(path: PathLike) -> Attribution:
"""Read general clauses from JSON, without canonicalizing them."""
return dict_to_attribution(json.loads(Path(path).read_text(encoding="utf-8")))
[docs]
def family_to_dict(family: SetFamily) -> dict[str, Any]:
"""Serialize exactly the given family, without inserting endpoints."""
return {"domain": sorted(family.domain), "sets": [sorted(s) for s in family]}
[docs]
def dict_to_family(data: dict[str, Any]) -> SetFamily:
"""Deserialize a family that need not be a knowledge structure."""
context = "SetFamily JSON"
return SetFamily(
_labels(_require_list(data, "domain", context)),
[_labels(s) for s in _require_list(data, "sets", context)],
)
[docs]
def write_family_json(family: SetFamily, path: PathLike) -> None:
"""Write an arbitrary family; bases can be exported via ``as_family()``."""
Path(path).write_text(
json.dumps(family_to_dict(family), indent=2, ensure_ascii=False), encoding="utf-8"
)
[docs]
def read_family_json(path: PathLike) -> SetFamily:
"""Read a family; an explicit ``to_knowledge_base()`` interprets generators."""
return dict_to_family(json.loads(Path(path).read_text(encoding="utf-8")))