Spaces:
Sleeping
Sleeping
| """Data models for the Store OpenEnv simulation.""" | |
| from __future__ import annotations | |
| from typing import Any | |
| from pydantic import BaseModel, Field | |
| try: | |
| from ._compat import Action, Observation | |
| except ImportError: # pragma: no cover - supports direct module imports | |
| from _compat import Action, Observation | |
| class OrderRequest(BaseModel): | |
| """One supplier order request.""" | |
| product: str = Field(..., description="Product name from the market catalog") | |
| seller: str = Field(..., description="Seller name, for example Seller A") | |
| quantity: int = Field(..., gt=0, description="Number of units to order") | |
| class PriceUpdate(BaseModel): | |
| """One price update for an in-stock product.""" | |
| product: str = Field(..., description="Product name currently stocked by the store") | |
| price: int = Field(..., gt=0, description="New selling price in INR") | |
| class StockItem(BaseModel): | |
| """Current in-store inventory row.""" | |
| product_name: str | |
| unit: str | |
| buying_price_inr: float | |
| selling_price_inr: float | |
| demand_score: float | |
| quantity_on_hand: float | |
| stock_value_inr: float | |
| class PendingOrder(BaseModel): | |
| """Supplier order still in transit.""" | |
| order_id: str | |
| product_name: str | |
| quantity: int | |
| seller_name: str | |
| buy_price_inr: float | |
| days_remaining: int | |
| status: str | |
| created_day: int | |
| class MarketOffer(BaseModel): | |
| """Market catalog row with seller quotes.""" | |
| product_name: str | |
| category: str | |
| unit: str | |
| demand_score: float | |
| seller_a_price_inr: int | |
| seller_a_etd_days: int | |
| seller_b_price_inr: int | |
| seller_b_etd_days: int | |
| seller_c_price_inr: int | |
| seller_c_etd_days: int | |
| seller_d_price_inr: int | |
| seller_d_etd_days: int | |
| class SellerProfile(BaseModel): | |
| """Seller reliability profile.""" | |
| seller_name: str | |
| timeline_behavior: str | |
| reliability_note: str | |
| default_etd_days: int | |
| class MarketingAction(BaseModel): | |
| """Durable operational action the agent can activate.""" | |
| action_id: int | |
| description: str | |
| customer_satisfaction_delta: int | |
| cost_inr: int | |
| recurring_cost_inr: int | |
| order_processing_capacity_delta: int | |
| is_active: bool = False | |
| class StoreSnapshot(BaseModel): | |
| """Top-level store state shown to the agent.""" | |
| day: int | |
| current_daily_expense_inr: float | |
| bank_balance_inr: float | |
| customer_satisfaction: float | |
| order_processing_capacity: float | |
| cumulative_profit_inr: float | |
| active_action_ids: list[int] = Field(default_factory=list) | |
| active_recurring_cost_inr: float = 0.0 | |
| episode_horizon_days: int | |
| commit_window_days: int | |
| reward_horizon_days: int | |
| class StoreAction(Action): | |
| """One bundled decision window for the environment.""" | |
| orders: list[OrderRequest] = Field( | |
| default_factory=list, | |
| description="Supplier orders to place at the start of the decision window", | |
| ) | |
| prices: list[PriceUpdate] | None = Field( | |
| default=None, | |
| description="Optional selling-price updates for products already in stock", | |
| ) | |
| action_ids: list[int] = Field( | |
| default_factory=list, | |
| description="Marketing or operations action IDs to activate", | |
| ) | |
| class AvailableActionSpec(BaseModel): | |
| """Agent-facing action guide included in every observation.""" | |
| name: str | |
| description: str | |
| pass_action: StoreAction | |
| example_action: StoreAction | |
| notes: list[str] = Field(default_factory=list) | |
| class StoreObservation(Observation): | |
| """Agent-facing observation for the store simulation.""" | |
| store: StoreSnapshot | |
| stock: list[StockItem] | |
| pending_orders: list[PendingOrder] | |
| market: list[MarketOffer] | |
| seller_profiles: list[SellerProfile] | |
| marketing_actions: list[MarketingAction] | |
| available_action: AvailableActionSpec | |
| summary: str = "" | |
| def dump_model(model: BaseModel, **kwargs: Any) -> dict[str, Any]: | |
| """Handle Pydantic v1/v2 model serialization.""" | |
| if hasattr(model, "model_dump"): | |
| return model.model_dump(**kwargs) | |
| return model.dict(**kwargs) | |