From 27e59654830f8818d48384a85b7d0fcbc0970785 Mon Sep 17 00:00:00 2001 From: Esteban Zimanyi Date: Tue, 19 May 2026 09:27:58 +0200 Subject: [PATCH] refactor: switch TCbuffer regular families to generated mixin Promotes the meos-idl.json-driven OO codegen from Draft preview to a wired-in solution for the TCbuffer vertical slice. - Generator (tools/oo_codegen/codegen.py): adds a behaviourally-faithful emitter (--mixin) that reproduces the hand-written isinstance ladder, argument transforms, result post-processing and super()/raise fallback exactly; only modelled families are wired (one family per PR). The Draft _preview/ path and the #89 coverage gate are unchanged. - pymeos/main/_generated/tcbuffer_methods.py: generated TCbufferRegularMixin (30 regular methods: comparison, spatial relationship, distance, restriction), each dispatching to the exact pymeos_cffi backing the hand-written method used. - pymeos/main/tcbuffer.py: inherits TCbufferRegularMixin; the 30 now-generated regular methods are removed (never_* and the irregular hand-written core are kept). - Provenance: regen-from-meos-api.sh + README document that the catalog is byte-schema-identical to MEOS-API run.py output and the generator runs unchanged on it. Behavioural equivalence is proven by the existing suite: the full tests/main/tcbuffer_test.py is byte-identical before and after the switch -- 26 passed, 1 failed, where the single failure is a pre-existing unrelated pymeos_cffi cbuffer_as_hexwkb out-parameter bug in the base Cbuffer class (not temporal, not codegen). Static check: 30/30 generated methods are backing-and-fallback identical to the hand-written oracle. Stacked on #89; relies on #88's TCbuffer being instantiable. --- pymeos/main/_generated/__init__.py | 1 + pymeos/main/_generated/tcbuffer_methods.py | 573 ++++++++++++++++++ pymeos/main/tcbuffer.py | 660 +-------------------- tools/oo_codegen/README.md | 19 + tools/oo_codegen/codegen.py | 287 +++++++++ tools/oo_codegen/regen-from-meos-api.sh | 43 ++ 6 files changed, 925 insertions(+), 658 deletions(-) create mode 100644 pymeos/main/_generated/__init__.py create mode 100644 pymeos/main/_generated/tcbuffer_methods.py create mode 100755 tools/oo_codegen/regen-from-meos-api.sh diff --git a/pymeos/main/_generated/__init__.py b/pymeos/main/_generated/__init__.py new file mode 100644 index 00000000..a3f7eeb8 --- /dev/null +++ b/pymeos/main/_generated/__init__.py @@ -0,0 +1 @@ +# Generated OO method-family mixins (see tools/oo_codegen). DO NOT EDIT. diff --git a/pymeos/main/_generated/tcbuffer_methods.py b/pymeos/main/_generated/tcbuffer_methods.py new file mode 100644 index 00000000..da3e88b1 --- /dev/null +++ b/pymeos/main/_generated/tcbuffer_methods.py @@ -0,0 +1,573 @@ +# Copyright (c) 2016-2026, Université libre de Bruxelles and PyMEOS +# contributors. Licensed under the PostgreSQL License (see LICENSE). +# +# ============================================================================ +# GENERATED by tools/oo_codegen/codegen.py -- DO NOT EDIT. +# Regenerate: +# python3 tools/oo_codegen/codegen.py --mixin cbuffer \ +# --mixin-out pymeos/main/_generated/cbuffer_methods.py +# ============================================================================ +# +# Wired into pymeos.main.TCbuffer via TCbufferRegularMixin. Every method +# dispatches by argument type to the EXACT pymeos_cffi backing the +# hand-written method used -- same native call, same transforms, same +# result, never reimplemented: identical by construction. +"""Generated regular OO methods for the TCbuffer family.""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +import shapely.geometry.base as shpb +from pymeos_cffi import * + +from ...temporal import Temporal +from ...collections.cbuffer import Cbuffer + +if TYPE_CHECKING: + from ..tcbuffer import TCbuffer + + +class TCbufferRegularMixin: + """Generated regular families (comparison, spatial relationship, + distance, restriction) for :class:`TCbuffer`.""" + + def always_equal(self, other): + """Generated regular ``always_equal``. + + MEOS Functions: + always_eq_cbuffer_tcbuffer, always_eq_tcbuffer_cbuffer, always_eq_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, Cbuffer): + result = always_eq_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = always_eq_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result > 0 + + def always_intersects(self, other): + """Generated regular ``always_intersects``. + + MEOS Functions: + aintersects_tcbuffer_cbuffer, aintersects_tcbuffer_geo, aintersects_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = aintersects_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = aintersects_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = aintersects_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def always_not_equal(self, other): + """Generated regular ``always_not_equal``. + + MEOS Functions: + always_ne_cbuffer_tcbuffer, always_ne_tcbuffer_cbuffer, always_ne_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, Cbuffer): + result = always_ne_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = always_ne_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result > 0 + + def always_touches(self, other): + """Generated regular ``always_touches``. + + MEOS Functions: + atouches_tcbuffer_cbuffer, atouches_tcbuffer_geo, atouches_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = atouches_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = atouches_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = atouches_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def at(self, other): + """Generated regular ``at``. + + MEOS Functions: + tcbuffer_at_cbuffer, tcbuffer_at_geom, tcbuffer_at_stbox + """ + from ...boxes import STBox + + if isinstance(other, shpb.BaseGeometry): + result = tcbuffer_at_geom(self._inner, geo_to_gserialized(other, False)) + elif isinstance(other, Cbuffer): + result = tcbuffer_at_cbuffer(self._inner, other._inner) + elif isinstance(other, STBox): + result = tcbuffer_at_stbox(self._inner, other._inner, True) + else: + return super().at(other) + return Temporal._factory(result) + + def contains(self, other): + """Generated regular ``contains``. + + MEOS Functions: + tcontains_cbuffer_tcbuffer, tcontains_geo_tcbuffer, tcontains_tcbuffer_cbuffer, tcontains_tcbuffer_geo, tcontains_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = tcontains_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = tcontains_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = tcontains_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) + + def covers(self, other): + """Generated regular ``covers``. + + MEOS Functions: + tcovers_cbuffer_tcbuffer, tcovers_geo_tcbuffer, tcovers_tcbuffer_cbuffer, tcovers_tcbuffer_geo, tcovers_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = tcovers_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) + elif isinstance(other, Cbuffer): + result = tcovers_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = tcovers_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) + + def disjoint(self, other): + """Generated regular ``disjoint``. + + MEOS Functions: + tdisjoint_cbuffer_tcbuffer, tdisjoint_geo_tcbuffer, tdisjoint_tcbuffer_cbuffer, tdisjoint_tcbuffer_geo, tdisjoint_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = tdisjoint_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = tdisjoint_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = tdisjoint_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) + + def distance(self, other): + """Generated regular ``distance``. + + MEOS Functions: + tdistance_tcbuffer_cbuffer, tdistance_tcbuffer_geo, tdistance_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = tdistance_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = tdistance_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = tdistance_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) + + def ever_equal(self, other): + """Generated regular ``ever_equal``. + + MEOS Functions: + ever_eq_cbuffer_tcbuffer, ever_eq_tcbuffer_cbuffer, ever_eq_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, Cbuffer): + result = ever_eq_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = ever_eq_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result > 0 + + def ever_intersects(self, other): + """Generated regular ``ever_intersects``. + + MEOS Functions: + eintersects_tcbuffer_cbuffer, eintersects_tcbuffer_geo, eintersects_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = eintersects_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = eintersects_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = eintersects_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def ever_not_equal(self, other): + """Generated regular ``ever_not_equal``. + + MEOS Functions: + ever_ne_cbuffer_tcbuffer, ever_ne_tcbuffer_cbuffer, ever_ne_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, Cbuffer): + result = ever_ne_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = ever_ne_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result > 0 + + def ever_touches(self, other): + """Generated regular ``ever_touches``. + + MEOS Functions: + etouches_tcbuffer_cbuffer, etouches_tcbuffer_geo, etouches_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = etouches_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = etouches_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = etouches_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def intersects(self, other): + """Generated regular ``intersects``. + + MEOS Functions: + tintersects_cbuffer_tcbuffer, tintersects_geo_tcbuffer, tintersects_tcbuffer_cbuffer, tintersects_tcbuffer_geo, tintersects_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = tintersects_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = tintersects_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = tintersects_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) + + def is_always_contains(self, other): + """Generated regular ``is_always_contains``. + + MEOS Functions: + acontains_cbuffer_tcbuffer, acontains_geo_tcbuffer, acontains_tcbuffer_cbuffer, acontains_tcbuffer_geo + """ + if isinstance(other, shpb.BaseGeometry): + result = acontains_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = acontains_tcbuffer_cbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def is_always_covers(self, other): + """Generated regular ``is_always_covers``. + + MEOS Functions: + acovers_cbuffer_tcbuffer, acovers_geo_tcbuffer, acovers_tcbuffer_cbuffer, acovers_tcbuffer_geo + """ + if isinstance(other, shpb.BaseGeometry): + result = acovers_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) + elif isinstance(other, Cbuffer): + result = acovers_tcbuffer_cbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def is_always_disjoint(self, other): + """Generated regular ``is_always_disjoint``. + + MEOS Functions: + adisjoint_tcbuffer_cbuffer, adisjoint_tcbuffer_geo, adisjoint_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = adisjoint_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = adisjoint_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = adisjoint_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def is_always_within_distance(self, other, distance): + """Generated regular ``is_always_within_distance``. + + MEOS Functions: + adwithin_tcbuffer_cbuffer, adwithin_tcbuffer_geo, adwithin_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = adwithin_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False), distance + ) + elif isinstance(other, Cbuffer): + result = adwithin_tcbuffer_cbuffer(self._inner, other._inner, distance) + elif isinstance(other, TCbuffer): + result = adwithin_tcbuffer_tcbuffer(self._inner, other._inner, distance) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def is_ever_contains(self, other): + """Generated regular ``is_ever_contains``. + + MEOS Functions: + econtains_cbuffer_tcbuffer, econtains_tcbuffer_cbuffer, econtains_tcbuffer_geo + """ + if isinstance(other, shpb.BaseGeometry): + result = econtains_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = econtains_tcbuffer_cbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def is_ever_covers(self, other): + """Generated regular ``is_ever_covers``. + + MEOS Functions: + ecovers_cbuffer_tcbuffer, ecovers_tcbuffer_cbuffer, ecovers_tcbuffer_geo, ecovers_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = ecovers_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) + elif isinstance(other, Cbuffer): + result = ecovers_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = ecovers_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def is_ever_disjoint(self, other): + """Generated regular ``is_ever_disjoint``. + + MEOS Functions: + edisjoint_tcbuffer_cbuffer, edisjoint_tcbuffer_geo + """ + if isinstance(other, shpb.BaseGeometry): + result = edisjoint_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = edisjoint_tcbuffer_cbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def is_ever_within_distance(self, other, distance): + """Generated regular ``is_ever_within_distance``. + + MEOS Functions: + edwithin_tcbuffer_cbuffer, edwithin_tcbuffer_geo, edwithin_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = edwithin_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False), distance + ) + elif isinstance(other, Cbuffer): + result = edwithin_tcbuffer_cbuffer(self._inner, other._inner, distance) + elif isinstance(other, TCbuffer): + result = edwithin_tcbuffer_tcbuffer(self._inner, other._inner, distance) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result == 1 + + def minus(self, other): + """Generated regular ``minus``. + + MEOS Functions: + tcbuffer_minus_cbuffer, tcbuffer_minus_geom, tcbuffer_minus_stbox + """ + from ...boxes import STBox + + if isinstance(other, shpb.BaseGeometry): + result = tcbuffer_minus_geom(self._inner, geo_to_gserialized(other, False)) + elif isinstance(other, Cbuffer): + result = tcbuffer_minus_cbuffer(self._inner, other._inner) + elif isinstance(other, STBox): + result = tcbuffer_minus_stbox(self._inner, other._inner, True) + else: + return super().minus(other) + return Temporal._factory(result) + + def nearest_approach_distance(self, other): + """Generated regular ``nearest_approach_distance``. + + MEOS Functions: + nad_cbuffer_stbox, nad_tcbuffer_cbuffer, nad_tcbuffer_geo, nad_tcbuffer_stbox, nad_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + from ...boxes import STBox + + if isinstance(other, shpb.BaseGeometry): + result = nad_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) + elif isinstance(other, Cbuffer): + result = nad_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = nad_tcbuffer_tcbuffer(self._inner, other._inner) + elif isinstance(other, STBox): + result = nad_tcbuffer_stbox(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return result + + def nearest_approach_instant(self, other): + """Generated regular ``nearest_approach_instant``. + + MEOS Functions: + nai_tcbuffer_cbuffer, nai_tcbuffer_geo, nai_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = nai_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) + elif isinstance(other, Cbuffer): + result = nai_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = nai_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) + + def shortest_line(self, other): + """Generated regular ``shortest_line``. + + MEOS Functions: + shortestline_tcbuffer_cbuffer, shortestline_tcbuffer_geo, shortestline_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = shortestline_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = shortestline_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = shortestline_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return gserialized_to_shapely_geometry(result, 10) + + def temporal_equal(self, other): + """Generated regular ``temporal_equal``. + + MEOS Functions: + teq_cbuffer_tcbuffer, teq_tcbuffer_cbuffer + """ + if isinstance(other, Cbuffer): + result = teq_tcbuffer_cbuffer(self._inner, other._inner) + else: + return super().temporal_equal(other) + return Temporal._factory(result) + + def temporal_not_equal(self, other): + """Generated regular ``temporal_not_equal``. + + MEOS Functions: + tne_cbuffer_tcbuffer, tne_tcbuffer_cbuffer + """ + if isinstance(other, Cbuffer): + result = tne_tcbuffer_cbuffer(self._inner, other._inner) + else: + return super().temporal_not_equal(other) + return Temporal._factory(result) + + def touches(self, other): + """Generated regular ``touches``. + + MEOS Functions: + ttouches_cbuffer_tcbuffer, ttouches_geo_tcbuffer, ttouches_tcbuffer_cbuffer, ttouches_tcbuffer_geo, ttouches_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = ttouches_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False) + ) + elif isinstance(other, Cbuffer): + result = ttouches_tcbuffer_cbuffer(self._inner, other._inner) + elif isinstance(other, TCbuffer): + result = ttouches_tcbuffer_tcbuffer(self._inner, other._inner) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) + + def within_distance(self, other, distance): + """Generated regular ``within_distance``. + + MEOS Functions: + tdwithin_geo_tcbuffer, tdwithin_tcbuffer_cbuffer, tdwithin_tcbuffer_geo, tdwithin_tcbuffer_tcbuffer + """ + from ..tcbuffer import TCbuffer + + if isinstance(other, shpb.BaseGeometry): + result = tdwithin_tcbuffer_geo( + self._inner, geo_to_gserialized(other, False), distance + ) + elif isinstance(other, Cbuffer): + result = tdwithin_tcbuffer_cbuffer(self._inner, other._inner, distance) + elif isinstance(other, TCbuffer): + result = tdwithin_tcbuffer_tcbuffer(self._inner, other._inner, distance) + else: + raise TypeError(f"Operation not supported with type " f"{other.__class__}") + return Temporal._factory(result) diff --git a/pymeos/main/tcbuffer.py b/pymeos/main/tcbuffer.py index 35e1ee2b..ff8cf0bd 100644 --- a/pymeos/main/tcbuffer.py +++ b/pymeos/main/tcbuffer.py @@ -12,6 +12,7 @@ from ..collections.cbuffer import Cbuffer, CbufferSet from ..mixins import TTemporallyComparable from ..temporal import Temporal, TInstant, TSequence, TSequenceSet, TInterpolation +from ._generated.tcbuffer_methods import TCbufferRegularMixin if TYPE_CHECKING: from .tbool import TBool @@ -21,6 +22,7 @@ class TCbuffer( + TCbufferRegularMixin, Temporal[Cbuffer, "TCbuffer", "TCbufferInst", "TCbufferSeq", "TCbufferSeqSet"], TTemporallyComparable, ABC, @@ -297,90 +299,6 @@ def from_mfjson(cls, mfjson: str) -> TCbuffer: ) # ------------------------- Ever and Always Comparisons ------------------- - def always_equal(self, value: Union[Cbuffer, TCbuffer]) -> bool: - """ - Returns whether the values of `self` are always equal to `value`. - - Args: - value: :class:`Cbuffer` or :class:`TCbuffer` to compare. - - Returns: - `True` if the values of `self` are always equal to `value`, - `False` otherwise. - - MEOS Functions: - always_eq_tcbuffer_cbuffer, always_eq_tcbuffer_tcbuffer - """ - if isinstance(value, Cbuffer): - return always_eq_tcbuffer_cbuffer(self._inner, value._inner) > 0 - elif isinstance(value, TCbuffer): - return always_eq_tcbuffer_tcbuffer(self._inner, value._inner) > 0 - else: - raise TypeError(f"Operation not supported with type {value.__class__}") - - def always_not_equal(self, value: Union[Cbuffer, TCbuffer]) -> bool: - """ - Returns whether the values of `self` are always not equal to `value`. - - Args: - value: :class:`Cbuffer` or :class:`TCbuffer` to compare. - - Returns: - `True` if the values of `self` are always not equal to `value`, - `False` otherwise. - - MEOS Functions: - always_ne_tcbuffer_cbuffer, always_ne_tcbuffer_tcbuffer - """ - if isinstance(value, Cbuffer): - return always_ne_tcbuffer_cbuffer(self._inner, value._inner) > 0 - elif isinstance(value, TCbuffer): - return always_ne_tcbuffer_tcbuffer(self._inner, value._inner) > 0 - else: - raise TypeError(f"Operation not supported with type {value.__class__}") - - def ever_equal(self, value: Union[Cbuffer, TCbuffer]) -> bool: - """ - Returns whether the values of `self` are ever equal to `value`. - - Args: - value: :class:`Cbuffer` or :class:`TCbuffer` to compare. - - Returns: - `True` if the values of `self` are ever equal to `value`, `False` - otherwise. - - MEOS Functions: - ever_eq_tcbuffer_cbuffer, ever_eq_tcbuffer_tcbuffer - """ - if isinstance(value, Cbuffer): - return ever_eq_tcbuffer_cbuffer(self._inner, value._inner) > 0 - elif isinstance(value, TCbuffer): - return ever_eq_tcbuffer_tcbuffer(self._inner, value._inner) > 0 - else: - raise TypeError(f"Operation not supported with type {value.__class__}") - - def ever_not_equal(self, value: Union[Cbuffer, TCbuffer]) -> bool: - """ - Returns whether the values of `self` are ever not equal to `value`. - - Args: - value: :class:`Cbuffer` or :class:`TCbuffer` to compare. - - Returns: - `True` if the values of `self` are ever not equal to `value`, - `False` otherwise. - - MEOS Functions: - ever_ne_tcbuffer_cbuffer, ever_ne_tcbuffer_tcbuffer - """ - if isinstance(value, Cbuffer): - return ever_ne_tcbuffer_cbuffer(self._inner, value._inner) > 0 - elif isinstance(value, TCbuffer): - return ever_ne_tcbuffer_tcbuffer(self._inner, value._inner) > 0 - else: - raise TypeError(f"Operation not supported with type {value.__class__}") - def never_equal(self, value: Union[Cbuffer, TCbuffer]) -> bool: """ Returns whether the values of `self` are never equal to `value`. @@ -414,585 +332,11 @@ def never_not_equal(self, value: Union[Cbuffer, TCbuffer]) -> bool: return not self.ever_not_equal(value) # ------------------------- Temporal Comparisons -------------------------- - def temporal_equal(self, other: Union[Cbuffer, TCbuffer]) -> TBool: - """ - Returns the temporal equality relation between `self` and `other`. - - Args: - other: A :class:`Cbuffer` or temporal object to compare to `self`. - - Returns: - A :class:`TBool` with the result of the temporal equality relation. - - MEOS Functions: - teq_tcbuffer_cbuffer, teq_temporal_temporal - """ - if isinstance(other, Cbuffer): - result = teq_tcbuffer_cbuffer(self._inner, other._inner) - else: - return super().temporal_equal(other) - return Temporal._factory(result) - - def temporal_not_equal(self, other: Union[Cbuffer, TCbuffer]) -> TBool: - """ - Returns the temporal not equal relation between `self` and `other`. - - Args: - other: A :class:`Cbuffer` or temporal object to compare to `self`. - - Returns: - A :class:`TBool` with the result of the temporal not equal - relation. - - MEOS Functions: - tne_tcbuffer_cbuffer, tne_temporal_temporal - """ - if isinstance(other, Cbuffer): - result = tne_tcbuffer_cbuffer(self._inner, other._inner) - else: - return super().temporal_not_equal(other) - return Temporal._factory(result) - # ------------------------- Restrictions ---------------------------------- - def at(self, other: Union[Cbuffer, shpb.BaseGeometry, STBox, Time]) -> TCbuffer: - """ - Returns a new temporal circular buffer with the values of `self` - restricted to `other`. - - Args: - other: An object to restrict the values of `self` to. - - Returns: - A new :class:`TCbuffer` with the values of `self` restricted to - `other`. - - MEOS Functions: - tcbuffer_at_cbuffer, tcbuffer_at_geom, tcbuffer_at_stbox, - temporal_at_timestamp, temporal_at_tstzset, temporal_at_tstzspan, - temporal_at_tstzspanset - """ - from ..boxes import STBox - - if isinstance(other, Cbuffer): - result = tcbuffer_at_cbuffer(self._inner, other._inner) - elif isinstance(other, shpb.BaseGeometry): - gs = geo_to_gserialized(other, False) - result = tcbuffer_at_geom(self._inner, gs) - elif isinstance(other, STBox): - result = tcbuffer_at_stbox(self._inner, other._inner, True) - else: - return super().at(other) - return Temporal._factory(result) - - def minus(self, other: Union[Cbuffer, shpb.BaseGeometry, STBox, Time]) -> TCbuffer: - """ - Returns a new temporal circular buffer with the values of `self` - restricted to the complement of `other`. - - Args: - other: An object to restrict the values of `self` to the - complement of. - - Returns: - A new :class:`TCbuffer` with the values of `self` restricted to - the complement of `other`. - - MEOS Functions: - tcbuffer_minus_cbuffer, tcbuffer_minus_geom, tcbuffer_minus_stbox, - temporal_minus_timestamp, temporal_minus_tstzset, - temporal_minus_tstzspan, temporal_minus_tstzspanset - """ - from ..boxes import STBox - - if isinstance(other, Cbuffer): - result = tcbuffer_minus_cbuffer(self._inner, other._inner) - elif isinstance(other, shpb.BaseGeometry): - gs = geo_to_gserialized(other, False) - result = tcbuffer_minus_geom(self._inner, gs) - elif isinstance(other, STBox): - result = tcbuffer_minus_stbox(self._inner, other._inner, True) - else: - return super().minus(other) - return Temporal._factory(result) - # ------------------------- Distance Operations --------------------------- - def distance(self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer]) -> TFloat: - """ - Returns the temporal distance between `self` and `other`. - - Args: - other: An object to check the distance to. - - Returns: - A new :class:`TFloat` with the temporal distance. - - MEOS Functions: - tdistance_tcbuffer_geo, tdistance_tcbuffer_cbuffer, - tdistance_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - gs = geo_to_gserialized(other, False) - result = tdistance_tcbuffer_geo(self._inner, gs) - elif isinstance(other, Cbuffer): - result = tdistance_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = tdistance_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - - def nearest_approach_distance( - self, other: Union[shpb.BaseGeometry, Cbuffer, STBox, TCbuffer] - ) -> float: - """ - Returns the nearest approach distance between `self` and `other`. - - Args: - other: An object to check the nearest approach distance to. - - Returns: - A :class:`float` with the nearest approach distance. - - MEOS Functions: - nad_tcbuffer_geo, nad_tcbuffer_cbuffer, nad_tcbuffer_stbox, - nad_tcbuffer_tcbuffer - """ - from ..boxes import STBox - - if isinstance(other, shpb.BaseGeometry): - gs = geo_to_gserialized(other, False) - return nad_tcbuffer_geo(self._inner, gs) - elif isinstance(other, Cbuffer): - return nad_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, STBox): - return nad_tcbuffer_stbox(self._inner, other._inner) - elif isinstance(other, TCbuffer): - return nad_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - - def nearest_approach_instant( - self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer] - ) -> TCbufferInst: - """ - Returns the nearest approach instant between `self` and `other`. - - Args: - other: An object to check the nearest approach instant to. - - Returns: - A new :class:`TCbufferInst` with the nearest approach instant. - - MEOS Functions: - nai_tcbuffer_geo, nai_tcbuffer_cbuffer, nai_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - gs = geo_to_gserialized(other, False) - result = nai_tcbuffer_geo(self._inner, gs) - elif isinstance(other, Cbuffer): - result = nai_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = nai_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - - def shortest_line( - self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer] - ) -> shpb.BaseGeometry: - """ - Returns the shortest line between `self` and `other`. - - Args: - other: An object to check the shortest line to. - - Returns: - A new :class:`~shapely.geometry.base.BaseGeometry` with the - shortest line. - - MEOS Functions: - shortestline_tcbuffer_geo, shortestline_tcbuffer_cbuffer, - shortestline_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - gs = geo_to_gserialized(other, False) - result = shortestline_tcbuffer_geo(self._inner, gs) - elif isinstance(other, Cbuffer): - result = shortestline_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = shortestline_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return gserialized_to_shapely_geometry(result, 10) - # ------------------------- Ever Spatial Relationships -------------------- - def is_ever_contains(self, other: Union[shpb.BaseGeometry, Cbuffer]) -> bool: - """ - Returns whether `self` ever contains `other`. - - MEOS Functions: - econtains_tcbuffer_geo, econtains_tcbuffer_cbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = econtains_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = econtains_tcbuffer_cbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def is_ever_covers( - self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer] - ) -> bool: - """ - Returns whether `self` ever covers `other`. - - MEOS Functions: - ecovers_tcbuffer_geo, ecovers_tcbuffer_cbuffer, - ecovers_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = ecovers_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) - elif isinstance(other, Cbuffer): - result = ecovers_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = ecovers_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def is_ever_disjoint(self, other: Union[shpb.BaseGeometry, Cbuffer]) -> bool: - """ - Returns whether `self` is ever disjoint from `other`. - - MEOS Functions: - edisjoint_tcbuffer_geo, edisjoint_tcbuffer_cbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = edisjoint_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = edisjoint_tcbuffer_cbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def is_ever_within_distance( - self, - other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer], - distance: float, - ) -> bool: - """ - Returns whether `self` is ever within `distance` of `other`. - - MEOS Functions: - edwithin_tcbuffer_geo, edwithin_tcbuffer_cbuffer, - edwithin_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = edwithin_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False), distance - ) - elif isinstance(other, Cbuffer): - result = edwithin_tcbuffer_cbuffer(self._inner, other._inner, distance) - elif isinstance(other, TCbuffer): - result = edwithin_tcbuffer_tcbuffer(self._inner, other._inner, distance) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def ever_intersects( - self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer] - ) -> bool: - """ - Returns whether `self` ever intersects `other`. - - MEOS Functions: - eintersects_tcbuffer_geo, eintersects_tcbuffer_cbuffer, - eintersects_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = eintersects_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = eintersects_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = eintersects_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def ever_touches(self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer]) -> bool: - """ - Returns whether `self` ever touches `other`. - - MEOS Functions: - etouches_tcbuffer_geo, etouches_tcbuffer_cbuffer, - etouches_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = etouches_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = etouches_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = etouches_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - # ------------------------- Always Spatial Relationships ------------------ - def is_always_contains(self, other: Union[shpb.BaseGeometry, Cbuffer]) -> bool: - """ - Returns whether `self` always contains `other`. - - MEOS Functions: - acontains_tcbuffer_geo, acontains_tcbuffer_cbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = acontains_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = acontains_tcbuffer_cbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def is_always_covers(self, other: Union[shpb.BaseGeometry, Cbuffer]) -> bool: - """ - Returns whether `self` always covers `other`. - - MEOS Functions: - acovers_tcbuffer_geo, acovers_tcbuffer_cbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = acovers_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) - elif isinstance(other, Cbuffer): - result = acovers_tcbuffer_cbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def is_always_disjoint( - self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer] - ) -> bool: - """ - Returns whether `self` is always disjoint from `other`. - - MEOS Functions: - adisjoint_tcbuffer_geo, adisjoint_tcbuffer_cbuffer, - adisjoint_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = adisjoint_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = adisjoint_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = adisjoint_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def is_always_within_distance( - self, - other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer], - distance: float, - ) -> bool: - """ - Returns whether `self` is always within `distance` of `other`. - - MEOS Functions: - adwithin_tcbuffer_geo, adwithin_tcbuffer_cbuffer, - adwithin_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = adwithin_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False), distance - ) - elif isinstance(other, Cbuffer): - result = adwithin_tcbuffer_cbuffer(self._inner, other._inner, distance) - elif isinstance(other, TCbuffer): - result = adwithin_tcbuffer_tcbuffer(self._inner, other._inner, distance) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def always_intersects( - self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer] - ) -> bool: - """ - Returns whether `self` always intersects `other`. - - MEOS Functions: - aintersects_tcbuffer_geo, aintersects_tcbuffer_cbuffer, - aintersects_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = aintersects_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = aintersects_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = aintersects_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - - def always_touches( - self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer] - ) -> bool: - """ - Returns whether `self` always touches `other`. - - MEOS Functions: - atouches_tcbuffer_geo, atouches_tcbuffer_cbuffer, - atouches_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = atouches_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = atouches_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = atouches_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return result == 1 - # ------------------------- Temporal Spatial Relationships ---------------- - def contains(self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer]) -> TBool: - """ - Returns a temporal boolean of whether `self` contains `other`. - - MEOS Functions: - tcontains_tcbuffer_geo, tcontains_tcbuffer_cbuffer, - tcontains_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = tcontains_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = tcontains_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = tcontains_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - - def covers(self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer]) -> TBool: - """ - Returns a temporal boolean of whether `self` covers `other`. - - MEOS Functions: - tcovers_tcbuffer_geo, tcovers_tcbuffer_cbuffer, - tcovers_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = tcovers_tcbuffer_geo(self._inner, geo_to_gserialized(other, False)) - elif isinstance(other, Cbuffer): - result = tcovers_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = tcovers_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - - def disjoint(self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer]) -> TBool: - """ - Returns a temporal boolean of whether `self` is disjoint from `other`. - - MEOS Functions: - tdisjoint_tcbuffer_geo, tdisjoint_tcbuffer_cbuffer, - tdisjoint_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = tdisjoint_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = tdisjoint_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = tdisjoint_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - - def within_distance( - self, - other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer], - distance: float, - ) -> TBool: - """ - Returns a temporal boolean of whether `self` is within `distance` of - `other`. - - MEOS Functions: - tdwithin_tcbuffer_geo, tdwithin_tcbuffer_cbuffer, - tdwithin_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = tdwithin_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False), distance - ) - elif isinstance(other, Cbuffer): - result = tdwithin_tcbuffer_cbuffer(self._inner, other._inner, distance) - elif isinstance(other, TCbuffer): - result = tdwithin_tcbuffer_tcbuffer(self._inner, other._inner, distance) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - - def intersects(self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer]) -> TBool: - """ - Returns a temporal boolean of whether `self` intersects `other`. - - MEOS Functions: - tintersects_tcbuffer_geo, tintersects_tcbuffer_cbuffer, - tintersects_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = tintersects_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = tintersects_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = tintersects_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - - def touches(self, other: Union[shpb.BaseGeometry, Cbuffer, TCbuffer]) -> TBool: - """ - Returns a temporal boolean of whether `self` touches `other`. - - MEOS Functions: - ttouches_tcbuffer_geo, ttouches_tcbuffer_cbuffer, - ttouches_tcbuffer_tcbuffer - """ - if isinstance(other, shpb.BaseGeometry): - result = ttouches_tcbuffer_geo( - self._inner, geo_to_gserialized(other, False) - ) - elif isinstance(other, Cbuffer): - result = ttouches_tcbuffer_cbuffer(self._inner, other._inner) - elif isinstance(other, TCbuffer): - result = ttouches_tcbuffer_tcbuffer(self._inner, other._inner) - else: - raise TypeError(f"Operation not supported with type {other.__class__}") - return Temporal._factory(result) - # ------------------------- Database Operations --------------------------- @staticmethod def read_from_cursor(value, _=None): diff --git a/tools/oo_codegen/README.md b/tools/oo_codegen/README.md index 9b0983bb..7a9e13f0 100644 --- a/tools/oo_codegen/README.md +++ b/tools/oo_codegen/README.md @@ -105,3 +105,22 @@ back and re-run: cp ../PyMEOS-CFFI/builder/meos-idl.json tools/oo_codegen/meos-idl.json python3 tools/oo_codegen/codegen.py && black tools/oo_codegen ``` + +### Provenance: regenerate directly from MEOS-API + +`PyMEOS-CFFI/builder/meos-idl.json` is itself a snapshot of the +[MEOS-API](https://github.com/MobilityDB/MEOS-API) parser's output +(`run.py`), byte-schema-identical: `functions[].{file,name,params, +returnType{c,canonical}}`, `params[].{name,cType,canonical}`, +`structs[].fields[].offset_bits`, `enums`. The generator therefore +consumes the canonical MEOS-API catalog with no transformation — it runs +unchanged on a fresh `run.py` output. To reproduce the catalog directly +from MEOS-API against a chosen MEOS ref (no PyMEOS-CFFI intermediary): + +``` +tools/oo_codegen/regen-from-meos-api.sh [MEOS_API_REF] +``` + +The only schema differences a fresh `run.py` shows are additive and +irrelevant to this generator: a top-level `portableAliases` block (used +by PR #87, not here) and a denser `meta/meos-meta.json` enrichment merge. diff --git a/tools/oo_codegen/codegen.py b/tools/oo_codegen/codegen.py index a4d6039c..314b86f9 100644 --- a/tools/oo_codegen/codegen.py +++ b/tools/oo_codegen/codegen.py @@ -176,6 +176,10 @@ class Method: # arg-kind token -> backing pymeos_cffi C function name overloads: dict[str, str] = field(default_factory=dict) c_names: list[str] = field(default_factory=list) + # backing C function name -> its IDL canonical param type list (used by + # the faithful-mixin emitter to detect the trailing `distance` arg and + # the restriction `*_stbox` border flag). + params: dict[str, list[str]] = field(default_factory=dict) @dataclass @@ -225,6 +229,9 @@ def collect(idl: dict) -> tuple[dict[str, dict[str, Method]], Stats]: # so a reviewer sees every backing symbol. meth.overloads.setdefault(token or fam, cname) meth.c_names.append(cname) + meth.params[cname] = [ + p.get("canonical", p.get("cType", "")) for p in entry["params"] + ] st.emitted_overloads += 1 st.by_family[fam] += 1 return fams, st @@ -346,6 +353,255 @@ def emit(fams: dict[str, dict[str, Method]], out_dir: Path) -> None: (out_dir / f"{family}_methods.py").write_text("".join(body)) +# --- faithful wired-in mixin emission ----------------------------------- +# +# The Draft `_preview/` above is a shape sketch. The faithful emitter below +# produces a mixin that is BEHAVIOURALLY identical to the hand-written +# pymeos/main/.py regular families -- same isinstance ladder, same +# argument transforms, same result post-processing, same super()/raise +# fallback -- so the existing pytest suite passes unchanged against it. It +# is modelled per type family; only families with a FAMILY_MODEL entry are +# wired in (the staged migration is one family per PR). + +# Per-family binding: base scalar class, the family's own temporal class, +# the import path of each, and the arg-token -> (isinstance type, call-arg +# expression) map. ``$o`` is the Python argument. Tokens absent from a +# family's map are reported (never silently dropped). +FAMILY_MODEL = { + "cbuffer": { + "mixin_class": "TCbufferRegularMixin", + "base_class": "Cbuffer", + "base_import": "from ...collections.cbuffer import Cbuffer", + "temporal_class": "TCbuffer", + "temporal_import": "from ..tcbuffer import TCbuffer", + # The temporal operand's C-name token. A faithful temporal-type + # mixin uses ONLY `__` overloads (the + # temporal value is the left operand); the reversed `_ + # _` and base-class forms belong to the base + # scalar class, not here -- exactly as the hand-written code does. + "temporal_token": "tcbuffer", + "tokens": { + "cbuffer": ("Cbuffer", "$o._inner"), + "tcbuffer": ("TCbuffer", "$o._inner"), + "geo": ("shpb.BaseGeometry", "geo_to_gserialized($o, False)"), + "geom": ("shpb.BaseGeometry", "geo_to_gserialized($o, False)"), + "stbox": ("STBox", "$o._inner"), + }, + "stbox_lazy": "from ...boxes import STBox", + }, +} + +# Result post-processing, derived verbatim from the hand-written oracle. +_BOOL_GT0 = {"always_equal", "always_not_equal", "ever_equal", "ever_not_equal"} +_BOOL_EQ1 = { + "is_ever_contains", + "is_ever_covers", + "is_ever_disjoint", + "is_ever_within_distance", + "ever_intersects", + "ever_touches", + "is_always_contains", + "is_always_covers", + "is_always_disjoint", + "is_always_within_distance", + "always_intersects", + "always_touches", +} +_SHAPELY = {"shortest_line"} +_RAW = {"nearest_approach_distance"} +# Members whose unmatched-type branch delegates to the generic base impl +# (the rest raise TypeError, mirroring the hand-written code exactly). +_SUPER_FALLBACK = { + "at", + "minus", + "temporal_equal", + "temporal_not_equal", + "temporal_less", + "temporal_less_or_equal", + "temporal_greater", + "temporal_greater_or_equal", +} +# Members that take a trailing ``distance`` argument. +_WITHIN_DISTANCE = { + "is_ever_within_distance", + "is_always_within_distance", + "within_distance", +} + +_ORDER = [ + "geo", + "geom", + "cbuffer", + "tcbuffer", + "npoint", + "tnpoint", + "pose", + "tpose", + "rgeometry", + "trgeometry", + "stbox", +] + +# oo_method_name -> the C-name member prefix (reverse of MEMBER_SPEC). +_OO_TO_CPREFIX = {oo: pfx for pfx, (oo, _k) in MEMBER_SPEC.items()} + + +def _faithful_overloads(oo_name: str, m: Method, ttok: str) -> dict[str, str]: + """Rebuild the dispatch table for a temporal-type mixin: keep only the + overloads whose C name is ``__`` (or, for + restriction, ``_(at|minus)_``) and key on ````. This + discards reversed-operand / base-class forms so the generated method + calls exactly the backing the hand-written method called.""" + out: dict[str, str] = {} + if oo_name in ("at", "minus"): + pat = re.compile(rf"^{re.escape(ttok)}_{oo_name}_(.+)$") + else: + cpre = _OO_TO_CPREFIX[oo_name] + pat = re.compile(rf"^{re.escape(cpre)}_{re.escape(ttok)}_(.+)$") + for cn in m.c_names: + mm = pat.match(cn) + if mm: + out.setdefault(mm.group(1), cn) + return out + + +_MIXIN_HEADER = '''\ +# Copyright (c) 2016-2026, Université libre de Bruxelles and PyMEOS +# contributors. Licensed under the PostgreSQL License (see LICENSE). +# +# ============================================================================ +# GENERATED by tools/oo_codegen/codegen.py -- DO NOT EDIT. +# Regenerate: +# python3 tools/oo_codegen/codegen.py --mixin {family} \\ +# --mixin-out pymeos/main/_generated/{family}_methods.py +# ============================================================================ +# +# Wired into pymeos.main.{temporal_class} via {mixin_class}. Every method +# dispatches by argument type to the EXACT pymeos_cffi backing the +# hand-written method used -- same native call, same transforms, same +# result, never reimplemented: identical by construction. +"""Generated regular OO methods for the {temporal_class} family.""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +import shapely.geometry.base as shpb +from pymeos_cffi import * + +from ...temporal import Temporal +{base_import} + +if TYPE_CHECKING: + {temporal_import} + + +class {mixin_class}: + """Generated regular families (comparison, spatial relationship, + distance, restriction) for :class:`{temporal_class}`.""" +''' + + +def _result_return(oo_name: str) -> str: + if oo_name in _BOOL_GT0: + return " return result > 0\n" + if oo_name in _BOOL_EQ1: + return " return result == 1\n" + if oo_name in _RAW: + return " return result\n" + if oo_name in _SHAPELY: + return " return gserialized_to_shapely_geometry(result, 10)\n" + return " return Temporal._factory(result)\n" + + +def emit_faithful_mixin(family: str, methods: dict[str, Method]) -> str: + """Emit the behaviourally-faithful wired-in mixin for one modelled + family. Raises if a regular overload's arg token has no model mapping + (so a coverage gap is loud, never silent).""" + model = FAMILY_MODEL[family] + tokens = model["tokens"] + ttok = model["temporal_token"] + out = [ + _MIXIN_HEADER.format( + family=family, + mixin_class=model["mixin_class"], + temporal_class=model["temporal_class"], + base_import=model["base_import"], + temporal_import=model["temporal_import"], + ) + ] + + # Faithful dispatch tables (temporal operand first); reversed/base forms + # discarded. Any arg token without a model mapping is loud, not silent. + faithful = {oo: _faithful_overloads(oo, m, ttok) for oo, m in methods.items()} + unmodelled = sorted( + {tok for tbl in faithful.values() for tok in tbl if tok not in tokens} + ) + if unmodelled: + raise SystemExit( + f"[{family}] arg tokens with no FAMILY_MODEL mapping: " + f"{unmodelled} -- extend FAMILY_MODEL before wiring" + ) + + for oo_name in sorted(methods): + m = methods[oo_name] + ov = faithful[oo_name] + if not ov: + # No temporal-operand overload (member is base-class only); + # the hand-written class does not expose it either. + continue + has_dist = oo_name in _WITHIN_DISTANCE + sig = "self, other, distance" if has_dist else "self, other" + body: list[str] = [] + # Lazy imports mirroring the hand-written idiom (self temporal type + # and STBox are imported inside the method to avoid import cycles). + needs_self = any(tokens[t][0] == model["temporal_class"] for t in ov) + needs_stbox = "stbox" in ov + if needs_self: + body.append(f" {model['temporal_import']}\n") + if needs_stbox: + body.append(f" {model['stbox_lazy']}\n") + + branches = [t for t in _ORDER if t in ov] + for i, tok in enumerate(branches): + cfn = ov[tok] + pytype, argexpr = tokens[tok] + argexpr = argexpr.replace("$o", "other") + nparams = len(m.params.get(cfn, [])) + call_args = ["self._inner", argexpr] + if has_dist: + call_args.append("distance") + elif oo_name in ("at", "minus") and tok == "stbox" and nparams == 3: + call_args.append("True") + kw = "if" if i == 0 else "elif" + body.append( + f" {kw} isinstance(other, {pytype}):\n" + f" result = {cfn}({', '.join(call_args)})\n" + ) + if oo_name in _SUPER_FALLBACK: + body.append( + f" else:\n" f" return super().{oo_name}(other)\n" + ) + else: + body.append( + " else:\n" + " raise TypeError(\n" + ' f"Operation not supported with type "\n' + ' f"{other.__class__}"\n' + " )\n" + ) + body.append(_result_return(oo_name)) + + meos_fns = ", ".join(sorted(set(m.c_names))) + out.append( + f"\n def {oo_name}({sig}):\n" + f' """Generated regular ``{oo_name}``.\n\n' + f" MEOS Functions:\n" + f" {meos_fns}\n" + f' """\n' + "".join(body) + ) + return "".join(out) + + # --- driver ------------------------------------------------------------- @@ -359,10 +615,41 @@ def main() -> int: help="exit non-zero if any in-scope function is neither " "emitted nor counted as an explicit exclusion", ) + ap.add_argument( + "--mixin", + metavar="FAMILY", + help="emit the behaviourally-faithful wired-in mixin for one " + "modelled family instead of the Draft preview", + ) + ap.add_argument( + "--mixin-out", + metavar="PATH", + help="destination for --mixin (default: stdout)", + ) args = ap.parse_args() idl = json.loads(Path(args.idl).read_text()) fams, st = collect(idl) + + if args.mixin: + if args.mixin not in FAMILY_MODEL: + raise SystemExit( + f"--mixin {args.mixin!r}: no FAMILY_MODEL (modelled: " + f"{sorted(FAMILY_MODEL)})" + ) + src = emit_faithful_mixin(args.mixin, fams[args.mixin]) + if args.mixin_out: + Path(args.mixin_out).parent.mkdir(parents=True, exist_ok=True) + Path(args.mixin_out).write_text(src) + print( + f"[oo-codegen] wrote faithful mixin " + f"{FAMILY_MODEL[args.mixin]['mixin_class']} -> " + f"{args.mixin_out} ({len(fams[args.mixin])} methods)" + ) + else: + print(src) + return 0 + emit(fams, Path(args.out)) st.emitted_methods = sum(len(m) for m in fams.values()) diff --git a/tools/oo_codegen/regen-from-meos-api.sh b/tools/oo_codegen/regen-from-meos-api.sh new file mode 100755 index 00000000..20973fd9 --- /dev/null +++ b/tools/oo_codegen/regen-from-meos-api.sh @@ -0,0 +1,43 @@ +#!/bin/bash +# Regenerate tools/oo_codegen/meos-idl.json directly from the MEOS-API parser +# (github.com/MobilityDB/MEOS-API run.py), proving the vendored catalog IS +# the canonical MEOS-API artifact and not a hand-massaged copy. +# +# The vendored copy is normally taken from PyMEOS-CFFI's builder/meos-idl.json +# (itself a MEOS-API run.py snapshot, byte-schema-identical: functions{file, +# name,params,returnType}, structs[].fields[].offset_bits, enums). This script +# reproduces that snapshot from MEOS-API directly against a chosen MEOS ref. +# +# Usage: +# tools/oo_codegen/regen-from-meos-api.sh \ +# [MEOS_API_REF] # default: feat/portable-aliases (PR #8) +# +# Then review the diff and re-run the generator + Black: +# python3 tools/oo_codegen/codegen.py --check +# python3 tools/oo_codegen/codegen.py --mixin cbuffer \ +# --mixin-out pymeos/main/_generated/tcbuffer_methods.py +# black tools/oo_codegen pymeos/main/_generated +set -euo pipefail + +INCLUDE_DIR=${1:?usage: regen-from-meos-api.sh [REF]} +REF=${2:-feat/portable-aliases} +HERE=$(cd "$(dirname "$0")" && pwd) +WORK=$(mktemp -d) +trap 'rm -rf "$WORK"' EXIT + +echo "[regen] cloning MEOS-API@$REF ..." +git clone --depth 1 -b "$REF" \ + https://github.com/MobilityDB/MEOS-API "$WORK/MEOS-API" >/dev/null 2>&1 +python3 -m pip install --quiet --user 'libclang==18.1.1' >/dev/null 2>&1 || true + +echo "[regen] running MEOS-API run.py against $INCLUDE_DIR ..." +( cd "$WORK/MEOS-API" && python3 run.py "$INCLUDE_DIR" ) + +cp "$WORK/MEOS-API/output/meos-idl.json" "$HERE/meos-idl.json" +python3 - "$HERE/meos-idl.json" <<'PY' +import json, sys +d = json.load(open(sys.argv[1])) +print(f"[regen] wrote meos-idl.json: {len(d['functions'])} functions, " + f"{len(d.get('structs', []))} structs, {len(d.get('enums', []))} enums") +PY +echo "[regen] done -- review 'git diff', then regenerate the mixin + Black."