From 8a0186f63d4b86475617af993fb49e5015ee877b Mon Sep 17 00:00:00 2001 From: Simone Baffelli Date: Mon, 15 Sep 2025 13:20:26 +0200 Subject: [PATCH] Trying to generate metadataentry automatically --- .../crate/type_property.py | 267 +++++++----------- 1 file changed, 96 insertions(+), 171 deletions(-) diff --git a/0.2.x/lib/python/lib-ro-crate-schema/src/lib_ro_crate_schema/crate/type_property.py b/0.2.x/lib/python/lib-ro-crate-schema/src/lib_ro_crate_schema/crate/type_property.py index 3356116..e58b4ea 100644 --- a/0.2.x/lib/python/lib-ro-crate-schema/src/lib_ro_crate_schema/crate/type_property.py +++ b/0.2.x/lib/python/lib-ro-crate-schema/src/lib_ro_crate_schema/crate/type_property.py @@ -6,10 +6,10 @@ from typing import Annotated, Iterable, List, Optional, Union, TYPE_CHECKING from lib_ro_crate_schema.crate.rdf import SCHEMA, is_type, object_id from lib_ro_crate_schema.crate.literal_type import LiteralType, to_rdf from lib_ro_crate_schema.crate.registry import ForwardRef, Registry -from pydantic import BaseModel, Field, ValidationError, ValidationInfo, field_validator +from pydantic import BaseModel, Field, ValidationError, ValidationInfo, field_validator, create_model from pydantic_rdf import BaseRdfModel, WithPredicate -from rdflib import BNode, Graph, Namespace, URIRef, RDF, RDFS, Literal, OWL, XSD +from rdflib import BNode, Graph, Namespace, URIRef, RDF, RDFS, Literal, OWL, XSD, SDO from pydantic import computed_field @@ -48,21 +48,64 @@ class LiteralType(Enum): case _: raise ValueError(f"Unknown LiteralType: {self}") + @classmethod + def from_external(cls, value: str | URIRef | object) -> "LiteralType": + """ + Import a LiteralType from an external representation. + Accepts: + - enum value (e.g. 'xsd:boolean') + - full URI string (e.g. 'http://www.w3.org/2001/XMLSchema#boolean') + - rdflib URIRef (e.g. XSD.boolean) + - direct rdflib type (e.g. XSD.boolean) + """ + match value: + case str() as s: + for lt in cls: + if s == lt.value: + return lt + for lt in cls: + if s == str(lt.to_internal()): + return lt + case URIRef() as u: + for lt in cls: + if u == lt.to_internal(): + return lt + case _: + for lt in cls: + if value is lt.to_internal(): + return lt + raise ValueError(f"No LiteralType for external value: {value}") + class RdfPropertyType(BaseRdfModel): rdf_type = RDF.Property _rdf_namespace = RDF label: Annotated[str | None, WithPredicate(RDFS.label)] = Field(...) range_includes: Annotated[ - list[Union[URIRef, "RdfType"]], WithPredicate(SCHEMA.RangeIncludes) + list[Union[URIRef, "RdfType"]], WithPredicate(SDO.RangeIncludes) ] = Field(...) def to_external(self) -> "PropertyType": - return PropertyType(id=self.uri, label=self.label, ) + return PropertyType( + id=self.uri, + label=self.label, + range_includes=[convert_range_to_external(r) for r in self.range_includes], + ) +def convert_range_to_external( + range: Union[URIRef, "RdfType"], +) -> Union[LiteralType, "Type"]: + match range: + case URIRef() as ref: + return LiteralType.from_external(ref) + case RdfType() as rdf: + return rdf.to_external() -def convert_range(range: LiteralType | "Type") -> URIRef | "RdfType": + +def convert_range_to_internal( + range: Union[LiteralType, "Type"], +) -> Union[URIRef, "RdfType"]: match range: case LiteralType() as lt: return lt.to_internal() @@ -73,14 +116,14 @@ def convert_range(range: LiteralType | "Type") -> URIRef | "RdfType": class PropertyType(BaseModel): id: str label: str | None - range_includes: Union[LiteralType | "Type"] + range_includes: list[Union[LiteralType, "Type"]] def to_internal(self) -> RdfPropertyType: return RdfPropertyType( uri=self.id, label=self.label, range_includes=[ - convert_range(includes) for includes in self.range_includes + convert_range_to_internal(includes) for includes in self.range_includes ], ) @@ -88,7 +131,7 @@ class PropertyType(BaseModel): class Restriction(BaseRdfModel): rdf_type = OWL.Restriction _rdf_namespace = MY_NS - on_property: Annotated[TypeProperty, WithPredicate(OWL.onProperty)] = Field(...) + on_property: Annotated[RdfPropertyType, WithPredicate(OWL.onProperty)] = Field(...) min_cardinality: Annotated[int, WithPredicate(OWL.minCardinality)] = Field(...) max_cardinality: Annotated[int, WithPredicate(OWL.maxCardinality)] = Field(...) @@ -108,6 +151,9 @@ class RdfType(BaseRdfModel): default=[] ) + def to_external(self) -> "Type": + pass + class Type(BaseModel): id: str @@ -115,19 +161,21 @@ class Type(BaseModel): subclass_of: list["Type"] = Field(default=[]) label: str | None = Field(default=None) comment: str | None = Field(default=None) - properties: list[TypeProperty] = Field(default=[]) + properties: list[PropertyType] = Field(default=[]) def restrictions(self) -> list[Restriction]: return [ Restriction( - uri=BNode(), on_property=prop, min_cardinality=0, max_cardinality=1 + uri=BNode(), + on_property=prop.to_internal(), + min_cardinality=0, + max_cardinality=1, ) for prop in self.properties ] def to_internal(self) -> RdfType: restrictions: list[Restriction] = self.restrictions() - breakpoint() return RdfType( uri=self.id, subclass_of=[c.to_internal() for c in self.subclass_of], @@ -137,6 +185,33 @@ class Type(BaseModel): ) +def generate_model + +class MetadataEntry(BaseModel): + id: str + type: Type + properties: dict[str, str | int | float | bool] + + def to_internal(self) -> BaseRdfModel: + return RdfMetadataEntry.from_external(self) + + +create_model(__base__= BaseRdfModel) +# class RdfMetadataEntry(BaseRdfModel): +# rdf_type: URIRef +# properties: Annotated(dict[str, str], WithPredicate()) +# _rdf_namespace = MY_NS +# @classmethod +# def from_external(cls: type["RdfMetadataEntry"], external: MetadataEntry): +# breakpoint() + +# return cls( +# rdf_type=external.type.to_internal().uri, +# uri=external.id, +# **external.properties, +# ) + + def merge_graphs_from_lists(*graph_lists: Iterable[list[Graph]]) -> Graph: merged = Graph() for g in itertools.chain.from_iterable(graph_lists): @@ -146,175 +221,25 @@ def merge_graphs_from_lists(*graph_lists: Iterable[list[Graph]]) -> Graph: class SchemaFacade(BaseModel): types: List[Type] + entries: List[MetadataEntry] + + def add_type(model: BaseModel): + pass def to_rdf(self): rdf_types: list[Graph] = [t.to_internal().model_dump_rdf() for t in self.types] - breakpoint() - merged = merge_graphs_from_lists(rdf_types) + entries: list[Graph] = [md.to_internal().model_dump_rdf() for e in self.entries] + merged = merge_graphs_from_lists(rdf_types + entries) return merged t0 = Type(id="root", subclass_of=[]) -p1 = PropertyType(uri="d", label="a", range_includes=[LiteralType.INTEGER]) -p2 = PropertyType(uri="d1", label="a1", range_includes=[LiteralType.XML_LITERAL]) +p1 = PropertyType(id="d", label="a", range_includes=[LiteralType.INTEGER]) +p2 = PropertyType(id="d1", label="a1", range_includes=[LiteralType.XML_LITERAL]) t1 = Type(id="c", equivalent_class="a", subclass_of=[t0], properties=[p1, p2]) +md = MetadataEntry(id="a", type=t1, properties={"d": "a", "d1": "a"}) - -class Molecule(BaseModel): - SMILES: str - - -"""" - -1. Importing - -#Gives you all types that exsits in the crate -crate.get_types() -> List[BaseModel]: - -#Reads in an object given a Pydantic BaseModel (which is defined on the receiving side) -# (Static workflow if we know that we have a structrually compatible type) -crate.read_as(Molecule, my_crate, id) -> Molecule | None: - -2. Exporting - -# Will add the schema to a crate -create.add_to_schema(Molecule) - -m1 = Molecule() - -#Will add the metadat and the schema to the crate -crate.add(m1) - - -3. Manual / Fine Grained implementation (for conformity with the Java Implementation) - -p1 = Property(...) -t1 = Type(properties=[p1,...]) - -4. Confortimy: -Internally we convert into RdfsClasses and RdfTypes -and we expose a java style API for conformity with the openBIS requirements. -The export is valid by definiton because we generate valid rdf and add it to a ro-crate - -"""" - - -f1 = SchemaFacade(types=[t1]) +f1 = SchemaFacade(types=[t1], entries=[md]) g1 = f1.to_rdf() print(g1.serialize(format="turtle")) - - -# class TypeProperty(BaseModel): -# id: str -# label: Optional[str] = None -# comment: Optional[str] = None -# _domain_includes: Optional[List[ForwardRef["Type"]]] = None # internal use only -# range_includes: Optional[List[Union[LiteralType, ForwardRef["Type"], "Type"]]] = ( -# None -# ) -# ontological_annotations: Optional[List[str]] = None - -# @field_validator("range_includes", mode="before") -# @classmethod -# def wrap_forward_refs( -# cls, v: Optional[List[Union[LiteralType, ForwardRef["Type"]]]] -# ): -# """ -# Allows the user-facing API to specify the forward reference as a string -# """ -# match v: -# case None: -# return v -# case els: -# values = [] -# for range_element in els: -# match range_element: -# case LiteralType(): -# values.append(range_element) -# case ForwardRef(ref): -# values.append(range_element) -# case str(ref): -# values.append(ForwardRef(ref=ref)) -# return values - -# @property -# def domain_includes(self) -> Optional[List[str]]: -# # For serialization only -# return self._domain_includes - -# # @property -# # def range_includes(self) -> Optional[List[str]]: -# # # For serialization only -# # return self._range_includes - -# def resolve(self, registry: Registry): -# """ -# Resolve all references to types -# """ -# from lib_ro_crate_schema.crate.type import Type -# range_includes = [] -# domain_includes = [] -# for range_element in self.range_includes: -# match range_element: -# case Type() | LiteralType(): -# range_includes.append(range_element) -# case ForwardRef(): -# print(range_element) -# range_includes.append(registry.resolve(range_element)) -# case _: -# raise TypeError( -# f"Unsupported range_includes element: {range_element!r}" -# ) - -# for domain_element in self._domain_includes if self._domain_includes else []: -# match domain_element: -# case Type(): -# domain_includes.append(domain_element) -# case ForwardRef(): -# domain_element.append(registry.resolve(domain_element)) -# case _: -# raise TypeError( -# f"Unsupported range_includes element: {domain_element!r}" -# ) -# self._domain_includes = domain_includes -# self.range_includes = range_includes - -# def _resolve_range_includes(self): -# """ """ -# from lib_ro_crate_schema.crate.type import Type - -# resolved = [] -# if not self.range_includes: -# return resolved -# for range_element in self.range_includes: -# match range_element: -# case Type(id=tid): -# resolved.append(object_id(tid)) -# case LiteralType(): -# resolved.append(to_rdf(range_element)) -# case str(ref): -# resolved.append(URIRef(ref)) -# case _: -# raise TypeError( -# f"Unsupported range_includes element: {range_element!r}" -# ) -# return resolved - -# def to_triples(self, subject=None): -# subj = object_id(self.id) if subject is None else subject -# yield (subj, RDF.type, RDF.Property) -# if self.label: -# yield (subj, RDFS.label, Literal(self.label)) -# if self.comment: -# yield (subj, RDFS.comment, Literal(self.comment)) -# if self.domain_includes: -# for d in self.domain_includes: -# yield (subj, SCHEMA.domainIncludes, URIRef(d)) -# for r in self._resolve_range_includes(): -# print(type(r), r) -# yield (subj, SCHEMA.rangeIncludes, r) -# if self.ontological_annotations: -# for r in self.ontological_annotations: -# yield (subj, OWL.equivalentClass, URIRef(r)) -# # Add more as needed for data types, annotations, etc.