Spaces:
Sleeping
Sleeping
File size: 4,076 Bytes
b7a5bab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """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)
|