""" core.resource ============= Generic resource representation for WorldSmithAI. This module defines the Resource runtime object used by the simulation engine. Resources are domain-agnostic quantities that agents and behaviors may observe, consume, regenerate, exchange, transform, or produce. Examples of resources include: - food - grass - water - money - knowledge - mana - ore - influence - oxygen - compute The Resource class does not know what any resource type means. Domain semantics are supplied by the DSL, behaviors, policies, and world rules. Minimal usage example --------------------- resource = Resource( id="food_patch_1", type="food", amount=10.0, position=[2.0, 3.0], metadata={ "capacity": 20.0, "regeneration_rate": 1.5, }, ) depletion = resource.deplete(3.0) print(depletion.new_amount) regeneration = resource.regenerate() print(regeneration.new_amount) """ from __future__ import annotations import logging from copy import deepcopy from dataclasses import dataclass, field from numbers import Real from typing import Any, Literal, Mapping, Sequence, TypeAlias import numpy as np from numpy.typing import NDArray logger = logging.getLogger(__name__) PositionInput: TypeAlias = Sequence[float] | NDArray[np.float64] ResourceOperation: TypeAlias = Literal["deplete", "regenerate", "set_amount"] @dataclass(frozen=True, slots=True) class ResourceOperationResult: """ Structured result of a resource operation. Attributes ---------- resource_id: Unique id of the affected resource. resource_type: Domain-level resource type string. operation: Operation name, such as ``"deplete"`` or ``"regenerate"``. requested_amount: Amount requested by the caller. actual_delta: Signed change applied to the resource amount. Regeneration produces a positive delta. Depletion produces a negative delta. previous_amount: Resource amount before the operation. new_amount: Resource amount after the operation. success: Whether the operation completed successfully. partial: Whether the operation applied less than the requested amount. message: Human-readable operation summary. metadata: Additional structured operation metadata. """ resource_id: str resource_type: str operation: ResourceOperation requested_amount: float actual_delta: float previous_amount: float new_amount: float success: bool partial: bool = False message: str = "" metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate operation result fields. Raises ------ ValueError If any numeric field is non-finite or invalid. TypeError If metadata is not a mapping. """ numeric_fields = { "requested_amount": self.requested_amount, "actual_delta": self.actual_delta, "previous_amount": self.previous_amount, "new_amount": self.new_amount, } for name, value in numeric_fields.items(): if not np.isfinite(value): raise ValueError(f"{name} must be finite.") if self.requested_amount < 0.0: raise ValueError("requested_amount cannot be negative.") if self.previous_amount < 0.0: raise ValueError("previous_amount cannot be negative.") if self.new_amount < 0.0: raise ValueError("new_amount cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("metadata must be a mapping.") object.__setattr__(self, "requested_amount", float(self.requested_amount)) object.__setattr__(self, "actual_delta", float(self.actual_delta)) object.__setattr__(self, "previous_amount", float(self.previous_amount)) object.__setattr__(self, "new_amount", float(self.new_amount)) object.__setattr__(self, "metadata", dict(self.metadata)) def to_dict(self) -> dict[str, Any]: """ Convert the operation result into a JSON-friendly dictionary. Returns ------- dict[str, Any] Serializable operation result. """ return { "resource_id": self.resource_id, "resource_type": self.resource_type, "operation": self.operation, "requested_amount": self.requested_amount, "actual_delta": self.actual_delta, "previous_amount": self.previous_amount, "new_amount": self.new_amount, "success": self.success, "partial": self.partial, "message": self.message, "metadata": dict(self.metadata), } @dataclass(slots=True) class Resource: """ Generic domain-agnostic world resource. A resource stores a finite non-negative quantity at a numeric position. Its semantic meaning is entirely defined by the DSL and behaviors. Attributes ---------- id: Unique resource identifier. type: Domain-level resource type string, such as ``"food"``, ``"money"``, ``"knowledge"``, or ``"mana"``. amount: Current non-negative resource quantity. position: Numeric position vector. The engine does not assume 2D only. metadata: Optional structured metadata. Common optional keys may include ``"capacity"``, ``"regeneration_rate"``, ``"owner_id"``, ``"tags"``, or domain-specific configuration. """ id: str type: str amount: float position: PositionInput metadata: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize resource fields. Raises ------ ValueError If id, type, amount, or position are invalid. TypeError If metadata has invalid type or non-string keys. """ if not isinstance(self.id, str) or not self.id.strip(): raise ValueError("Resource.id must be a non-empty string.") if not isinstance(self.type, str) or not self.type.strip(): raise ValueError("Resource.type must be a non-empty string.") self.id = self.id.strip() self.type = self.type.strip() self.amount = self._normalize_non_negative_float(self.amount, "amount") self.position = self._normalize_position(self.position) if not isinstance(self.metadata, Mapping): raise TypeError("Resource.metadata must be a mapping.") self.metadata = dict(self.metadata) self._validate_string_keys(self.metadata, "metadata") capacity = self.capacity if capacity is not None and self.amount > capacity: logger.warning( "Resource amount exceeds metadata capacity: resource_id=%s " "amount=%s capacity=%s", self.id, self.amount, capacity, ) @property def capacity(self) -> float | None: """ Optional maximum amount for this resource. The value is read from ``metadata["capacity"]`` when present. Returns ------- float | None Non-negative finite capacity, or None if no capacity is configured. """ return self._read_optional_metadata_float("capacity") @property def regeneration_rate(self) -> float: """ Optional default regeneration amount per resource update. The value is read from ``metadata["regeneration_rate"]`` when present. Missing values default to ``0.0``. Returns ------- float Non-negative finite regeneration amount. """ rate = self._read_optional_metadata_float("regeneration_rate") return 0.0 if rate is None else rate @property def is_empty(self) -> bool: """ Return whether this resource has no available quantity. Returns ------- bool True if amount is zero, otherwise False. """ return self.amount <= 0.0 def regenerate( self, amount: float | None = None, *, capacity: float | None = None, metadata: Mapping[str, Any] | None = None, ) -> ResourceOperationResult: """ Increase the resource amount. If ``amount`` is omitted, the method uses ``metadata["regeneration_rate"]`` when available, otherwise ``0.0``. If ``capacity`` is omitted, the method uses ``metadata["capacity"]`` when available. If no capacity exists, the resource may grow without an upper bound. Parameters ---------- amount: Non-negative amount to regenerate. Defaults to the configured regeneration rate. capacity: Optional non-negative maximum resource amount for this operation. metadata: Additional metadata attached to the operation result. Returns ------- ResourceOperationResult Structured result of the regeneration operation. """ requested_amount = ( self.regeneration_rate if amount is None else self._normalize_non_negative_float(amount, "amount") ) resolved_capacity = self._resolve_capacity(capacity) previous_amount = self.amount if requested_amount == 0.0: return self._operation_result( operation="regenerate", requested_amount=requested_amount, previous_amount=previous_amount, new_amount=previous_amount, success=True, partial=False, message="No regeneration requested.", metadata=metadata, ) if resolved_capacity is None: applied_amount = requested_amount else: available_capacity = max(0.0, resolved_capacity - previous_amount) applied_amount = min(requested_amount, available_capacity) new_amount = previous_amount + applied_amount self.amount = new_amount partial = applied_amount < requested_amount success = applied_amount > 0.0 message = ( "Resource regenerated." if success and not partial else "Resource regenerated partially." if success and partial else "Resource could not regenerate because capacity was reached." ) return self._operation_result( operation="regenerate", requested_amount=requested_amount, previous_amount=previous_amount, new_amount=new_amount, success=success, partial=partial, message=message, metadata={ **dict(metadata or {}), "capacity": resolved_capacity, }, ) def deplete( self, amount: float, *, allow_partial: bool = True, metadata: Mapping[str, Any] | None = None, ) -> ResourceOperationResult: """ Decrease the resource amount. Parameters ---------- amount: Non-negative amount requested for depletion. allow_partial: Whether to allow depleting the remaining available amount when the requested amount exceeds the current amount. metadata: Additional metadata attached to the operation result. Returns ------- ResourceOperationResult Structured result of the depletion operation. Raises ------ TypeError If ``allow_partial`` is not a boolean. """ requested_amount = self._normalize_non_negative_float(amount, "amount") if not isinstance(allow_partial, bool): raise TypeError("allow_partial must be a boolean.") previous_amount = self.amount if requested_amount == 0.0: return self._operation_result( operation="deplete", requested_amount=requested_amount, previous_amount=previous_amount, new_amount=previous_amount, success=True, partial=False, message="No depletion requested.", metadata=metadata, ) if previous_amount >= requested_amount: applied_amount = requested_amount elif allow_partial and previous_amount > 0.0: applied_amount = previous_amount else: applied_amount = 0.0 new_amount = max(0.0, previous_amount - applied_amount) self.amount = new_amount partial = 0.0 < applied_amount < requested_amount success = applied_amount > 0.0 and (allow_partial or applied_amount == requested_amount) if applied_amount == requested_amount: message = "Resource depleted." elif partial: message = "Resource depleted partially." else: message = "Resource could not be depleted." return self._operation_result( operation="deplete", requested_amount=requested_amount, previous_amount=previous_amount, new_amount=new_amount, success=success, partial=partial, message=message, metadata={ **dict(metadata or {}), "allow_partial": allow_partial, }, ) def set_amount( self, amount: float, *, metadata: Mapping[str, Any] | None = None, ) -> ResourceOperationResult: """ Set the resource amount directly. This method is useful for world events, scripted interventions, tests, and world factory initialization corrections. Behaviors should usually prefer ``deplete()`` or ``regenerate()`` so operation semantics remain explicit. Parameters ---------- amount: New non-negative finite resource amount. metadata: Additional metadata attached to the operation result. Returns ------- ResourceOperationResult Structured result of the set operation. """ normalized_amount = self._normalize_non_negative_float(amount, "amount") previous_amount = self.amount self.amount = normalized_amount return self._operation_result( operation="set_amount", requested_amount=normalized_amount, previous_amount=previous_amount, new_amount=normalized_amount, success=True, partial=False, message="Resource amount set.", metadata=metadata, ) def set_position(self, position: PositionInput) -> None: """ Set the resource position. Parameters ---------- position: New numeric position vector. Raises ------ ValueError If the position is invalid. """ self.position = self._normalize_position(position) def distance_to(self, position: PositionInput) -> float: """ Compute Euclidean distance from this resource to a position. Parameters ---------- position: Numeric position vector. Returns ------- float Euclidean distance. Raises ------ ValueError If dimensionality does not match. """ other_position = self._normalize_position(position) if self.position.shape != other_position.shape: raise ValueError( "Position dimensionality mismatch. " f"Resource position has shape {self.position.shape}; " f"target position has shape {other_position.shape}." ) return float(np.linalg.norm(self.position - other_position)) def snapshot(self) -> dict[str, Any]: """ Return a JSON-friendly snapshot of this resource. Returns ------- dict[str, Any] Serializable resource snapshot. """ return { "id": self.id, "type": self.type, "amount": self.amount, "position": self.position.tolist(), "metadata": deepcopy(self.metadata), } def copy(self) -> Resource: """ Return a deep copy of this resource. Returns ------- Resource Independent copy of this resource. """ return Resource( id=self.id, type=self.type, amount=self.amount, position=self.position.copy(), metadata=deepcopy(self.metadata), ) def _operation_result( self, *, operation: ResourceOperation, requested_amount: float, previous_amount: float, new_amount: float, success: bool, partial: bool, message: str, metadata: Mapping[str, Any] | None, ) -> ResourceOperationResult: """ Build a normalized resource operation result. Parameters ---------- operation: Operation name. requested_amount: Requested non-negative amount. previous_amount: Amount before the operation. new_amount: Amount after the operation. success: Whether the operation succeeded. partial: Whether less than the requested amount was applied. message: Human-readable operation summary. metadata: Additional result metadata. Returns ------- ResourceOperationResult Structured operation result. """ if metadata is not None and not isinstance(metadata, Mapping): raise TypeError("metadata must be a mapping when provided.") return ResourceOperationResult( resource_id=self.id, resource_type=self.type, operation=operation, requested_amount=requested_amount, actual_delta=new_amount - previous_amount, previous_amount=previous_amount, new_amount=new_amount, success=success, partial=partial, message=message, metadata=dict(metadata or {}), ) def _resolve_capacity(self, capacity: float | None) -> float | None: """ Resolve operation-specific or metadata-defined capacity. Parameters ---------- capacity: Explicit capacity override. Returns ------- float | None Non-negative finite capacity, or None. """ if capacity is not None: return self._normalize_non_negative_float(capacity, "capacity") return self.capacity def _read_optional_metadata_float(self, key: str) -> float | None: """ Read an optional non-negative float from resource metadata. Parameters ---------- key: Metadata key. Returns ------- float | None Normalized non-negative float, or None. Raises ------ TypeError If key is invalid. ValueError If the metadata value is not a finite non-negative number. """ if not isinstance(key, str) or not key: raise TypeError("metadata key must be a non-empty string.") if key not in self.metadata or self.metadata[key] is None: return None return self._normalize_non_negative_float( self.metadata[key], f"metadata['{key}']", ) @staticmethod def _normalize_position(position: PositionInput) -> NDArray[np.float64]: """ Normalize a position input into a one-dimensional numpy float array. Parameters ---------- position: Sequence or numpy array representing a position. Returns ------- NDArray[np.float64] Normalized position vector. Raises ------ ValueError If the position is empty, non-finite, or not one-dimensional. """ normalized = np.asarray(position, dtype=np.float64) if normalized.ndim != 1: raise ValueError("Resource.position must be a one-dimensional vector.") if normalized.size == 0: raise ValueError("Resource.position cannot be empty.") if not np.all(np.isfinite(normalized)): raise ValueError("Resource.position must contain only finite values.") return normalized @staticmethod def _normalize_non_negative_float(value: Any, label: str) -> float: """ Normalize a value into a finite non-negative float. Parameters ---------- value: Candidate numeric value. label: Human-readable value label used in error messages. Returns ------- float Normalized non-negative float. Raises ------ TypeError If value is not numeric. ValueError If value is negative or non-finite. """ if isinstance(value, bool) or not isinstance(value, Real): raise TypeError(f"{label} must be a real number.") normalized = float(value) if not np.isfinite(normalized): raise ValueError(f"{label} must be finite.") if normalized < 0.0: raise ValueError(f"{label} cannot be negative.") return normalized @staticmethod def _validate_string_keys(mapping: Mapping[str, Any], label: str) -> None: """ Validate that all mapping keys are strings. Parameters ---------- mapping: Mapping to validate. label: Human-readable mapping label used in error messages. Raises ------ TypeError If any mapping key is not a string. """ for key in mapping: if not isinstance(key, str): raise TypeError(f"Resource.{label} keys must be strings.") __all__ = [ "PositionInput", "Resource", "ResourceOperation", "ResourceOperationResult", ]