""" core.resource ============= Generic resource representation for WorldSmithAI. Resources are domain-agnostic quantities that agents and behaviors may observe, consume, regenerate, exchange, transform, or produce. Examples: - 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. """ 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] | None ResourceOperation: TypeAlias = Literal["deplete", "regenerate", "set_amount"] @dataclass(frozen=True, slots=True) class ResourceOperationResult: """Structured result of a resource operation.""" 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.""" 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.""" 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. Runtime compatibility: - ``position`` is always normalized to a NumPy array. - ``regeneration_rate`` and ``max_amount`` are accepted directly because WorldFactory may pass or attach them. - ``max_amount`` is mirrored into metadata as both ``max_amount`` and ``capacity`` for compatibility with older code. """ id: str type: str amount: float position: PositionInput = None metadata: dict[str, Any] = field(default_factory=dict) regeneration_rate: float = 0.0 max_amount: float | None = None dsl_spec: Any | None = None def __post_init__(self) -> None: """Validate and normalize resource fields.""" 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.") if not isinstance(self.metadata, Mapping): raise TypeError("Resource.metadata must be a mapping.") 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) self.metadata = dict(self.metadata) self._validate_string_keys(self.metadata, "metadata") self.regeneration_rate = self._resolve_regeneration_rate(self.regeneration_rate) self.max_amount = self._resolve_max_amount(self.max_amount) self.metadata.setdefault("regeneration_rate", self.regeneration_rate) if self.max_amount is not None: self.metadata.setdefault("max_amount", self.max_amount) self.metadata.setdefault("capacity", self.max_amount) if self.capacity is not None and self.amount > self.capacity: logger.warning( "Resource amount exceeds capacity: resource_id=%s amount=%s capacity=%s", self.id, self.amount, self.capacity, ) @property def capacity(self) -> float | None: """Optional maximum amount for this resource.""" if self.max_amount is not None: return self.max_amount for key in ("capacity", "max_amount"): if key in self.metadata and self.metadata[key] is not None: return self._normalize_non_negative_float( self.metadata[key], f"metadata['{key}']", ) return None @property def is_empty(self) -> bool: """Return whether this resource has no available quantity.""" 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.""" 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.""" 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.""" 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.""" self.position = self._normalize_position(position) def distance_to(self, position: PositionInput) -> float: """Compute Euclidean distance from this resource to a position.""" 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.""" return { "id": self.id, "type": self.type, "amount": self.amount, "position": self.position.tolist(), "regeneration_rate": self.regeneration_rate, "max_amount": self.max_amount, "metadata": deepcopy(self.metadata), } def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly dictionary representation.""" return self.snapshot() def copy(self) -> Resource: """Return a deep copy of this resource.""" return Resource( id=self.id, type=self.type, amount=self.amount, position=self.position.copy(), metadata=deepcopy(self.metadata), regeneration_rate=self.regeneration_rate, max_amount=self.max_amount, dsl_spec=self.dsl_spec, ) 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.""" 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 resource-defined capacity.""" if capacity is not None: return self._normalize_non_negative_float(capacity, "capacity") return self.capacity def _resolve_regeneration_rate(self, value: Any) -> float: """Resolve regeneration rate from explicit value or metadata.""" if value is not None and float(value) != 0.0: return self._normalize_non_negative_float(value, "regeneration_rate") if "regeneration_rate" in self.metadata and self.metadata["regeneration_rate"] is not None: return self._normalize_non_negative_float( self.metadata["regeneration_rate"], "metadata['regeneration_rate']", ) return self._normalize_non_negative_float(value or 0.0, "regeneration_rate") def _resolve_max_amount(self, value: Any) -> float | None: """Resolve max amount from explicit value or metadata aliases.""" if value is not None: return self._normalize_non_negative_float(value, "max_amount") for key in ("max_amount", "capacity"): if key in self.metadata and self.metadata[key] is not None: return self._normalize_non_negative_float( self.metadata[key], f"metadata['{key}']", ) return None @staticmethod def _normalize_position(position: PositionInput) -> NDArray[np.float64]: """Normalize a position input into a one-dimensional numpy float array.""" if position is None: return np.zeros(2, dtype=np.float64) normalized = np.asarray(position, dtype=np.float64).reshape(-1) 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.astype(np.float64, copy=True) @staticmethod def _normalize_non_negative_float(value: Any, label: str) -> float: """Normalize a value into a finite non-negative float.""" 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.""" for key in mapping: if not isinstance(key, str): raise TypeError(f"Resource.{label} keys must be strings.") __all__ = [ "PositionInput", "Resource", "ResourceOperation", "ResourceOperationResult", ]