Model

model

Package containing models, the pythonic representation of feature classes.

They inherit from BaseFeature, which extends the Pydantic BaseModel with some utility, and are resolved via an Appschema instance. A feature collection is represented by the BaseCollection class.

Example

Load the BP_Plan model for XPlanung v6.0 and instantiate it with some data:

from xplan_tools.model.base import Appschema

plan = Appschema.from_prefix("xplan", "6.0").model_factory("BP_Plan")
instance = plan.model_validate(
    {
        "name": "Testplan",
        "gemeinde": [
            {
                "ags": "1234"
            }
        ],
        "raeumlicherGeltungsbereich": {
            "srid": 25832,
            "wkt": <WKT-String>
        }
    }
)

Appschema

Bases: BaseModel

Models a supported appschema.

Used to access metadata like name, version and description for an appschema.

During instantion, validation if an appschema is supported occurs, based on the modules in appschema subdirectory.

The instance stores a reference to the appschema's module, which is used in the model_factory method the retrieve appschema classes.

Usage example

# get a featuretype from appschema
appschema = Appschema.from_prefix(prefix="xplan", version="6.0")
plan = appschema.model_factory(name="BP_Plan")
# show list of supported appschemas
Appschema.supported_appschemas()
# get an enum of supported appschemas
SupportedAppschemas = Appschema.enum()
SupportedAppschemas.XPLAN_6_0.value # -> XPlanGML 6_0

code property

The enum code of the appschema.

version property

The version of the appschema in <major>.<minor> format.

enum() classmethod

Return an enumeration of supported appschemas.

Source code in xplan_tools/model/base.py

@classmethod
def enum(cls) -> type[StrEnum]:
    """Return an enumeration of supported appschemas."""
    if not cls._enum:

        def _new(
            cls,
            value,
            appschema,
        ):
            obj = str.__new__(cls, value)
            obj._value_ = value
            obj.appschema = appschema
            return obj

        cls._enum = StrEnum(
            "SupportedAppschemas",
            {
                f"{appschema.prefix.upper()}_{appschema.full_version.major}_{appschema.full_version.minor}": (
                    f"{appschema.prefix.upper()}_{appschema.version.replace('.', '_')}",
                    appschema,
                )
                for appschema in cls.supported_appschemas()
            }
            | {"__new__": _new, "__annotations__": {"appschema": Appschema}},
        )
    return cls._enum

from_enum(code) classmethod

Return an Appschema instance from enum code.

Parameters:

Name	Type	Description	Default
code	str	an appschema code, e.g. XPLAN_6_0	required

Source code in xplan_tools/model/base.py

@classmethod
def from_enum(cls, code: str) -> Appschema:
    """Return an Appschema instance from enum code.

    Args:
        code: an appschema code, e.g. `XPLAN_6_0`
    """
    if enum := getattr(cls.enum(), code, None):
        return enum.appschema
    else:
        raise NotImplementedError(f"no appschema with code {code!r}")

from_module(module_name) cached classmethod

Builds an Appschema instance from module name.

Parameters:

Name	Type	Description	Default
module_name	str	the fully qualified module name	required

Source code in xplan_tools/model/base.py

@classmethod
@functools.cache
def from_module(cls, module_name: str) -> Appschema:
    """Builds an Appschema instance from module name.

    Args:
        module_name: the fully qualified module name
    """
    try:
        module = import_module(module_name)
        model_cls: type[RootModel] = getattr(module, "Model")
    except (ImportError, AttributeError) as exc:
        raise RuntimeError(
            f"Unable to resolve appschema metadata for module {module_name!r}"
        ) from exc

    if not issubclass(model_cls, RootModel):
        raise ValueError(f"expected RootModel, got {model_cls!r}")

    metadata = model_cls.model_fields["root"]

    return cls.model_validate(
        metadata.json_schema_extra
        | {"description": metadata.description, "module": module}
    )

from_namespace(namespace) classmethod

Builds an Appschema instance from the appschema's namespace.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name	Type	Description	Default
namespace	str	the namespace URI of the appschema	required

Source code in xplan_tools/model/base.py

@classmethod
def from_namespace(cls, namespace: str) -> Appschema:
    """Builds an Appschema instance from the appschema's namespace.

    Might return a compatible appschema version if no exact match is found.

    Args:
        namespace: the namespace URI of the appschema
    """
    uri = AnyUrl(namespace)
    candidates: list[Appschema] = []
    for appschema in cls.supported_appschemas():
        if appschema.namespace_uri == uri:
            return appschema
        elif str(appschema.namespace_uri)[:-1] == namespace[:-1]:
            candidates.append(appschema)
    if candidates:
        return sorted(candidates)[-1]
    raise NotImplementedError(f"no appschema with namespace {namespace!r}")

from_prefix(prefix, version) classmethod

Builds an Appschema instance from the appschema's prefix and version.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name	Type	Description	Default
prefix	str	the namespace prefix of the appschema, e.g. xplan or xtrasse	required
version	str	the version of the appschema; major and minor are required	required

Source code in xplan_tools/model/base.py

@classmethod
def from_prefix(cls, prefix: str, version: str) -> Appschema:
    """Builds an Appschema instance from the appschema's prefix and version.

    Might return a compatible appschema version if no exact match is found.

    Args:
        prefix: the namespace prefix of the appschema, e.g. `xplan` or `xtrasse`
        version: the version of the appschema; major and minor are required
    """
    sem_version = Version.parse(version, optional_minor_and_patch=True)
    versions = []
    candidates = []
    for appschema in filter(
        lambda appschema: appschema.prefix == prefix, cls.supported_appschemas()
    ):
        # if exact match, return
        if (
            sem_version.major == appschema.full_version.major
            and sem_version.minor == appschema.full_version.minor
        ):
            return appschema
        # if versions are compatible (modules minor version higher or equal), add to candidates
        elif sem_version.is_compatible(appschema.full_version):
            candidates.append(appschema)
        else:
            versions.append(
                f"{appschema.full_version.major}.{appschema.full_version.minor}"
            )
    # return appschema with newer minor version, if found
    # TODO: ensure minor version compatibility without version migration, e.g. via model_validator
    if candidates:
        return sorted(candidates)[-1]
    if versions:
        e = NotImplementedError(
            f"version {version!r} not supported for prefix {prefix!r}"
        )
        e.add_note(f"available versions: {', '.join(sorted(versions))}")
        raise e
    else:
        raise NotImplementedError(f"no appschema with prefix {prefix!r}")

model_factory(name)

Factory method for retrieving the corresponding pydantic model representation of a feature class.

Parameters:

Name	Type	Description	Default
name	str	name of the feature class or enumeration	required

Raises:

Type	Description
ValueError	requested FeatureType not found

Returns:

Type	Description
type[BaseFeature]	The concrete feature class inheriting from BaseFeature.

Source code in xplan_tools/model/base.py

def model_factory(self, name: str) -> type[BaseFeature]:
    """Factory method for retrieving the corresponding pydantic model representation of a feature class.

    Args:
        name: name of the feature class or enumeration

    Raises:
        ValueError: requested FeatureType not found

    Returns:
        The concrete feature class inheriting from BaseFeature.
    """
    try:
        cls: BaseFeature = getattr(self.module, name)
        return _ensure_subclass(cls, BaseFeature)
    except AttributeError:
        raise ValueError(f"featuretype {name!r} not found for appschema {self!r}")

supported_appschemas() classmethod

Return a list of currently supported appschemas.

Source code in xplan_tools/model/base.py

@classmethod
def supported_appschemas(cls) -> list[Appschema]:
    """Return a list of currently supported appschemas."""
    if not cls._supported_appschemas:
        cls._supported_appschemas = [
            cls.from_module(f"{__package__}.appschema.{module_name}")
            for module_name in APPSCHEMA_MODULE_NAMES
        ]
    return cls._supported_appschemas

BaseFeature

Bases: _ConfiguredBaseModel, GMLAdapter, CoretableAdapter, JsonFGAdapter

Base class for application schema classes.

It extends pydantic BaseModel with Feature-related helper methods as well as conversion capabilities from/to other formats via inheriting from respective adapter classes.

appschema() cached classmethod

Return metadata about the application schema for the feature class.

Source code in xplan_tools/model/base.py

@classmethod
@functools.cache
def appschema(cls) -> Appschema:
    """Return metadata about the application schema for the feature class."""
    return Appschema.from_module(cls.__module__)

get_associations() cached classmethod

Returns the classes association fields.

Source code in xplan_tools/model/base.py

@classmethod
@functools.cache
def get_associations(cls) -> list[str]:
    """Returns the classes association fields."""
    return [
        assoc
        for assoc in cls.model_fields.keys()
        if cls.get_property_info(assoc).stereotype == "Association"
    ]

get_geom_field() cached classmethod

Returns the classes geometry field name, if any.

Source code in xplan_tools/model/base.py

@classmethod
@functools.cache
def get_geom_field(cls) -> str | None:
    """Returns the classes geometry field name, if any."""
    if attr := {
        "position",
        "geltungsbereich",
        "raeumlicherGeltungsbereich",
        "geltungsbereichAenderung",
        "extent",
        "geometry",
    }.intersection(set(cls.model_fields.keys())):
        return attr.pop()

get_geom_srid()

Returns the object's geometry's SRID, if any.

Source code in xplan_tools/model/base.py

def get_geom_srid(self) -> int | None:
    """Returns the object's geometry's SRID, if any."""
    if geom_field := self.get_geom_field():
        if geom := getattr(self, geom_field, None):
            return geom.srid

get_geom_types() cached classmethod

Returns the types of the geometry attribute.

Source code in xplan_tools/model/base.py

@classmethod
@functools.cache
def get_geom_types(cls) -> list[type[BaseFeature]] | None:
    """Returns the types of the geometry attribute."""
    if geom_field := cls.get_geom_field():
        geom_annotation = cls.model_fields[geom_field].annotation
        args = get_args(geom_annotation)
        if not args:
            geom_model = [geom_annotation]
        else:
            geom_model = [arg for arg in args if arg is not NoneType]
        return geom_model

get_geom_wkt()

Returns the object's eWKT geometry's WKT representation withouth SRID, if any.

Source code in xplan_tools/model/base.py

def get_geom_wkt(self) -> str | None:
    """Returns the object's eWKT geometry's WKT representation withouth SRID, if any."""
    if geom_field := self.get_geom_field():
        if geom := getattr(self, geom_field, None):
            return geom.wkt

get_name() cached classmethod

Returns the canonical name of the FeatureClass.

Source code in xplan_tools/model/base.py

@classmethod
@functools.cache
def get_name(cls) -> str:
    """Returns the canonical name of the FeatureClass."""
    return cls.__name__

get_property_info(name) cached classmethod

Property information.

Parameters:

Name	Type	Description	Default
name	str	The property's name.	required

Returns:

Type	Description
PropertyInfo	A typed dataclass holding the information.

Raises:

Type	Description
AttributeError	The name was not found in the model fields.

Source code in xplan_tools/model/base.py

@classmethod
@functools.cache
def get_property_info(cls, name: str) -> PropertyInfo:
    """Property information.

    Args:
        name: The property's name.

    Returns:
        A typed dataclass holding the information.

    Raises:
        AttributeError: The name was not found in the model fields.
    """
    try:
        field_info = cls.model_fields[name]
        extra_info = field_info.json_schema_extra or {}
    except KeyError:
        raise AttributeError(f"Unknown property: {name}")
    else:
        stereotype: Final = extra_info["stereotype"]
        typename: str | list[str] = extra_info["typename"]
        return PropertyInfo(
            stereotype=stereotype,
            typename=typename,
            list=get_origin(field_info.annotation) is list
            or list
            in [
                arg.__origin__
                for arg in get_args(field_info.annotation)
                if getattr(arg, "__origin__", None)
            ],
            nullable=NoneType in get_args(field_info.annotation),
            uom=extra_info.get("uom", None),
            enum=_ensure_subclass(
                getattr(cls.appschema().module, str(typename)), BaseEnum
            )
            if stereotype == "Enumeration"
            else None,
            assoc_info=None
            if stereotype != "Association"
            else AssocInfo(
                reverse=extra_info.get("reverseProperty", None),
                source_or_target=extra_info.get("sourceOrTarget", None),
            ),
        )

model_dump_coretable()

Dumps the model data to a coretable Feature object to store in a database.

Source code in xplan_tools/model/base.py

def model_dump_coretable(
    self,
) -> Feature:
    """Dumps the model data to a coretable Feature object to store in a database."""
    return self._to_coretable()

model_dump_coretable_bulk()

Dumps the model data to feature and refs dicts to bulk insert in a database.

Source code in xplan_tools/model/base.py

def model_dump_coretable_bulk(self) -> tuple[dict, list[dict]]:
    """Dumps the model data to feature and refs dicts to bulk insert in a database."""
    return self._to_coretable(bulk_mode=True)

model_dump_gml(**kwargs)

Dumps the model data to a GML structure held in an etree.Element.

Source code in xplan_tools/model/base.py

def model_dump_gml(
    self,
    **kwargs,
) -> _Element:
    """Dumps the model data to a GML structure held in an etree.Element."""
    return self._to_etree(**kwargs)

model_dump_jsonfg(**kwargs)

Dumps the model data to a JSON-FG object.

Source code in xplan_tools/model/base.py

def model_dump_jsonfg(
    self,
    **kwargs,
) -> dict:
    """Dumps the model data to a JSON-FG object."""
    return self._to_jsonfg(**kwargs)

BaseCollection

Bases: BaseModel

Container for features that provides validation of references.

The features are stored in a dictionary with their ID as key and the feature instance as value.

add_style_properties(to_text=False, always_populate_schriftinhalt=False)

Add styling properties to presentational objects.

This method parses object (dientZurDarstellungVon) and property (art) references from presentational objects and derives styling information (stylesheetId, schriftinhalt) based on a set of defined rules.

Parameters:

Name	Type	Description	Default
to_text	bool	Whether to convert symbolic presentational objects to textual ones. Defaults to False.	False
always_populate_schriftinhalt	bool	Populate schriftinhalt even if a rule has no text template.	False

Source code in xplan_tools/model/base.py

def add_style_properties(
    self, to_text: bool = False, always_populate_schriftinhalt: bool = False
) -> None:
    """Add styling properties to presentational objects.

    This method parses object (dientZurDarstellungVon) and property (art) references from
    presentational objects and derives styling information (stylesheetId, schriftinhalt)
    based on a set of defined rules.

    Args:
        to_text: Whether to convert symbolic presentational objects to textual ones. Defaults to False.
        always_populate_schriftinhalt: Populate `schriftinhalt` even if a rule has no text template.
    """
    logger.info("adding style properties to collection")

    for obj in filter(lambda x: hasattr(x, "stylesheetId"), self.get_features()):
        if not obj.dientZurDarstellungVon:
            logger.info(
                f"Feature {obj.id}: dientZurDarstellungVon not set, skipping"
            )
            continue
        elif len(obj.dientZurDarstellungVon) > 1:
            logger.warning(
                f"Feature {obj.id}: references to multiple objects '{obj.dientZurDarstellungVon}' not supported, skipping"
            )
            continue
        elif not obj.art:
            logger.info(f"Feature {obj.id}: art not set, skipping")
            continue
        ref_id = obj.dientZurDarstellungVon[0]
        try:
            ref_obj = self.features[ref_id]
        except KeyError:
            raise ValueError(
                f"Feature {obj.id}: dientZurDarstellungVon references unknown feature {ref_id}"
            )
        new_obj = add_style_properties_to_feature(
            obj, ref_obj, to_text, always_populate_schriftinhalt
        )
        self.features[obj.id] = new_obj

    logger.info("finished adding style properties to collection")

check_references_and_srs()

Checks if all objects referenced via UUID are part of the collection and if all features have the same SRS, version and appschema.

Source code in xplan_tools/model/base.py

@model_validator(mode="after")
def check_references_and_srs(self) -> Self:
    """Checks if all objects referenced via UUID are part of the collection and if all features have the same SRS, version and appschema."""
    logger.debug("checking feature references")
    srids = []
    appschemas = []

    def _validate_ref(feature: BaseFeature, name: str, value: UUID):
        prop_info = feature.get_property_info(name)
        try:
            ref_feature = self.features[value]
        except KeyError:
            raise ValueError(
                f"reference {name}: {value} in object {feature.id} not resolvable"
            )
        if not prop_info.is_type_ok((ref_name := ref_feature.get_name())):
            raise ValueError(
                f"association {name} in object {feature.id} references incompatible type {ref_name}"
            )

    for feature in self.features.values():
        if srid := feature.get_geom_srid():
            if srids and srid not in srids:
                raise ValueError(
                    f"Multiple SRS within collection not supported: SRID {srid} != {srids[0]} for feature {feature.id}"
                )
            else:
                srids.append(srid)
        appschema = feature.appschema()
        if appschemas and appschema not in appschemas:
            raise ValueError(
                f"Multiple appschemas within collection not supported: appschema {appschema} != {appschemas[0]} for feature {feature.id}"
            )
        else:
            appschemas.append(appschema)

        for assoc in feature.get_associations():
            value = getattr(feature, assoc)
            if isinstance(value, UUID):
                _validate_ref(feature, assoc, value)
            elif isinstance(value, list):
                for item in value:
                    if isinstance(item, UUID):
                        _validate_ref(feature, assoc, item)

    logger.debug("all feature references resolvable")
    return self

get_features()

Yields features stored in the collection.

Source code in xplan_tools/model/base.py

def get_features(self) -> Iterator["BaseFeature"]:
    """Yields features stored in the collection."""
    return (feature for feature in self.features.values())

get_single_plans(with_name=False)

Yields BaseCollection objects for every plan in the original collection.

Source code in xplan_tools/model/base.py

def get_single_plans(
    self, with_name: bool = False
) -> Iterator[BaseCollection] | Iterator[tuple[str, BaseCollection]]:
    """Yields BaseCollection objects for every plan in the original collection."""
    for plan in filter(
        lambda x: x.get_name().endswith("Plan"), self.features.values()
    ):
        collection = {plan.id: plan}
        for attr in ["texte", "begruendungsTexte"]:
            if refs := getattr(plan, attr):
                collection.update({ref: self.features[ref] for ref in refs})
        for bereich in filter(
            lambda x: getattr(x, "gehoertZuPlan", None) == plan.id,
            self.features.values(),
        ):
            collection[bereich.id] = bereich
            for feature in filter(
                lambda x: getattr(x, "gehoertZuBereich", None) == bereich.id,
                self.features.values(),
            ):
                collection[feature.id] = feature
                for attr in ["refBegruendungInhalt", "refTextInhalt"]:
                    if refs := getattr(feature, attr, None):
                        collection.update({ref: self.features[ref] for ref in refs})
            if raster := getattr(bereich, "rasterBasis", None):
                collection[raster] = self.features[raster]
            for attr in [
                "nachrichtlich",
                "inhaltBPlan",
                "inhaltFPlan",
                "inhaltLPlan",
                "inhaltRPlan",
                "inhaltSoPlan",
                "rasterAenderung",
            ]:
                if refs := getattr(bereich, attr, None):
                    collection.update({ref: self.features[ref] for ref in refs})
        yield (
            (
                plan.name,
                BaseCollection(
                    features=collection,
                    srid=self.srid,
                    appschema=self.appschema,
                ),
            )
            if with_name
            else BaseCollection(
                features=collection,
                srid=self.srid,
                appschema=self.appschema,
            )
        )

list_to_dict(data) classmethod

Takes a list of BaseFeatures and returns a BaseCollection dict.

Source code in xplan_tools/model/base.py

@model_validator(mode="before")
@classmethod
def list_to_dict(cls, data: Any) -> Any:
    """Takes a list of BaseFeatures and returns a BaseCollection dict."""
    if isinstance(data, list):
        data_dict = {}
        for feature in data:
            if not isinstance(feature, BaseFeature):
                raise TypeError(
                    f"Object is not an instance of BaseFeature: {feature}"
                )
            data_dict[feature.id] = feature
        return data_dict
    return data

make_copy(with_id_map=False)

Return a copy of the collection with new IDs.

All feature IDs are renewed and respective references are updated.

Parameters:

Name	Type	Description	Default
with_id_map	bool	whether to additionally return a map of old IDs to new IDs	False

Source code in xplan_tools/model/base.py

def make_copy(
    self, with_id_map: bool = False
) -> BaseCollection | tuple[BaseCollection, dict]:
    """Return a copy of the collection with new IDs.

    All feature IDs are renewed and respective references are updated.

    Args:
        with_id_map: whether to additionally return a map of old IDs to new IDs
    """
    id_map = {key: uuid4() for key in self.features.keys()}
    new_features = {}
    for old_feature in self.features.values():
        new_id = id_map[old_feature.id]
        new_feature = old_feature.model_copy(deep=True)
        new_feature.id = new_id
        for assocation in old_feature.get_associations():
            old_value = getattr(old_feature, assocation)
            if isinstance(old_value, UUID):
                new_value = id_map[old_value]
                setattr(new_feature, assocation, new_value)
            elif isinstance(old_value, list):
                new_list = [
                    id_map[item] if isinstance(item, UUID) else item
                    for item in old_value
                ]
                setattr(new_feature, assocation, new_list)
        new_features[new_id] = new_feature

    new_collection = BaseCollection(
        features=new_features,
        srid=self.srid,
        appschema=self.appschema,
    )
    if with_id_map:
        return new_collection, id_map
    else:
        return new_collection