| """Attract / match scoring (R4, R5) — pure functions.""" |
| from __future__ import annotations |
|
|
| from typing import Dict, Iterable, Mapping, Optional, Sequence, Set |
|
|
| def clamp(x: float, lo: float = 0.0, hi: float = 1.0) -> float: |
| return max(lo, min(hi, x)) |
|
|
| def l1_normalize(vec: Mapping[str, float]) -> Dict[str, float]: |
| cleaned = {k: max(0.0, float(v)) for k, v in vec.items()} |
| s = sum(cleaned.values()) |
| if s <= 0: |
| return {k: 0.0 for k in cleaned} |
| return {k: v / s for k, v in cleaned.items()} |
|
|
| def preference_match( |
| preferred_atmosphere: Mapping[str, float], |
| shop_atmosphere_snapshot: Mapping[str, float], |
| ) -> float: |
| """dot(L1_norm(pref), shop_snapshot) clamped — shop dims NOT renormalized.""" |
| pref = l1_normalize(preferred_atmosphere) |
| keys = set(pref) | set(shop_atmosphere_snapshot) |
| total = 0.0 |
| for k in keys: |
| total += pref.get(k, 0.0) * float(shop_atmosphere_snapshot.get(k, 0.0)) |
| return clamp(total) |
|
|
| def shop_product_match(desired_tags: Sequence[str], displayed_product_tags: Iterable[str]) -> float: |
| desired = set(desired_tags) |
| if not desired: |
| return 0.0 |
| disp = set(displayed_product_tags) |
| return len(desired & disp) / max(1, len(desired)) |
|
|
| def item_match(desired_tags: Sequence[str], product_tags: Sequence[str]) -> float: |
| desired = set(desired_tags) |
| if not desired: |
| return 0.0 |
| return len(desired & set(product_tags)) / max(1, len(desired)) |
|
|
| def layout_comfort(has_free_browse_spot: bool, current_customers: int, max_customers: int = 5) -> float: |
| free = 1.0 if has_free_browse_spot else 0.3 |
| crowding = (current_customers / max(1, max_customers)) if max_customers else 0.0 |
| return clamp(0.5 * free + 0.5 * (1.0 - crowding)) |
|
|
| def relationship_pull(stage: str, table: Optional[Mapping[str, float]] = None) -> float: |
| default = { |
| "STRANGER": 0.0, |
| "ACQUAINTANCE": 0.25, |
| "FAMILIAR": 0.45, |
| "TRUSTED": 0.65, |
| "BOND": 0.85, |
| } |
| t = table or default |
| return float(t.get(stage, 0.0)) |
|
|
| def compute_attract( |
| *, |
| preferred_atmosphere: Mapping[str, float], |
| shop_atmosphere: Mapping[str, float], |
| desired_tags: Sequence[str], |
| displayed_tags: Iterable[str], |
| has_free_browse_spot: bool, |
| current_customers: int, |
| stage: str = "STRANGER", |
| weights: Optional[Mapping[str, float]] = None, |
| rel_table: Optional[Mapping[str, float]] = None, |
| max_customers: int = 5, |
| weather_time_mod: float = 0.0, |
| ) -> float: |
| w = { |
| "w_atm": 0.35, |
| "w_prod": 0.30, |
| "w_layout": 0.15, |
| "w_rel": 0.15, |
| "w_weather": 0.0, |
| } |
| if weights: |
| w.update({k: float(v) for k, v in weights.items()}) |
| pref = preference_match(preferred_atmosphere, shop_atmosphere) |
| prod = shop_product_match(desired_tags, displayed_tags) |
| lay = layout_comfort(has_free_browse_spot, current_customers, max_customers) |
| rel = relationship_pull(stage, rel_table) |
| return ( |
| w["w_atm"] * pref |
| + w["w_prod"] * prod |
| + w["w_layout"] * lay |
| + w["w_rel"] * rel |
| + w["w_weather"] * weather_time_mod |
| ) |
|
|
| def displayed_tags_from_state(state: dict, products: Mapping[str, dict]) -> Set[str]: |
| tags: Set[str] = set() |
| for slot in state.get("shop", {}).get("displays", {}).values(): |
| if not slot: |
| continue |
| pid = slot.get("product_id") |
| qty = int(slot.get("qty") or 0) |
| if not pid or qty <= 0: |
| continue |
| p = products.get(pid) or {} |
| tags.update(p.get("tags") or []) |
| return tags |
|
|
