Skip to content

rdfs_tools: derive class attributes from schema:domainIncludes as well as rdfs:domain #81

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Haigutus merged 2 commits into from
Jul 18, 2026
Merged

rdfs_tools: derive class attributes from schema:domainIncludes as well as rdfs:domain #81

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
46cf99d
feat(rdfs): derive class attributes from schema:domainIncludes as wel…
Jul 18, 2026
d728d5f
feat(rdfs): emit attributes with no class binding as top-level schema…
Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 61 additions & 0 deletions tests/test_rdfs_tools.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,34 @@ def test_returns_data(self, rdfs_profile):
params = rdfs_tools.get_class_parameters(rdfs_profile, classes[0])
assert params is not None

def test_domain_and_domainincludes_both_bind_attributes(self, tmp_path):
"""Attribute→class binding is read from rdfs:domain (CIM-owned terms) and
schema:domainIncludes (reused external terms) alike — the DatasetMetadata
convention (see application-profiles-library#92)."""
rdfs = tmp_path / "mini.rdf"
rdfs.write_text("""<?xml version="1.0" encoding="UTF-8"?>
<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"
xmlns:cims="http://iec.ch/TC57/1999/rdf-schema-extensions-19990926#"
xmlns:schema="https://schema.org/">
<rdf:Description rdf:about="http://www.w3.org/ns/dcat#Dataset">
<rdf:type rdf:resource="http://www.w3.org/2000/01/rdf-schema#Class"/>
</rdf:Description>
<rdf:Description rdf:about="https://cim4.eu/ns/Metadata-European#usedSettings">
<rdf:type rdf:resource="http://www.w3.org/1999/02/22-rdf-syntax-ns#Property"/>
<rdfs:domain rdf:resource="http://www.w3.org/ns/dcat#Dataset"/>
</rdf:Description>
<rdf:Description rdf:about="http://purl.org/dc/terms/accessRights">
<rdf:type rdf:resource="http://www.w3.org/1999/02/22-rdf-syntax-ns#Property"/>
<schema:domainIncludes rdf:resource="http://www.w3.org/ns/dcat#Dataset"/>
</rdf:Description>
</rdf:RDF>
""", encoding="utf-8")
data = rdfs_tools.load_all_to_dataframe(str(rdfs))
found = set(rdfs_tools.get_class_parameters(data, "http://www.w3.org/ns/dcat#Dataset")["parameters"]["ID"])
assert "https://cim4.eu/ns/Metadata-European#usedSettings" in found # via rdfs:domain
assert "http://purl.org/dc/terms/accessRights" in found # via schema:domainIncludes


class TestParametersTableview:
def test_returns_tuple(self, rdfs_profile):
Expand Down Expand Up @@ -114,3 +142,36 @@ def test_convert_profile(self, rdfs_profile):
result = cim_rdfs_to_json.convert_profile(rdfs_profile)
assert isinstance(result, dict)
assert len(result) > 0


class TestOrphanedAttributes:
"""An attribute with no class binding (no rdfs:domain / schema:domainIncludes)
is still emitted as a top-level schema entry — just not listed under any class —
with a warning. A consistent profile produces no orphan warning."""
DM = Path("rdfs/ENTSOE_NC_2.4.1/DatasetMetadata-AP-Voc-RDFS2020.rdf")
LOG = "triplets.rdfs_tools.cim_rdfs_to_json"

def _data(self):
if not self.DM.exists():
pytest.skip("NC 2.4.1 DatasetMetadata RDFS not available")
return rdfs_tools.load_all_to_dataframe(str(self.DM))

def test_consistent_profile_has_no_orphans(self, caplog):
from triplets.rdfs_tools import cim_rdfs_to_json
with caplog.at_level("WARNING", logger=self.LOG):
profile = cim_rdfs_to_json.convert_profile(self._data())
assert "title" in profile["Dataset"]["parameters"] # bound attribute, listed
assert not any("no class binding" in r.getMessage() for r in caplog.records)

def test_orphaned_attribute_emitted_without_class(self, caplog):
from triplets.rdfs_tools import cim_rdfs_to_json
data = self._data()
title = "http://purl.org/dc/terms/title"
# strip title's class binding → orphan it
orphaned = data[~((data.ID == title) & (data.KEY.isin(["domain", "domainIncludes"])))].copy()
with caplog.at_level("WARNING", logger=self.LOG):
profile = cim_rdfs_to_json.convert_profile(orphaned)
assert "title" in profile # definition still emitted
assert profile["title"].get("dataType") == "String" # with its datatype preserved
assert "title" not in profile["Dataset"]["parameters"] # but no class references it
assert any("no class binding" in r.getMessage() for r in caplog.records)
224 changes: 124 additions & 100 deletions triplets/rdfs_tools/cim_rdfs_to_json.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,108 @@ def convert_profile(profile_data, serialization_version="552_ED2"):
# profile and referenced by about (e.g. NC associations attached to EQ objects)
export_classes = list(dict.fromkeys(rdfs_tools.concrete_classes_list(profile_data) + classes_defined_externally))

def add_parameter(parameter, parameter_meta):
"""Emit one attribute's definition as a top-level profile entry (datatype,
multiplicity, enum values, ...). Returns its name, or None for an unused
association (skipped). Does not attach it to a class — the caller does that."""

# Pivot yields NaN for keys a parameter does not have, semantics expect them absent
parameter_dict = {key: value for key, value in parameter_meta.to_dict().items() if pandas.notna(value)}

# TODO - export this and add it to Association metadata
association_used = parameter_dict.get("AssociationUsed")

# If it is association but not used, we don't export it
if association_used == 'No':
return None

parameter_namespace, parameter_name = rdfs_tools.get_namespace_and_name(parameter, default_namespace=xml_base)

parameter_def = {
"description": parameter_dict.get("comment", ""),
"multiplicity": parameter_dict["multiplicity"].split("M:")[1],
"namespace": parameter_namespace
}

parameter_def["xsd:minOccours"], parameter_def["xsd:maxOccours"] = rdfs_tools.parse_multiplicity(parameter_dict["multiplicity"])

# If association
if association_used == 'Yes':
parameter_def["attrib"] = {
"attribute": resource_attribute,
"value_prefix": resource_prefix
}

parameter_def["type"] = "Association"
parameter_def["xsd:type"] = "xsd:anyURI"
parameter_def["range"] = parameter_dict["range"]

else:
data_type = parameter_dict.get("dataType")

# If regular attribute, find its data type and add to export
if data_type:

# Set parameter type to Attribute
parameter_def["type"] = "Attribute"

# Get the attribute data type and add to export
data_type_namespace, data_type_name = rdfs_tools.get_namespace_and_name(data_type, default_namespace=xml_base)

data_type_meta = profile_data.get_object_data(data_type).to_dict()

if data_type_namespace == "":
data_type_namespace = xml_base

data_type_def = {
"description": data_type_meta.get("comment", ""),
"type": data_type_meta.get("stereotype", ""),
"xsd:type": cgmes_data_types_map.get(data_type_name, ""),
"namespace": data_type_namespace
}

# Add data type to export
profile[data_type_name] = data_type_def

# Add data type to attribute definition
parameter_def["dataType"] = data_type_name
parameter_def["xsd:type"] = data_type_def["xsd:type"]

# If enumeration
else:
parameter_def["attrib"] = {
"attribute": enumeration_attribute,
"value_prefix": enumeration_prefix # TODO - prefix should be used per value
}
parameter_def["type"] = "Enumeration"
parameter_def["xsd:type"] = "xsd:anyURI"
parameter_def["range"] = parameter_dict["range"].replace("#", "")
parameter_def["values"] = []

# Add allowed values
values = profile_data.query(f"VALUE == '{parameter_dict['range']}' and KEY == 'type'").ID.tolist()

for value in values:

value_namespace, value_name = rdfs_tools.get_namespace_and_name(value, default_namespace=xml_base)
value_meta = profile_data.get_object_data(value).to_dict()

if value_namespace == "":
value_namespace = xml_base

value_def = {
"description": value_meta.get("comment", ""),
"namespace": value_namespace,
"type": "EnumerationValue"
}

parameter_def["values"].append(value_name)
profile[value_name] = value_def

# Add parameter definition
profile[parameter_name] = parameter_def
return parameter_name

for concrete_class in export_classes:

# Define class namespace
Expand Down Expand Up @@ -143,107 +245,29 @@ def convert_profile(profile_data, serialization_version="552_ED2"):
}

# Add attributes

for parameter, parameter_meta in class_parameters_table.iterrows():

# Pivot yields NaN for keys a parameter does not have, semantics expect them absent
parameter_dict = {key: value for key, value in parameter_meta.to_dict().items() if pandas.notna(value)}

# TODO - export this and add it to Association metadata
association_used = parameter_dict.get("AssociationUsed")

# If it is association but not used, we don't export it
if association_used == 'No':
continue

parameter_namespace, parameter_name = rdfs_tools.get_namespace_and_name(parameter, default_namespace=xml_base)

parameter_def = {
"description": parameter_dict.get("comment", ""),
"multiplicity": parameter_dict["multiplicity"].split("M:")[1],
"namespace": parameter_namespace
}

parameter_def["xsd:minOccours"], parameter_def["xsd:maxOccours"] = rdfs_tools.parse_multiplicity(parameter_dict["multiplicity"])

# If association
if association_used == 'Yes':
parameter_def["attrib"] = {
"attribute": resource_attribute,
"value_prefix": resource_prefix
}

parameter_def["type"] = "Association"
parameter_def["xsd:type"] = "xsd:anyURI"
parameter_def["range"] = parameter_dict["range"]

else:
data_type = parameter_dict.get("dataType")

# If regular attribute, find its data type and add to export
if data_type:

# Set parameter type to Attribute
parameter_def["type"] = "Attribute"

# Get the attribute data type and add to export
data_type_namespace, data_type_name = rdfs_tools.get_namespace_and_name(data_type, default_namespace=xml_base)

data_type_meta = profile_data.get_object_data(data_type).to_dict()

if data_type_namespace == "":
data_type_namespace = xml_base

data_type_def = {
"description": data_type_meta.get("comment", ""),
"type": data_type_meta.get("stereotype", ""),
"xsd:type": cgmes_data_types_map.get(data_type_name, ""),
"namespace": data_type_namespace
}

# Add data type to export
profile[data_type_name] = data_type_def

# Add data type to attribute definition
parameter_def["dataType"] = data_type_name
parameter_def["xsd:type"] = data_type_def["xsd:type"]

# If enumeration
else:
parameter_def["attrib"] = {
"attribute": enumeration_attribute,
"value_prefix": enumeration_prefix # TODO - prefix should be used per value
}
parameter_def["type"] = "Enumeration"
parameter_def["xsd:type"] = "xsd:anyURI"
parameter_def["range"] = parameter_dict["range"].replace("#", "")
parameter_def["values"] = []

# Add allowed values
values = profile_data.query(f"VALUE == '{parameter_dict['range']}' and KEY == 'type'").ID.tolist()

for value in values:

value_namespace, value_name = rdfs_tools.get_namespace_and_name(value, default_namespace=xml_base)
value_meta = profile_data.get_object_data(value).to_dict()

if value_namespace == "":
value_namespace = xml_base

value_def = {
"description": value_meta.get("comment", ""),
"namespace": value_namespace,
"type": "EnumerationValue"
}

parameter_def["values"].append(value_name)
profile[value_name] = value_def

# Add parameter definition
profile[parameter_name] = parameter_def

# Add to class
profile[class_name]["parameters"].append(parameter_name)
parameter_name = add_parameter(parameter, parameter_meta)
if parameter_name is not None:
profile[class_name]["parameters"].append(parameter_name)

# Orphaned attributes: rdf:Property records with no class binding at all (no
# rdfs:domain and no schema:domainIncludes). Their definition is still emitted
# as a top-level entry so it is not lost — it is simply not referenced by any
# class. A consumer can recover the unreferenced set by diffing the property
# entries against the classes' parameter lists. A property that carries a
# binding but is skipped for other reasons (e.g. an unused inverse association)
# is bound, not orphaned.
bound_ids = set(profile_data.query("KEY in ['domain', 'domainIncludes']")["ID"])
all_property_ids = profile_data.query(
"KEY == 'type' and VALUE == 'http://www.w3.org/1999/02/22-rdf-syntax-ns#Property'")["ID"].unique()
orphan_ids = [pid for pid in all_property_ids if pid not in bound_ids]
if orphan_ids:
logger.warning("%d attribute(s) have no class binding (no rdfs:domain / schema:domainIncludes) "
"— emitted without a class: %s", len(orphan_ids), ", ".join(sorted(orphan_ids)))
orphan_rows = profile_data[profile_data["ID"].isin(orphan_ids)].drop_duplicates(["ID", "KEY"])
orphan_table = orphan_rows.pivot(index="ID", columns="KEY")["VALUE"]
for parameter, parameter_meta in orphan_table.iterrows():
add_parameter(parameter, parameter_meta)

return profile

Expand Down
7 changes: 5 additions & 2 deletions triplets/rdfs_tools/rdfs_tools.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,11 @@ def get_class_parameters(data, class_name):

class_data = {"name": class_name}

# Get parameters
class_data["parameters"] = data.query("VALUE == @class_name & KEY == 'domain'")
# Attribute→class binding. CIM-owned terms use rdfs:domain ('domain');
# reused external terms (dcterms:, prov:, dcat:, …) use the non-inferential
# schema:domainIncludes ('domainIncludes') so their meaning is not hijacked
# — both express the same "this attribute belongs to this class".
class_data["parameters"] = data.query("VALUE == @class_name & KEY in ['domain', 'domainIncludes']")

# Add parent classes (if present)
class_data["extends"] = list(data.query("ID == @class_name and KEY == 'subClassOf'")["VALUE"].unique())
Expand Down
Loading