# Copyright (c) 2024-2026 Karana Dynamics Pty Ltd. All rights reserved.
#
# NOTICE TO USER:
#
# This source code and/or documentation (the "Licensed Materials") is
# the confidential and proprietary information of Karana Dynamics Inc.
# Use of these Licensed Materials is governed by the terms and conditions
# of a separate software license agreement between Karana Dynamics and the
# Licensee ("License Agreement"). Unless expressly permitted under that
# agreement, any reproduction, modification, distribution, or disclosure
# of the Licensed Materials, in whole or in part, to any third party
# without the prior written consent of Karana Dynamics is strictly prohibited.
#
# THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
# KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
# BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
# FITNESS FOR A PARTICULAR PURPOSE.
#
# IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
# INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
# ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
# OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
#
# U.S. Government End Users: The Licensed Materials are a "commercial item"
# as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
# only as a commercial end item under the terms of this license.
#
# Any use of the Licensed Materials in individual or commercial software must
# include, in the user documentation and internal source code comments,
# this Notice, Disclaimer, and U.S. Government Use Provision.
"""KCore contains a collection of classes that form core infrastructure for many other classes through Karana software.
Examples include Base and LockingBase which serve as base classes for most simulation-related objects.
"""
# Ignore F405 for the whole file. This takes care of things imported from *
# ruff: noqa: F405
import datetime
from os import getenv
from collections.abc import Callable
from typing import (
Annotated,
TYPE_CHECKING,
Self,
TypeVar,
cast,
overload,
TypeAlias,
Union,
Generic,
get_args,
get_origin,
Any,
)
from traceback import format_stack
import numpy as np
from numpy.typing import NDArray
import Karana.Math as km
from Karana.Math import Ktyping as kt, Kquantities as kq
# Define Json type alias before loading in pybind11
Json: TypeAlias = dict[str, Union[str, int, float, "Json"]]
from Karana.Core._KCore_pybind11_Py import * # noqa: E402, F403
from Karana.Core._KCore_pybind11_Py import deprecated as _deprecated # noqa: E402
from Karana.Core._KCore_pybind11_Py import Var as _BoundVar # noqa: E402
import Karana.Core._KCore_pybind11_Py as _bindings # noqa: E402
Char: TypeAlias = Annotated[str, "Char"]
"""A single character stored by the C++ char specialization."""
def _specializeVar(cls: type[_BoundVar], value_type: object) -> type[_BoundVar]:
"""Select a bound C++ Var specialization without constructing an object.
Parameters
----------
cls : type[Var]
The subscribed class; subscription is supported on the Var base only.
value_type : object
A supported Python scalar, Ktyping array alias,
or Karana.Math class.
Returns
-------
type[Var]
The existing concrete class, with its constructors and create method.
Raises
------
TypeError
If the class or value type has no registered specialization.
"""
if cls is not _BoundVar:
raise TypeError("Use Var[value_type] to select a specialization")
specializations: dict[object, type[_BoundVar]] = {
float: _bindings._VarDouble,
np.float64: _bindings._VarDouble,
np.float32: _bindings._VarFloat,
int: _bindings._VarInt,
np.int16: _bindings._VarShort,
np.int32: _bindings._VarInt,
np.int64: _bindings._VarLong,
np.uint64: _bindings._VarSizeT,
bool: _bindings._VarBool,
str: _bindings._VarString,
Char: _bindings._VarChar,
list[str]: _bindings._VarVecString,
kt.Vec: _bindings._VarVec,
kt.Vec3: _bindings._VarVec3,
kt.Vec4: _bindings._VarVec4,
kt.Vec6: _bindings._VarVec6,
kt.VecInt: _bindings._VarVecInt,
kt.Mat: _bindings._VarMat,
kt.Mat33: _bindings._VarMat33,
kt.Mat66: _bindings._VarMat66,
km.UnitQuaternion: _bindings._VarUnitQuaternion,
km.SpatialVector: _bindings._VarSpatialVector,
km.SpatialVelocity: _bindings._VarSpatialVelocity,
km.SpatialAcceleration: _bindings._VarSpatialAcceleration,
km.SpatialForce: _bindings._VarSpatialForce,
km.HomTran: _bindings._VarHomTran,
}
origin = get_origin(value_type)
if origin is not None and getattr(origin, "__module__", "").startswith("Karana.Math.Ktyping"):
(magnitude,) = get_args(value_type)
quantity_name = getattr(origin, "__name__", "")
numeric = specializations.get(magnitude)
if numeric is not None:
for suffix in (
"Double",
"Float",
"Vec3",
"Vec4",
"Vec6",
"Vec",
"Mat33",
"Mat66",
"Mat",
):
if numeric is getattr(_bindings, "_Var" + suffix):
bound = getattr(_bindings, "_Var" + quantity_name + suffix, None)
if bound is not None:
return bound
raise TypeError(f"Unsupported quantity Var value type: {value_type!r}")
try:
return specializations[value_type]
except (KeyError, TypeError):
raise TypeError(f"Unsupported Var value type: {value_type!r}") from None
def _resolveUnits(quantity_name: str, specification: object) -> Any:
"""Resolve specification in Karana's registry for quantity_name; return a Unit.
Empty strings use the quantity's configured default units. Explicit units
must match the physical quantity.
"""
expected = getattr(kq, quantity_name)
if specification == "":
return kq.getDefaultUnits(expected)
units = kq.ureg.Unit(str(specification))
if not kq.ureg.Quantity(1, units).check(expected):
raise ValueError(f"Expected {quantity_name} units, got {units}")
return units
def _resolveVarUnits(var: Any) -> Any:
"""Resolve var's declared units once per Python instance; return its Unit."""
if not hasattr(var, "_resolved_units"):
var._resolved_units = _resolveUnits(var._quantity_name, var.units())
return var._resolved_units
def _quantityConstructor(
quantity_alias: str,
fn: Callable[[], object],
args: tuple[object, ...],
kwargs: dict[str, object],
) -> tuple:
"""Normalize quantity Var arguments; return numeric callback, dimensions, description and units.
quantity_alias selects Math metadata. fn may return a raw magnitude or a Pint
quantity. args and kwargs hold numeric Var constructor dimensions and metadata.
"""
from pint import Quantity # pylint: disable=import-outside-toplevel
values = list(args)
dimensions = []
while values and isinstance(values[0], int):
dimensions.append(values.pop(0))
options = dict(kwargs)
for key in ("vector_size", "matrix_rows", "matrix_cols"):
if key in options:
dimensions.append(options.pop(key))
description = values.pop(0) if values else options.pop("description")
specification = values.pop(0) if values else options.pop("units", "")
if values or options:
raise TypeError("Unexpected quantity Var constructor arguments")
# The native class supplies the semantic name
quantity_name = getattr(_bindings, "_Var" + quantity_alias + "Double")._quantity_name
units = _resolveUnits(quantity_name, specification)
def numeric() -> object:
"""Invoke fn, converting quantity results to declared units; return the magnitude."""
value: Any = fn()
if isinstance(value, Quantity):
value = value.to(units).magnitude
# Logging allocates storage from the declared dynamic dimensions. Reject
# a changing callback shape before C++ can copy its numeric data.
if dimensions and np.shape(value) != tuple(dimensions):
raise ValueError(f"Expected magnitude shape {tuple(dimensions)}, got {np.shape(value)}")
return value
# Callback conversion needs a Unit object; the native constructor resolves
# the original specification into the canonical string stored on the Var.
return numeric, dimensions, description, str(specification)
def _quantityTo(self: Any, units: object) -> Any:
"""Return a native Var expressing self's numeric output in the requested units."""
source = _resolveVarUnits(self)
target = _resolveUnits(self._quantity_name, units)
factor = source._REGISTRY.Quantity(1, source).to(target).magnitude
return self._scaled(factor, str(target))
def _quantityCreate(cls: type, *args: object, **kwargs: object) -> object:
"""Construct cls from positional args and keyword kwargs; return the quantity Var."""
return cls(*args, **kwargs)
# Returned C++ objects use these registered classes too. Naming the classes
# gives constructed and returned Vars the same display without wrapping them.
# Stubgen needs valid identifiers for its private concrete declarations, so
# only the display-name changes are skipped while generating stubs.
setattr(_BoundVar, "__class_getitem__", classmethod(_specializeVar))
if not getenv("KARANA_STUBGEN"):
for _varClass, _valueName in {
_bindings._VarBool: "bool",
_bindings._VarDouble: "float",
_bindings._VarFloat: "np.float32",
_bindings._VarShort: "np.int16",
_bindings._VarInt: "int",
_bindings._VarLong: "np.int64",
_bindings._VarSizeT: "np.uint64",
_bindings._VarChar: "Char",
_bindings._VarString: "str",
_bindings._VarVecString: "list[str]",
_bindings._VarVec: "Vec",
_bindings._VarVec3: "Vec3",
_bindings._VarVec4: "Vec4",
_bindings._VarVec6: "Vec6",
_bindings._VarVecInt: "VecInt",
_bindings._VarMat: "Mat",
_bindings._VarMat33: "Mat33",
_bindings._VarMat66: "Mat66",
_bindings._VarUnitQuaternion: "UnitQuaternion",
_bindings._VarSpatialVector: "SpatialVector",
_bindings._VarSpatialVelocity: "SpatialVelocity",
_bindings._VarSpatialAcceleration: "SpatialAcceleration",
_bindings._VarSpatialForce: "SpatialForce",
_bindings._VarHomTran: "HomTran",
}.items():
_genericName = f"Var[{_valueName}]"
_varClass.__name__ = _genericName
_varClass.__qualname__ = _genericName
# Match the qualified class name with an attribute for introspection.
globals()[_genericName] = _varClass
_numericClass = _varClass.__base__
_numericName = f"NumericVar[{_valueName}]"
_numericClass.__name__ = _numericName
_numericClass.__qualname__ = _numericName
globals()[_numericName] = _numericClass
# NumericVar[T] selects the shared native interface, while Var[T] constructs
# the concrete unitless specialization. Static annotations below describe both.
class _NumericVarMeta(type(_BoundVar)):
"""Recognize native Var objects through the NumericVar runtime facade."""
def __instancecheck__(cls, value: object) -> bool:
"""Return whether value is a native Var instance."""
return isinstance(value, _BoundVar)
def __subclasscheck__(cls, value: type) -> bool:
"""Return whether value inherits the common native Var base."""
return issubclass(value, _BoundVar)
class NumericVar(_BoundVar, metaclass=_NumericVarMeta):
"""Runtime facade for native numeric interfaces and quantity-aware inputs."""
[docs]
@classmethod
def __class_getitem__(cls, value_type: object) -> type[_BoundVar]:
"""Return the numeric base for magnitude value_type; reject quantity aliases."""
concrete = _specializeVar(_BoundVar, value_type)
if "_quantity_alias" in vars(concrete):
raise TypeError("NumericVar expects a magnitude type")
return concrete.__base__
for _quantityClass in vars(_bindings).values():
if isinstance(_quantityClass, type) and "_quantity_alias" in vars(_quantityClass):
_quantityClass.to = _quantityTo
_quantityClass.create = classmethod(_quantityCreate)
if not getenv("KARANA_STUBGEN"):
_suffix = _quantityClass._magnitude_name
_magnitudeLabel = {"Double": "float", "Float": "np.float32"}.get(_suffix, _suffix)
_label = f"Var[{_quantityClass._quantity_alias}[{_magnitudeLabel}]]"
_quantityClass.__name__ = _label
_quantityClass.__qualname__ = _label
globals()[_label] = _quantityClass
if TYPE_CHECKING:
from pint.facets.numpy.quantity import NumpyQuantity as _NumpyQuantity
from pint.facets.plain.quantity import PlainQuantity as _PlainQuantity
_ValueT = TypeVar("_ValueT")
_MagnitudeT = TypeVar("_MagnitudeT", bound=float | np.floating | NDArray)
_PlotArrayT = TypeVar("_PlotArrayT", bound=kt.Vec | kt.Vec3)
_PlotSpatialT = TypeVar(
"_PlotSpatialT",
km.SpatialVector,
km.SpatialVelocity,
km.SpatialAcceleration,
km.SpatialForce,
)
class Var(_BoundVar, Generic[_ValueT]):
"""A callable value with an optional compile-time physical quantity."""
@overload
def __init__(
self: "_TypedVar[_NumpyQuantity[_MagnitudeT]]",
name: str,
fn: Callable[[], _MagnitudeT | _PlainQuantity[Any]],
description: str,
units: object = "",
) -> None:
"""Create a new Var.
Parameters
----------
name : str
The name of the Var[T]
fn : Callable[[], T | Quantity]
The callable that produces type T, either as a numeric value or a Quantity.
Quantity results are converted to units before passing their magnitudes to C++.
description : str
A description of the Var.
units : str or pint.Unit, optional
The units of the Var. An empty string selects the configured default units.
"""
...
@overload
def __init__(
self: "_TypedVar[_NumpyQuantity[_MagnitudeT]]",
name: str,
fn: Callable[[], _MagnitudeT | _PlainQuantity[Any]],
vector_size: int,
description: str,
units: object = "",
) -> None:
"""Create a new instance of Var[Vec].
Parameters
----------
name : str
The name of the new Var[Vec]
fn : Callable[[], T | Quantity]
The callable that produces Vec, either as a numeric value or a Quantity.
Quantity results are converted to units before passing their magnitudes to C++.
vector_size : int
The size of the km::Vec that the callable produces. This must
always be the same size.
description : str
A description of the Var.
units : str or pint.Unit, optional
The units of the Var. An empty string selects the configured default units.
"""
...
@overload
def __init__(
self: "_TypedVar[_NumpyQuantity[_MagnitudeT]]",
name: str,
fn: Callable[[], _MagnitudeT | _PlainQuantity[Any]],
matrix_rows: int,
matrix_cols: int,
description: str,
units: object = "",
) -> None:
"""Create a new instance of Var[Mat].
Parameters
----------
name : str
The name of the Var[Mat]
fn : Callable[[], T | Quantity]
The callable that produces Mat, either as a numeric value or a Quantity.
Quantity results are converted to units before passing their magnitudes to C++.
matrix_rows : int
The number of rows in the km::Mat that the callable produces. The
matrix that the callable produces must always be the same size.
matrix_cols : int
The number of cols in the km::Mat that the callable produces. The
matrix that the callable produces must always be the same size.
description : str
A description of the Var.
units : str or pint.Unit, optional
The units of the Var. An empty string selects the configured default units.
"""
...
@overload
def __init__(
self, name: str, fn: Callable[[], _ValueT], description: str, units: object = ""
) -> None:
"""Create a new instance of Var[T].
Parameters
----------
name : str
The name of the Var[T]
fn : Callable[[], T]
The callable that produces type T.
description : str
A description of the Var.
units : str, optional
The units of the Var.
"""
...
@overload
def __init__(
self,
name: str,
fn: Callable[[], _ValueT],
vector_size: int,
description: str,
units: object = "",
) -> None:
"""Create a new instance of Var[Vec].
Parameters
----------
name : str
The name of the new Var[Vec]
fn : Callable[[], T]
The callable that produces a Vec.
vector_size : int
The size of the km::Vec that the callable produces. This must
always be the same size.
description : str
A description of the Var.
units : str, optional
The units of the Var.
"""
...
@overload
def __init__(
self,
name: str,
fn: Callable[[], _ValueT],
matrix_rows: int,
matrix_cols: int,
description: str,
units: object = "",
) -> None:
"""Create a new instance of Var[Mat].
Parameters
----------
name : str
The name of the Var[Mat]
fn : Callable[[], T]
The callable that produces a Mat.
matrix_rows : int
The number of rows in the km::Mat that the callable produces. The
matrix that the callable produces must always be the same size.
matrix_cols : int
The number of cols in the km::Mat that the callable produces. The
matrix that the callable produces must always be the same size.
description : str
A description of the Var.
units : str, optional
The units of the Var.
"""
...
@overload
def __init__(self, name: str, other: Self, description: str) -> None:
"""Rename Var constructor.
This constructor keeps the same function, size, etc., but takes a new
name and description.
Parameters
----------
name : str
The name of the Var[T]
other : Var[T]
The Var[T] to copy the callable and units from.
description : str
A description of the Var.
"""
...
def __init__(self, *args: object, **kwargs: object) -> None:
"""Construct a numeric Var and resolve its units if Python is available.
Parameters
----------
*args : tuple
Name, callable or source Var, dimensions, description and units.
**kwargs : dict
Constructor arguments supplied by name.
"""
...
@overload
def __call__(self: "_TypedVar[np.float32] | _TypedVar[np.float64]") -> float:
"""Return the value that the Var contains.
Returns
-------
float
The value that the Var contains.
"""
...
@overload
def __call__(
self: "_TypedVar[np.int16] | _TypedVar[np.int32] | _TypedVar[np.int64] | _TypedVar[np.uint64]",
) -> int:
"""Return the value that the Var contains.
Returns
-------
int
The value that the Var contains.
"""
...
@overload
def __call__(self) -> _ValueT:
"""Return the value that the Var contains.
Returns
-------
T
The value that the Var contains.
"""
...
def __call__(self) -> object:
"""Return the value that the Var contains.
Returns
-------
object
The value that the Var contains.
"""
...
@overload
def getPlottingFunction(
self: "_TypedVar[float] | _TypedVar[np.float64] | _TypedVar[_NumpyQuantity[float]] | _TypedVar[_NumpyQuantity[np.float64]]",
) -> Callable[[], float]:
"""Return the plotting function associated with this Var.
Returns
-------
Callable[[], float]
The plotting function associated with this Var. For quantity Vars,
the function returns raw magnitudes in the Var's units.
"""
...
@overload
def getPlottingFunction(
self: "_TypedVar[_PlotArrayT] | _TypedVar[_NumpyQuantity[_PlotArrayT]]",
) -> Callable[[], _PlotArrayT]:
"""Return the plotting function associated with this Var.
Returns
-------
Callable[[], T]
The plotting function associated with this Var. For quantity Vars,
the function returns raw magnitudes in the Var's units.
"""
...
@overload
def getPlottingFunction(self: "_TypedVar[_PlotSpatialT]") -> Callable[[], _PlotSpatialT]:
"""Return the plotting function associated with this Var.
Returns
-------
Callable[[], T]
The plotting function associated with this Var.
"""
...
def getPlottingFunction(self) -> Callable[[], object]:
"""Return the plotting function associated with this Var.
Returns
-------
Callable[[], object]
The plotting function associated with this Var.
"""
...
def to(
self: "_TypedVar[_NumpyQuantity[_MagnitudeT]]", units: object
) -> "_TypedVar[_NumpyQuantity[_MagnitudeT]]":
"""Return a Var with this quantity and magnitude type expressed in units.
Parameters
----------
units : str or pint.Unit
Unit specification describing the scaled numeric values.
Returns
-------
Var[T]
A quantity Var with a native scaled callable and logging function.
"""
...
@classmethod
def create(cls, *args: object, **kwargs: object) -> Self:
"""Create a new instance of Var[T].
Parameters
----------
*args : tuple
Name, callable or source Var, dimensions, description and units.
The callable may produce a numeric value or a Quantity for a quantity Var.
**kwargs : dict
Constructor arguments supplied by name.
Returns
-------
Var[T]
The newly created instance of Var[T].
"""
...
# Self annotations distinguish the generic interface from the native base.
_TypedVar: TypeAlias = Var
type NumericVar[T: float | np.floating | NDArray] = _TypedVar[T] | _TypedVar[_NumpyQuantity[T]]
"""A numeric Var or a quantity Var with the same underlying magnitude type."""
# Giving credit to Basilisk. See https://github.com/AVSLab/basilisk/blob/19e5538bed96141634fac4570b5b50abb01836a3/src/utilities/deprecated.py
@overload
def deprecated(name: str, removal_date: datetime.date | str, msg: str):
"""
Print a deprecation message.
Parameters
----------
name : str
Name of the class, function, etc. being deprecated.
removal_date : datetime.date | str
The date at which the deprecated class, function, etc. will be removed.
msg : str
A message telling the user how to fix their deprecated code.
"""
... # pragma: no cover. Will not be run by code coverage.
@overload
def deprecated(removal_date: datetime.date | str, msg: str):
"""Use this decorator is used to mark functions or classes as deprecated.
Usage::
@deprecated.deprecated("2024-05-28", "Use MyNewClass")
class MyClass:
...
or::
@deprecated.deprecated("2024-05-28", "myFun is unsafe now")
def myFun(a, b):
...
or::
class ValidClass:
@deprecated.deprecated("2024-05-28", "myMethod is slow, use myBetterMethod")
def myMethod(self, a, b):
...
Parameters
----------
removal_date : datetime.date | str
The date at which the deprecated class, function, etc. will be removed.
msg : str
A message telling the user how to fix their deprecated code.
"""
... # pragma: no cover. Will not be run by code coverage.
def deprecated(*args, **_):
"""Use this decorator is used to mark functions or classes as deprecated.
See overloaded function signatures for details.
"""
if len(args) == 2:
# Decorator
def wrapper(func):
name = f"{func.__module__}.{func.__qualname__}"
removal_date = args[0]
if isinstance(removal_date, str):
removal_date = datetime.datetime.strptime(removal_date, r"%Y-%m-%d").date()
msg = args[1]
def innerWrapper(*args, **kwargs):
deprecated(name, removal_date, msg)
return func(*args, **kwargs)
return innerWrapper
return wrapper
elif len(args) == 3:
name = args[0].replace(")", "").replace("(", "")
removal_date = args[1]
if isinstance(removal_date, str):
removal_date = datetime.datetime.strptime(removal_date, r"%Y-%m-%d").date()
stack = "".join(format_stack()[:-1])
_deprecated(name, str(removal_date), args[2] + "\nStack follows:\n" + stack)
else:
raise ValueError("Got an invalid number of arguments.")
T = TypeVar("T", bound=Base)
class CppWeakRef(CppWeakRefBase, Generic[T]):
"""Weak reference to C++ object.
Use the get() method to return a Python object from the C++ object
if it exists.
"""
def __init__(self, base: T):
"""Create a CppWeakRef.
Parameters
----------
base : T
Object to create a CppWeakRef to.
"""
super().__init__(base)
[docs]
def __call__(self) -> T | None:
"""Return a Python object for the C++ class if it exists, or None if it does not.
Returns
-------
T | None
The Python object if the C++ object still exists. None if it does not.
"""
return cast(T, super().__call__())
