Trackers API Reference

This section provides technical documentation for the various trackers used to monitor game progress.

Achievement Tracker

trackers.achievement_tracker

Achievement Tracker for Pokemon LLM Agent.

Tracks game achievements/milestones and triggers tweet generation with customized image prompts for each achievement type.

Only triggers once per achievement type per run.

Classes

Achievement dataclass

A triggered achievement.

Source code in trackers/achievement_tracker.py

@dataclass
class Achievement:
    """A triggered achievement."""

    achievement_type: AchievementType
    triggered_at: str  # ISO timestamp
    context: Dict[str, Any] = field(default_factory=dict)  # Pokemon name, badge, etc.
    tweet_posted: bool = False
    tweet_url: Optional[str] = None

AchievementImagePrompt dataclass

Image generation prompts for an achievement.

Prompts are composed from 4 structured fields that are combined at runtime: - setting: background, location, camera angle - mood_expression: facial expression, emotional state - action: pose, what she's doing - extras: lighting, effects, additional elements

The positive_prompt property combines these fields into a single string.

Source code in trackers/achievement_tracker.py

@dataclass
class AchievementImagePrompt:
    """
    Image generation prompts for an achievement.

    Prompts are composed from 4 structured fields that are combined at runtime:
    - setting: background, location, camera angle
    - mood_expression: facial expression, emotional state
    - action: pose, what she's doing
    - extras: lighting, effects, additional elements

    The positive_prompt property combines these fields into a single string.
    """

    # Structured prompt components
    setting: str = ""  # background, location, camera angle
    mood_expression: str = ""  # facial expression, emotional state
    action: str = ""  # pose, what she's doing
    extras: str = ""  # lighting, effects, additional elements

    # Compose positive prompt from structured fields
    @property
    def positive_prompt(self) -> str:
        """Combine structured fields into a single prompt string."""
        parts = [
            self.setting,
            self.mood_expression,
            self.action,
            self.extras,
        ]
        # Filter empty parts and join with commas
        return ", ".join(p.strip() for p in parts if p and p.strip())

Attributes

positive_prompt property

Combine structured fields into a single prompt string.

AchievementTracker

Tracks game achievements and triggers tweet generation.

Persists state to disk to survive restarts. Only triggers each achievement type once per run.

Source code in trackers/achievement_tracker.py

class AchievementTracker:
    """
    Tracks game achievements and triggers tweet generation.

    Persists state to disk to survive restarts.
    Only triggers each achievement type once per run.
    """

    def __init__(
        self,
        storage_path: str = "data/achievements.json",
        reset_on_start: bool = False,
    ):
        """
        Initialize the achievement tracker.

        Args:
            storage_path: Path to persist achievement state
            reset_on_start: If True, clear all achievements on initialization
        """
        self.storage_path = storage_path
        self.triggered: Dict[
            str, Achievement
        ] = {}  # achievement_type.value -> Achievement
        self._pending_achievement: Optional[Achievement] = None

        # Track previous state for change detection
        self._prev_party: List[Dict] = []
        self._prev_badges: List[str] = []
        self._starter_species: Optional[str] = None  # Track which starter was chosen
        self._first_pokemon_count: int = 0  # Track if first catch already happened

        # Ensure data directory exists
        Path(os.path.dirname(storage_path)).mkdir(parents=True, exist_ok=True)

        if reset_on_start:
            self._save()
            log.info("Achievement tracker: Fresh start (all achievements reset)")
        else:
            self._load()
            log.info(f"Achievement tracker: Loaded {len(self.triggered)} achievements")

    def _load(self) -> None:
        """Load achievements from disk."""
        if not os.path.exists(self.storage_path):
            return

        try:
            with open(self.storage_path, "r") as f:
                data = json.load(f)

            for key, value in data.get("triggered", {}).items():
                try:
                    achievement_type = AchievementType(value["achievement_type"])
                    self.triggered[key] = Achievement(
                        achievement_type=achievement_type,
                        triggered_at=value["triggered_at"],
                        context=value.get("context", {}),
                        tweet_posted=value.get("tweet_posted", False),
                        tweet_url=value.get("tweet_url"),
                    )
                except (KeyError, ValueError) as e:
                    log.warning(f"Failed to load achievement {key}: {e}")

            # Restore tracking state
            self._starter_species = data.get("starter_species")
            self._first_pokemon_count = data.get("first_pokemon_count", 0)

        except Exception as e:
            log.error(f"Failed to load achievements: {e}")

    def _save(self) -> None:
        """Save achievements to disk."""
        try:
            data = {
                "triggered": {
                    key: {
                        "achievement_type": ach.achievement_type.value,
                        "triggered_at": ach.triggered_at,
                        "context": ach.context,
                        "tweet_posted": ach.tweet_posted,
                        "tweet_url": ach.tweet_url,
                    }
                    for key, ach in self.triggered.items()
                },
                "starter_species": self._starter_species,
                "first_pokemon_count": self._first_pokemon_count,
            }

            with open(self.storage_path, "w") as f:
                json.dump(data, f, indent=2)

        except Exception as e:
            log.error(f"Failed to save achievements: {e}")

    def is_triggered(self, achievement_type: AchievementType) -> bool:
        """Check if an achievement has already been triggered."""
        return achievement_type.value in self.triggered

    def trigger(
        self,
        achievement_type: AchievementType,
        context: Optional[Dict[str, Any]] = None,
    ) -> Optional[Achievement]:
        """
        Trigger an achievement if not already triggered.

        Args:
            achievement_type: The achievement to trigger
            context: Additional context (Pokemon name, badge, etc.)

        Returns:
            The Achievement if newly triggered, None if already triggered
        """
        if self.is_triggered(achievement_type):
            log.debug(f"Achievement already triggered: {achievement_type.value}")
            return None

        achievement = Achievement(
            achievement_type=achievement_type,
            triggered_at=datetime.now().isoformat(),
            context=context or {},
        )

        self.triggered[achievement_type.value] = achievement
        self._pending_achievement = achievement
        self._save()

        log.info(f"Achievement triggered: {achievement_type.value}")
        return achievement

    def get_pending_achievement(self) -> Optional[Achievement]:
        """Get the pending achievement for tweet generation."""
        return self._pending_achievement

    def clear_pending_achievement(self) -> None:
        """Clear the pending achievement after tweet generation."""
        self._pending_achievement = None

    def mark_tweet_posted(
        self, achievement_type: AchievementType, tweet_url: Optional[str] = None
    ) -> None:
        """Mark an achievement's tweet as posted."""
        if achievement_type.value in self.triggered:
            self.triggered[achievement_type.value].tweet_posted = True
            self.triggered[achievement_type.value].tweet_url = tweet_url
            self._save()

    def get_image_prompt(
        self, achievement_type: AchievementType
    ) -> AchievementImagePrompt:
        """Get the image prompt for an achievement type."""
        return ACHIEVEMENT_IMAGE_PROMPTS.get(
            achievement_type,
            AchievementImagePrompt(),  # Empty prompt as fallback
        )

    # =========================================================================
    # Achievement Detection Methods
    # =========================================================================

    def check_for_achievements(
        self,
        current_party: List[Dict],
        current_badges: List[str],
        current_map: str,
        game_state: Optional[Dict[str, Any]] = None,
    ) -> Optional[Achievement]:
        """
        Check current game state for new achievements.

        Should be called each game cycle.

        Returns:
            Newly triggered Achievement, or None
        """
        achievement = None

        # Check badge achievements
        achievement = achievement or self._check_badge_achievements(current_badges)

        # Check first Pokemon (starter from Oak)
        achievement = achievement or self._check_first_pokemon(current_party)

        # Check first catch (second Pokemon)
        achievement = achievement or self._check_first_catch(current_party)

        # Check evolution achievements
        achievement = achievement or self._check_evolution(current_party)

        # Check legendary catches
        achievement = achievement or self._check_legendary_catch(current_party)

        # Check champion victory
        achievement = achievement or self._check_champion(current_map, game_state)

        # Update previous state for next cycle
        self._prev_party = [
            p.copy() if isinstance(p, dict) else {} for p in current_party
        ]
        self._prev_badges = list(current_badges) if current_badges else []

        return achievement

    def _check_badge_achievements(
        self, current_badges: List[str]
    ) -> Optional[Achievement]:
        """Check for new badge achievements."""
        if not current_badges:
            return None

        prev_badge_set = set(self._prev_badges) if self._prev_badges else set()
        current_badge_set = set(current_badges)

        new_badges = current_badge_set - prev_badge_set

        for badge_name in new_badges:
            achievement_type = BADGE_TO_ACHIEVEMENT.get(badge_name)
            if achievement_type and not self.is_triggered(achievement_type):
                return self.trigger(
                    achievement_type,
                    context={
                        "badge_name": badge_name,
                        "total_badges": len(current_badges),
                    },
                )

        return None

    def _check_first_pokemon(self, current_party: List[Dict]) -> Optional[Achievement]:
        """Check if player just got their first Pokemon (starter)."""
        if self.is_triggered(AchievementType.FIRST_POKEMON):
            return None

        prev_count = len(self._prev_party)
        current_count = len(current_party)

        # First Pokemon obtained
        if prev_count == 0 and current_count > 0:
            first_pokemon = current_party[0]
            species = first_pokemon.get("name", "").upper()

            # Track starter species for evolution detection
            if species in STARTER_EVOLUTIONS:
                self._starter_species = species
                self._first_pokemon_count = 1

            return self.trigger(
                AchievementType.FIRST_POKEMON,
                context={
                    "pokemon": species,
                    "nickname": first_pokemon.get("nickname", ""),
                },
            )

        return None

    def _check_first_catch(self, current_party: List[Dict]) -> Optional[Achievement]:
        """Check if player caught their first wild Pokemon."""
        if self.is_triggered(AchievementType.FIRST_CATCH):
            return None

        prev_count = len(self._prev_party)
        current_count = len(current_party)

        # Second Pokemon obtained (first catch after starter)
        if prev_count == 1 and current_count == 2:
            caught_pokemon = current_party[1]
            species = caught_pokemon.get("name", "")

            return self.trigger(
                AchievementType.FIRST_CATCH,
                context={
                    "pokemon": species,
                    "nickname": caught_pokemon.get("nickname", ""),
                },
            )

        return None

    def _check_evolution(self, current_party: List[Dict]) -> Optional[Achievement]:
        """Check for starter evolution."""
        if not self._starter_species:
            return None

        # Find starter in current party by checking evolution line
        starter_line = STARTER_EVOLUTIONS.get(self._starter_species, {}).get("line")
        if not starter_line:
            return None

        for pokemon in current_party:
            species = pokemon.get("name", "").upper()
            evo_info = STARTER_EVOLUTIONS.get(species)

            if not evo_info or evo_info.get("line") != starter_line:
                continue

            # Check for first evolution (stage 2)
            if evo_info.get("stage") == 2 and not self.is_triggered(
                AchievementType.STARTER_EVOLUTION_1
            ):
                return self.trigger(
                    AchievementType.STARTER_EVOLUTION_1,
                    context={
                        "pokemon": species,
                        "evolved_from": self._starter_species,
                    },
                )

            # Check for final evolution (stage 3)
            if evo_info.get("stage") == 3 and not self.is_triggered(
                AchievementType.STARTER_EVOLUTION_2
            ):
                return self.trigger(
                    AchievementType.STARTER_EVOLUTION_2,
                    context={
                        "pokemon": species,
                        "starter": self._starter_species,
                    },
                )

        return None

    def _check_legendary_catch(
        self, current_party: List[Dict]
    ) -> Optional[Achievement]:
        """Check for legendary Pokemon catches."""
        prev_species = {p.get("name", "").upper() for p in self._prev_party if p}
        current_species = {p.get("name", "").upper() for p in current_party if p}

        new_pokemon = current_species - prev_species

        for species in new_pokemon:
            if species in LEGENDARY_POKEMON:
                achievement_type = LEGENDARY_TO_ACHIEVEMENT.get(species)
                if achievement_type and not self.is_triggered(achievement_type):
                    return self.trigger(
                        achievement_type,
                        context={"pokemon": species},
                    )

        return None

    def _check_champion(
        self, current_map: str, game_state: Optional[Dict[str, Any]] = None
    ) -> Optional[Achievement]:
        """Check for Pokemon Champion victory."""
        if self.is_triggered(AchievementType.POKEMON_CHAMPION):
            return None

        # Detect entering Hall of Fame
        map_upper = current_map.upper() if current_map else ""
        if "HALL_OF_FAME" in map_upper or "HALL OF FAME" in map_upper:
            return self.trigger(
                AchievementType.POKEMON_CHAMPION,
                context={"location": current_map},
            )

        return None

    def trigger_scripted_photo(
        self, photo_type: AchievementType
    ) -> Optional[Achievement]:
        """
        Trigger a scripted photo moment.

        Called by ScenarioManager when scripted moment is reached.
        Supports all photo-type achievements (scenic, milestone, story).
        """
        # All valid scripted photo types
        valid_photo_types = {
            # Original photo moments
            AchievementType.ROUTE_1_FLOWER,
            AchievementType.VIRIDIAN_FOREST_BREAK,
            # Story/location moments
            AchievementType.SS_ANNE_DECK,
            AchievementType.POKEMON_TOWER_SPOOKED,
            AchievementType.GAME_CORNER_SLOTS,
            AchievementType.FIGHTING_DOJO,
            AchievementType.SAFARI_ZONE_EXPLORER,
            AchievementType.TEAM_ROCKET_FIRST,
            # Scenic/Nature moments
            AchievementType.CERULEAN_CAPE,
            AchievementType.MT_MOON_EXIT,
            AchievementType.ROCK_TUNNEL_EXIT,
            AchievementType.CYCLING_ROAD,
            AchievementType.SEAFOAM_ISLANDS,
            AchievementType.ROUTE_12_FISHING,
            # Milestone moments
            AchievementType.PEWTER_GYM_ENTRANCE,
            AchievementType.INDIGO_PLATEAU,
            AchievementType.DAYCARE_VISIT,
        }

        if photo_type not in valid_photo_types:
            log.warning(f"Invalid scripted photo type: {photo_type}")
            return None

        return self.trigger(photo_type)

    def get_summary(self) -> str:
        """Get a summary of triggered achievements."""
        if not self.triggered:
            return "No achievements yet"

        lines = []
        for ach in self.triggered.values():
            status = "tweeted" if ach.tweet_posted else "pending"
            lines.append(f"- {ach.achievement_type.value}: {status}")

        return "\n".join(lines)

Functions

__init__(storage_path='data/achievements.json', reset_on_start=False)

Initialize the achievement tracker.

Parameters:

Name	Type	Description	Default
storage_path	str	Path to persist achievement state	'data/achievements.json'
reset_on_start	bool	If True, clear all achievements on initialization	False

Source code in trackers/achievement_tracker.py

def __init__(
    self,
    storage_path: str = "data/achievements.json",
    reset_on_start: bool = False,
):
    """
    Initialize the achievement tracker.

    Args:
        storage_path: Path to persist achievement state
        reset_on_start: If True, clear all achievements on initialization
    """
    self.storage_path = storage_path
    self.triggered: Dict[
        str, Achievement
    ] = {}  # achievement_type.value -> Achievement
    self._pending_achievement: Optional[Achievement] = None

    # Track previous state for change detection
    self._prev_party: List[Dict] = []
    self._prev_badges: List[str] = []
    self._starter_species: Optional[str] = None  # Track which starter was chosen
    self._first_pokemon_count: int = 0  # Track if first catch already happened

    # Ensure data directory exists
    Path(os.path.dirname(storage_path)).mkdir(parents=True, exist_ok=True)

    if reset_on_start:
        self._save()
        log.info("Achievement tracker: Fresh start (all achievements reset)")
    else:
        self._load()
        log.info(f"Achievement tracker: Loaded {len(self.triggered)} achievements")

check_for_achievements(current_party, current_badges, current_map, game_state=None)

Check current game state for new achievements.

Should be called each game cycle.

Returns:

Type	Description
Optional[Achievement]	Newly triggered Achievement, or None

Source code in trackers/achievement_tracker.py

def check_for_achievements(
    self,
    current_party: List[Dict],
    current_badges: List[str],
    current_map: str,
    game_state: Optional[Dict[str, Any]] = None,
) -> Optional[Achievement]:
    """
    Check current game state for new achievements.

    Should be called each game cycle.

    Returns:
        Newly triggered Achievement, or None
    """
    achievement = None

    # Check badge achievements
    achievement = achievement or self._check_badge_achievements(current_badges)

    # Check first Pokemon (starter from Oak)
    achievement = achievement or self._check_first_pokemon(current_party)

    # Check first catch (second Pokemon)
    achievement = achievement or self._check_first_catch(current_party)

    # Check evolution achievements
    achievement = achievement or self._check_evolution(current_party)

    # Check legendary catches
    achievement = achievement or self._check_legendary_catch(current_party)

    # Check champion victory
    achievement = achievement or self._check_champion(current_map, game_state)

    # Update previous state for next cycle
    self._prev_party = [
        p.copy() if isinstance(p, dict) else {} for p in current_party
    ]
    self._prev_badges = list(current_badges) if current_badges else []

    return achievement

clear_pending_achievement()

Clear the pending achievement after tweet generation.

Source code in trackers/achievement_tracker.py

def clear_pending_achievement(self) -> None:
    """Clear the pending achievement after tweet generation."""
    self._pending_achievement = None

get_image_prompt(achievement_type)

Get the image prompt for an achievement type.

Source code in trackers/achievement_tracker.py

def get_image_prompt(
    self, achievement_type: AchievementType
) -> AchievementImagePrompt:
    """Get the image prompt for an achievement type."""
    return ACHIEVEMENT_IMAGE_PROMPTS.get(
        achievement_type,
        AchievementImagePrompt(),  # Empty prompt as fallback
    )

get_pending_achievement()

Get the pending achievement for tweet generation.

Source code in trackers/achievement_tracker.py

def get_pending_achievement(self) -> Optional[Achievement]:
    """Get the pending achievement for tweet generation."""
    return self._pending_achievement

get_summary()

Get a summary of triggered achievements.

Source code in trackers/achievement_tracker.py

def get_summary(self) -> str:
    """Get a summary of triggered achievements."""
    if not self.triggered:
        return "No achievements yet"

    lines = []
    for ach in self.triggered.values():
        status = "tweeted" if ach.tweet_posted else "pending"
        lines.append(f"- {ach.achievement_type.value}: {status}")

    return "\n".join(lines)

is_triggered(achievement_type)

Check if an achievement has already been triggered.

Source code in trackers/achievement_tracker.py

def is_triggered(self, achievement_type: AchievementType) -> bool:
    """Check if an achievement has already been triggered."""
    return achievement_type.value in self.triggered

mark_tweet_posted(achievement_type, tweet_url=None)

Mark an achievement's tweet as posted.

Source code in trackers/achievement_tracker.py

def mark_tweet_posted(
    self, achievement_type: AchievementType, tweet_url: Optional[str] = None
) -> None:
    """Mark an achievement's tweet as posted."""
    if achievement_type.value in self.triggered:
        self.triggered[achievement_type.value].tweet_posted = True
        self.triggered[achievement_type.value].tweet_url = tweet_url
        self._save()

trigger(achievement_type, context=None)

Trigger an achievement if not already triggered.

Parameters:

Name	Type	Description	Default
achievement_type	AchievementType	The achievement to trigger	required
context	Optional[Dict[str, Any]]	Additional context (Pokemon name, badge, etc.)	None

Returns:

Type	Description
Optional[Achievement]	The Achievement if newly triggered, None if already triggered

Source code in trackers/achievement_tracker.py

def trigger(
    self,
    achievement_type: AchievementType,
    context: Optional[Dict[str, Any]] = None,
) -> Optional[Achievement]:
    """
    Trigger an achievement if not already triggered.

    Args:
        achievement_type: The achievement to trigger
        context: Additional context (Pokemon name, badge, etc.)

    Returns:
        The Achievement if newly triggered, None if already triggered
    """
    if self.is_triggered(achievement_type):
        log.debug(f"Achievement already triggered: {achievement_type.value}")
        return None

    achievement = Achievement(
        achievement_type=achievement_type,
        triggered_at=datetime.now().isoformat(),
        context=context or {},
    )

    self.triggered[achievement_type.value] = achievement
    self._pending_achievement = achievement
    self._save()

    log.info(f"Achievement triggered: {achievement_type.value}")
    return achievement

trigger_scripted_photo(photo_type)

Trigger a scripted photo moment.

Called by ScenarioManager when scripted moment is reached. Supports all photo-type achievements (scenic, milestone, story).

Source code in trackers/achievement_tracker.py

def trigger_scripted_photo(
    self, photo_type: AchievementType
) -> Optional[Achievement]:
    """
    Trigger a scripted photo moment.

    Called by ScenarioManager when scripted moment is reached.
    Supports all photo-type achievements (scenic, milestone, story).
    """
    # All valid scripted photo types
    valid_photo_types = {
        # Original photo moments
        AchievementType.ROUTE_1_FLOWER,
        AchievementType.VIRIDIAN_FOREST_BREAK,
        # Story/location moments
        AchievementType.SS_ANNE_DECK,
        AchievementType.POKEMON_TOWER_SPOOKED,
        AchievementType.GAME_CORNER_SLOTS,
        AchievementType.FIGHTING_DOJO,
        AchievementType.SAFARI_ZONE_EXPLORER,
        AchievementType.TEAM_ROCKET_FIRST,
        # Scenic/Nature moments
        AchievementType.CERULEAN_CAPE,
        AchievementType.MT_MOON_EXIT,
        AchievementType.ROCK_TUNNEL_EXIT,
        AchievementType.CYCLING_ROAD,
        AchievementType.SEAFOAM_ISLANDS,
        AchievementType.ROUTE_12_FISHING,
        # Milestone moments
        AchievementType.PEWTER_GYM_ENTRANCE,
        AchievementType.INDIGO_PLATEAU,
        AchievementType.DAYCARE_VISIT,
    }

    if photo_type not in valid_photo_types:
        log.warning(f"Invalid scripted photo type: {photo_type}")
        return None

    return self.trigger(photo_type)

AchievementType

Bases: Enum

All trackable achievements in the game.

Source code in trackers/achievement_tracker.py

class AchievementType(Enum):
    """All trackable achievements in the game."""

    # Stream events
    STREAM_START = "stream_start"

    # Early game milestones
    FIRST_POKEMON = "first_pokemon"  # Getting starter from Oak
    ROUTE_1_FLOWER = "route_1_flower"  # Scripted photo moment
    FIRST_CATCH = "first_catch"  # First Pokemon caught with Pokeball
    VIRIDIAN_FOREST_BREAK = "viridian_forest_break"  # Scripted photo moment
    TEAM_ROCKET_FIRST = "team_rocket_first"  # First encounter with Team Rocket

    # Evolution milestones
    STARTER_EVOLUTION_1 = (
        "starter_evolution_1"  # First evolution (e.g., Charmander -> Charmeleon)
    )
    STARTER_EVOLUTION_2 = (
        "starter_evolution_2"  # Final evolution (e.g., Charmeleon -> Charizard)
    )

    # Gym badges
    BADGE_BOULDER = "badge_boulder"  # Brock
    BADGE_CASCADE = "badge_cascade"  # Misty
    BADGE_THUNDER = "badge_thunder"  # Lt. Surge
    BADGE_RAINBOW = "badge_rainbow"  # Erika
    BADGE_SOUL = "badge_soul"  # Koga
    BADGE_MARSH = "badge_marsh"  # Sabrina
    BADGE_VOLCANO = "badge_volcano"  # Blaine
    BADGE_EARTH = "badge_earth"  # Giovanni

    # Scripted location moments
    SS_ANNE_DECK = "ss_anne_deck"  # Looking out from SS Anne bow
    POKEMON_TOWER_SPOOKED = "pokemon_tower_spooked"  # Scared in Pokemon Tower
    GAME_CORNER_SLOTS = "game_corner_slots"  # Playing slots in Game Corner
    FIGHTING_DOJO = "fighting_dojo"  # Training at Fighting Dojo
    SAFARI_ZONE_EXPLORER = "safari_zone_explorer"  # Exploring Safari Zone

    # Scenic/Nature photo moments
    CERULEAN_CAPE = "cerulean_cape"  # Ocean view near Bill's house
    MT_MOON_EXIT = "mt_moon_exit"  # Relief after exiting Mt. Moon
    ROCK_TUNNEL_EXIT = "rock_tunnel_exit"  # Daylight after Rock Tunnel
    CYCLING_ROAD = "cycling_road"  # Cruising down Cycling Road
    SEAFOAM_ISLANDS = "seafoam_islands"  # Icy cave adventure
    ROUTE_12_FISHING = "route_12_fishing"  # Fishing by the pier

    # Milestone photo moments
    PEWTER_GYM_ENTRANCE = "pewter_gym_entrance"  # First gym entrance
    INDIGO_PLATEAU = "indigo_plateau"  # Arriving at Pokemon League
    DAYCARE_VISIT = "daycare_visit"  # First visit to the Daycare

    # Legendary captures
    LEGENDARY_ARTICUNO = "legendary_articuno"
    LEGENDARY_ZAPDOS = "legendary_zapdos"
    LEGENDARY_MOLTRES = "legendary_moltres"
    LEGENDARY_MEWTWO = "legendary_mewtwo"

    # Endgame
    POKEMON_CHAMPION = "pokemon_champion"  # Defeated Elite 4 and Champion

Functions

build_dynamic_prompt_context(location='', map_type='', party=None, badges=0, recent_events='')

Build the dynamic prompt with game state filled in.

Parameters:

Name	Type	Description	Default
location	str	Current map/location name	''
map_type	str	Type of location (city, route, dungeon, etc.)	''
party	Optional[List[str]]	List of Pokemon names in party	None
badges	int	Number of badges earned	0
recent_events	str	Description of recent game events	''

Returns:

Type	Description
str	Formatted prompt string ready to send to LLM.

Source code in trackers/achievement_tracker.py

def build_dynamic_prompt_context(
    location: str = "",
    map_type: str = "",
    party: Optional[List[str]] = None,
    badges: int = 0,
    recent_events: str = "",
) -> str:
    """
    Build the dynamic prompt with game state filled in.

    Args:
        location: Current map/location name
        map_type: Type of location (city, route, dungeon, etc.)
        party: List of Pokemon names in party
        badges: Number of badges earned
        recent_events: Description of recent game events

    Returns:
        Formatted prompt string ready to send to LLM.
    """
    template = get_dynamic_prompt_template()
    party_str = ", ".join(party) if party else "None"

    return template.format(
        location=location or "Unknown",
        map_type=map_type or "unknown",
        party=party_str,
        badges=badges,
        recent_events=recent_events or "Continuing the adventure",
    )

create_achievement_tracker(storage_path='data/achievements.json', reset_on_start=False)

Factory function to create an AchievementTracker instance.

Source code in trackers/achievement_tracker.py

def create_achievement_tracker(
    storage_path: str = "data/achievements.json",
    reset_on_start: bool = False,
) -> AchievementTracker:
    """Factory function to create an AchievementTracker instance."""
    return AchievementTracker(storage_path=storage_path, reset_on_start=reset_on_start)

get_achievement_image_prompts()

Get achievement image prompts, preferring YAML source.

Source code in trackers/achievement_tracker.py

def get_achievement_image_prompts() -> Dict[AchievementType, AchievementImagePrompt]:
    """Get achievement image prompts, preferring YAML source."""
    yaml_prompts = _build_prompts_from_yaml()
    if yaml_prompts:
        return yaml_prompts
    return _FALLBACK_ACHIEVEMENT_IMAGE_PROMPTS

get_dynamic_prompt_template()

Get the dynamic prompt template for LLM-generated image prompts.

Used for CYCLE_CHECKPOINT (progress/fallback) posts where the LLM generates prompt components based on the current game state.

Returns:

Type	Description
str	The system prompt template with placeholders for game state.

Source code in trackers/achievement_tracker.py

def get_dynamic_prompt_template() -> str:
    """
    Get the dynamic prompt template for LLM-generated image prompts.

    Used for CYCLE_CHECKPOINT (progress/fallback) posts where the LLM
    generates prompt components based on the current game state.

    Returns:
        The system prompt template with placeholders for game state.
    """
    yaml_data = _load_yaml_prompts()
    dynamic = yaml_data.get("dynamic_template", {})
    return dynamic.get("system_prompt", _DEFAULT_DYNAMIC_TEMPLATE)

parse_dynamic_prompt_response(response)

Parse LLM response into an AchievementImagePrompt.

Expects JSON with: setting, mood_expression, action, extras

Parameters:

Name	Type	Description	Default
response	str	Raw LLM response string	required

Returns:

Type	Description
Optional[AchievementImagePrompt]	AchievementImagePrompt or None if parsing fails.

Source code in trackers/achievement_tracker.py

def parse_dynamic_prompt_response(response: str) -> Optional[AchievementImagePrompt]:
    """
    Parse LLM response into an AchievementImagePrompt.

    Expects JSON with: setting, mood_expression, action, extras

    Args:
        response: Raw LLM response string

    Returns:
        AchievementImagePrompt or None if parsing fails.
    """
    import json
    import re

    try:
        # Try to extract JSON from response (may have extra text)
        json_match = re.search(r"\{[^{}]*\}", response, re.DOTALL)
        if not json_match:
            log.warning("No JSON found in dynamic prompt response")
            return None

        data = json.loads(json_match.group())

        return AchievementImagePrompt(
            setting=data.get("setting", ""),
            mood_expression=data.get("mood_expression", ""),
            action=data.get("action", ""),
            extras=data.get("extras", ""),
        )
    except Exception as e:
        log.error(f"Error parsing dynamic prompt response: {e}")
        return None

Zora Achievement Tracker

trackers.zora_achievement_tracker

Zora Achievement Tracker for LLMLetsPlay Brand.

Extends the base AchievementTracker with additional minor achievements that are posted to Zora only (not Twitter). These create a richer documentary of the Pokemon journey.

Achievement Tiers: - MAJOR: Posted to Twitter + Zora with ComfyUI-generated image - MINOR: Posted to Zora only with screenshot - PROGRESS: Fallback checkpoint posts (every N cycles with no achievements)

Key Features: - First-catch-per-species detection (derived from Pokedex state) - Route/city/dungeon transition detection - Team change tracking - HM and key item detection - Fallback progress checkpoints

Integration: - Works alongside existing AchievementTracker - Stores Zora-specific state in separate JSON file - Provides post counter queried from Zora API on startup

Classes

ZoraAchievement dataclass

A Zora-specific achievement.

Source code in trackers/zora_achievement_tracker.py

@dataclass
class ZoraAchievement:
    """A Zora-specific achievement."""

    achievement_type: ZoraAchievementType
    tier: ZoraAchievementTier
    triggered_at: str  # ISO timestamp
    context: Dict[str, Any] = field(default_factory=dict)
    zora_posted: bool = False
    zora_coin_address: Optional[str] = None
    post_number: Optional[int] = None  # LLP-NNN number

    def get_title(self) -> str:
        """Generate post title based on achievement type and context."""
        prefix = f"LLP-{self.post_number:03d}" if self.post_number else "LLP"

        type_titles = {
            ZoraAchievementType.NEW_ROUTE: f"Exploring {self.context.get('location', 'New Route')}",
            ZoraAchievementType.NEW_CITY: f"Arrived at {self.context.get('location', 'New City')}",
            ZoraAchievementType.NEW_DUNGEON: f"Entering {self.context.get('location', 'Unknown')}",
            ZoraAchievementType.DUNGEON_EXIT: f"Escaped from {self.context.get('location', 'Unknown')}",
            ZoraAchievementType.FIRST_CATCH_SPECIES: f"Caught a {self.context.get('pokemon', 'Pokemon')}!",
            ZoraAchievementType.POKEMON_EVOLVED: f"{self.context.get('pokemon', 'Pokemon')} Evolved!",
            ZoraAchievementType.TEAM_FULL: "Team is Complete!",
            ZoraAchievementType.POKEDEX_MILESTONE: f"Pokedex: {self.context.get('count', 0)} Pokemon",
            ZoraAchievementType.RIVAL_DEFEATED: "Defeated Rival!",
            ZoraAchievementType.ROCKET_DEFEATED: "Team Rocket Defeated!",
            ZoraAchievementType.HM_OBTAINED: f"Obtained {self.context.get('item', 'HM')}!",
            ZoraAchievementType.KEY_ITEM: f"Got {self.context.get('item', 'Key Item')}!",
            ZoraAchievementType.CYCLE_CHECKPOINT: "Journey Progress",
            # Major achievements (these have their own handling)
            ZoraAchievementType.STREAM_START: "Lass Begins Her Journey",
            ZoraAchievementType.BADGE_BOULDER: "Boulder Badge Earned!",
            ZoraAchievementType.BADGE_CASCADE: "Cascade Badge Earned!",
            ZoraAchievementType.BADGE_THUNDER: "Thunder Badge Earned!",
            ZoraAchievementType.BADGE_RAINBOW: "Rainbow Badge Earned!",
            ZoraAchievementType.BADGE_SOUL: "Soul Badge Earned!",
            ZoraAchievementType.BADGE_MARSH: "Marsh Badge Earned!",
            ZoraAchievementType.BADGE_VOLCANO: "Volcano Badge Earned!",
            ZoraAchievementType.BADGE_EARTH: "Earth Badge Earned!",
            ZoraAchievementType.POKEMON_CHAMPION: "Pokemon Champion!",
        }

        title = type_titles.get(
            self.achievement_type, self.achievement_type.value.replace("_", " ").title()
        )
        return f"{prefix}: {title}"

    def should_include_footer(self) -> bool:
        """Check if this achievement should include cross-promotion footer."""
        return self.tier in (ZoraAchievementTier.MAJOR, ZoraAchievementTier.PROGRESS)

    def to_dict(self) -> dict:
        """Convert to dictionary for serialization."""
        return {
            "achievement_type": self.achievement_type.value,
            "tier": self.tier.value,
            "triggered_at": self.triggered_at,
            "context": self.context,
            "zora_posted": self.zora_posted,
            "zora_coin_address": self.zora_coin_address,
            "post_number": self.post_number,
        }

Functions

get_title()

Generate post title based on achievement type and context.

Source code in trackers/zora_achievement_tracker.py

def get_title(self) -> str:
    """Generate post title based on achievement type and context."""
    prefix = f"LLP-{self.post_number:03d}" if self.post_number else "LLP"

    type_titles = {
        ZoraAchievementType.NEW_ROUTE: f"Exploring {self.context.get('location', 'New Route')}",
        ZoraAchievementType.NEW_CITY: f"Arrived at {self.context.get('location', 'New City')}",
        ZoraAchievementType.NEW_DUNGEON: f"Entering {self.context.get('location', 'Unknown')}",
        ZoraAchievementType.DUNGEON_EXIT: f"Escaped from {self.context.get('location', 'Unknown')}",
        ZoraAchievementType.FIRST_CATCH_SPECIES: f"Caught a {self.context.get('pokemon', 'Pokemon')}!",
        ZoraAchievementType.POKEMON_EVOLVED: f"{self.context.get('pokemon', 'Pokemon')} Evolved!",
        ZoraAchievementType.TEAM_FULL: "Team is Complete!",
        ZoraAchievementType.POKEDEX_MILESTONE: f"Pokedex: {self.context.get('count', 0)} Pokemon",
        ZoraAchievementType.RIVAL_DEFEATED: "Defeated Rival!",
        ZoraAchievementType.ROCKET_DEFEATED: "Team Rocket Defeated!",
        ZoraAchievementType.HM_OBTAINED: f"Obtained {self.context.get('item', 'HM')}!",
        ZoraAchievementType.KEY_ITEM: f"Got {self.context.get('item', 'Key Item')}!",
        ZoraAchievementType.CYCLE_CHECKPOINT: "Journey Progress",
        # Major achievements (these have their own handling)
        ZoraAchievementType.STREAM_START: "Lass Begins Her Journey",
        ZoraAchievementType.BADGE_BOULDER: "Boulder Badge Earned!",
        ZoraAchievementType.BADGE_CASCADE: "Cascade Badge Earned!",
        ZoraAchievementType.BADGE_THUNDER: "Thunder Badge Earned!",
        ZoraAchievementType.BADGE_RAINBOW: "Rainbow Badge Earned!",
        ZoraAchievementType.BADGE_SOUL: "Soul Badge Earned!",
        ZoraAchievementType.BADGE_MARSH: "Marsh Badge Earned!",
        ZoraAchievementType.BADGE_VOLCANO: "Volcano Badge Earned!",
        ZoraAchievementType.BADGE_EARTH: "Earth Badge Earned!",
        ZoraAchievementType.POKEMON_CHAMPION: "Pokemon Champion!",
    }

    title = type_titles.get(
        self.achievement_type, self.achievement_type.value.replace("_", " ").title()
    )
    return f"{prefix}: {title}"

should_include_footer()

Check if this achievement should include cross-promotion footer.

Source code in trackers/zora_achievement_tracker.py

def should_include_footer(self) -> bool:
    """Check if this achievement should include cross-promotion footer."""
    return self.tier in (ZoraAchievementTier.MAJOR, ZoraAchievementTier.PROGRESS)

to_dict()

Convert to dictionary for serialization.

Source code in trackers/zora_achievement_tracker.py

def to_dict(self) -> dict:
    """Convert to dictionary for serialization."""
    return {
        "achievement_type": self.achievement_type.value,
        "tier": self.tier.value,
        "triggered_at": self.triggered_at,
        "context": self.context,
        "zora_posted": self.zora_posted,
        "zora_coin_address": self.zora_coin_address,
        "post_number": self.post_number,
    }

ZoraAchievementTier

Bases: Enum

Achievement tier determines where it's posted and image type.

Source code in trackers/zora_achievement_tracker.py

class ZoraAchievementTier(Enum):
    """Achievement tier determines where it's posted and image type."""

    MAJOR = "major"  # Twitter + Zora, ComfyUI image, with footer
    MINOR = "minor"  # Zora only, screenshot, no footer
    PROGRESS = "progress"  # Zora only, screenshot gallery, with footer

ZoraAchievementTracker

Tracks achievements for Zora posting.

Works alongside the existing AchievementTracker but tracks: - Minor achievements (Zora-only) - Caught species (for first-catch-per-species) - Visited locations (for route/city/dungeon detection) - Post counter (LLP-NNN)

The post counter is queried from Zora API on startup.

Source code in trackers/zora_achievement_tracker.py

class ZoraAchievementTracker:
    """
    Tracks achievements for Zora posting.

    Works alongside the existing AchievementTracker but tracks:
    - Minor achievements (Zora-only)
    - Caught species (for first-catch-per-species)
    - Visited locations (for route/city/dungeon detection)
    - Post counter (LLP-NNN)

    The post counter is queried from Zora API on startup.
    """

    def __init__(
        self,
        storage_path: str = "data/zora_achievements.json",
        fallback_cycles: int = 60,
        min_post_interval: int = 300,  # 5 minutes minimum between posts
    ):
        """
        Initialize the Zora achievement tracker.

        Args:
            storage_path: Path to persist Zora achievement state
            fallback_cycles: Cycles without achievements before checkpoint post
            min_post_interval: Minimum seconds between Zora posts
        """
        self.storage_path = storage_path
        self.fallback_cycles = fallback_cycles
        self.min_post_interval = min_post_interval

        # Achievement tracking
        self.triggered: Dict[str, ZoraAchievement] = {}
        self._pending_achievement: Optional[ZoraAchievement] = None

        # Species catch tracking (derived from Pokedex)
        self._caught_species: Set[str] = set()

        # Location tracking
        self._visited_routes: Set[str] = set()
        self._visited_cities: Set[str] = set()
        self._visited_dungeons: Set[str] = set()
        self._current_location: str = ""
        self._prev_location: str = ""

        # Party tracking
        self._prev_party: List[Dict] = []
        self._prev_party_size: int = 0

        # Pokedex tracking
        self._prev_pokedex_count: int = 0

        # Post counter (will be set from Zora API)
        self._post_counter: int = 0

        # Timing
        self._last_post_time: float = 0
        self._cycles_since_last_post: int = 0

        # Ensure data directory exists
        Path(os.path.dirname(storage_path)).mkdir(parents=True, exist_ok=True)

        self._load()
        log.info(
            f"Zora achievement tracker initialized: "
            f"{len(self.triggered)} achievements, "
            f"post counter at {self._post_counter}"
        )

    def set_post_counter(self, count: int):
        """
        Set the post counter (called after querying Zora API on startup).

        Args:
            count: Current post count from Zora
        """
        self._post_counter = count
        self._save()
        log.info(f"Zora post counter set to {count}")

    def get_next_post_number(self) -> int:
        """Get the next post number (increments counter)."""
        self._post_counter += 1
        self._save()
        return self._post_counter

    def _load(self) -> None:
        """Load state from disk."""
        if not os.path.exists(self.storage_path):
            return

        try:
            with open(self.storage_path, "r") as f:
                data = json.load(f)

            # Load triggered achievements
            for key, value in data.get("triggered", {}).items():
                try:
                    achievement_type = ZoraAchievementType(value["achievement_type"])
                    tier = ZoraAchievementTier(value["tier"])
                    self.triggered[key] = ZoraAchievement(
                        achievement_type=achievement_type,
                        tier=tier,
                        triggered_at=value["triggered_at"],
                        context=value.get("context", {}),
                        zora_posted=value.get("zora_posted", False),
                        zora_coin_address=value.get("zora_coin_address"),
                        post_number=value.get("post_number"),
                    )
                except (KeyError, ValueError) as e:
                    log.warning(f"Failed to load Zora achievement {key}: {e}")

            # Load tracking state
            self._caught_species = set(data.get("caught_species", []))
            self._visited_routes = set(data.get("visited_routes", []))
            self._visited_cities = set(data.get("visited_cities", []))
            self._visited_dungeons = set(data.get("visited_dungeons", []))
            self._post_counter = data.get("post_counter", 0)
            self._prev_pokedex_count = data.get("prev_pokedex_count", 0)
            self._cycles_since_last_post = data.get("cycles_since_last_post", 0)

        except Exception as e:
            log.error(f"Failed to load Zora achievements: {e}")

    def _save(self) -> None:
        """Save state to disk."""
        try:
            data = {
                "triggered": {
                    key: ach.to_dict() for key, ach in self.triggered.items()
                },
                "caught_species": list(self._caught_species),
                "visited_routes": list(self._visited_routes),
                "visited_cities": list(self._visited_cities),
                "visited_dungeons": list(self._visited_dungeons),
                "post_counter": self._post_counter,
                "prev_pokedex_count": self._prev_pokedex_count,
                "cycles_since_last_post": self._cycles_since_last_post,
            }

            with open(self.storage_path, "w") as f:
                json.dump(data, f, indent=2)

        except Exception as e:
            log.error(f"Failed to save Zora achievements: {e}")

    def trigger(
        self,
        achievement_type: ZoraAchievementType,
        context: Optional[Dict[str, Any]] = None,
        unique_key: Optional[str] = None,
    ) -> Optional[ZoraAchievement]:
        """
        Trigger a Zora achievement.

        Args:
            achievement_type: The type of achievement
            context: Additional context for the achievement
            unique_key: Optional unique key for achievements that can trigger multiple times
                        (e.g., "first_catch_species_PIDGEY")

        Returns:
            ZoraAchievement if newly triggered, None if already exists
        """
        # Generate key for this achievement
        key = unique_key or achievement_type.value

        if key in self.triggered:
            log.debug(f"Zora achievement already triggered: {key}")
            return None

        tier = ACHIEVEMENT_TIERS.get(achievement_type, ZoraAchievementTier.MINOR)
        post_number = self.get_next_post_number()

        achievement = ZoraAchievement(
            achievement_type=achievement_type,
            tier=tier,
            triggered_at=datetime.now().isoformat(),
            context=context or {},
            post_number=post_number,
        )

        self.triggered[key] = achievement
        self._pending_achievement = achievement
        self._cycles_since_last_post = 0
        self._save()

        log.info(
            f"Zora achievement triggered: {achievement.get_title()} "
            f"(tier: {tier.value})"
        )
        return achievement

    def get_pending_achievement(self) -> Optional[ZoraAchievement]:
        """Get the pending achievement for Zora posting."""
        return self._pending_achievement

    def clear_pending_achievement(self) -> None:
        """Clear the pending achievement after posting."""
        self._pending_achievement = None

    def mark_posted(
        self,
        key: str,
        coin_address: Optional[str] = None,
    ) -> None:
        """Mark an achievement as posted to Zora."""
        if key in self.triggered:
            self.triggered[key].zora_posted = True
            self.triggered[key].zora_coin_address = coin_address
            self._save()

    def check_for_achievements(
        self,
        current_party: List[Dict],
        current_map: str,
        pokedex_caught: int = 0,
        game_state: Optional[Dict[str, Any]] = None,
    ) -> Optional[ZoraAchievement]:
        """
        Check current game state for new Zora achievements.

        This checks for MINOR achievements only. MAJOR achievements
        are handled by the main AchievementTracker and forwarded here.

        Args:
            current_party: Current party Pokemon list
            current_map: Current map/location name
            pokedex_caught: Number of Pokemon caught (from Pokedex)
            game_state: Full game state dict

        Returns:
            Newly triggered ZoraAchievement, or None
        """
        self._cycles_since_last_post += 1
        achievement = None

        # Normalize map name
        map_normalized = self._normalize_location(current_map)

        # Check location changes (route/city/dungeon)
        achievement = achievement or self._check_location_change(map_normalized)

        # Check for new Pokemon species in party
        achievement = achievement or self._check_new_species(current_party)

        # Check team size changes
        achievement = achievement or self._check_team_changes(current_party)

        # Check Pokedex milestones
        achievement = achievement or self._check_pokedex_milestone(pokedex_caught)

        # Check for fallback checkpoint
        if not achievement:
            achievement = self._check_fallback_checkpoint()

        # Update previous state
        self._prev_location = map_normalized
        self._prev_party = [
            p.copy() if isinstance(p, dict) else {} for p in current_party
        ]
        self._prev_party_size = len(current_party)
        self._prev_pokedex_count = pokedex_caught

        return achievement

    def _normalize_location(self, location: str) -> str:
        """Normalize location name for comparison."""
        if not location:
            return ""
        return location.upper().replace(" ", "_").replace("-", "_")

    def _classify_location(self, location: str) -> Optional[str]:
        """
        Classify a location as route, city, or dungeon.

        Returns:
            "route", "city", "dungeon", or None
        """
        if not location:
            return None

        # Check exact matches first
        if location in ROUTES:
            return "route"
        if location in CITIES:
            return "city"
        if location in DUNGEONS:
            return "dungeon"

        # Check partial matches
        for route in ROUTES:
            if route in location or location in route:
                return "route"
        for city in CITIES:
            if city in location or location in city:
                return "city"
        for dungeon in DUNGEONS:
            if dungeon in location or location in dungeon:
                return "dungeon"

        return None

    def _check_location_change(
        self, current_location: str
    ) -> Optional[ZoraAchievement]:
        """Check for new location achievements."""
        if not current_location or current_location == self._prev_location:
            return None

        location_type = self._classify_location(current_location)

        if location_type == "route" and current_location not in self._visited_routes:
            self._visited_routes.add(current_location)
            return self.trigger(
                ZoraAchievementType.NEW_ROUTE,
                context={"location": current_location.replace("_", " ").title()},
                unique_key=f"new_route_{current_location}",
            )

        if location_type == "city" and current_location not in self._visited_cities:
            self._visited_cities.add(current_location)
            return self.trigger(
                ZoraAchievementType.NEW_CITY,
                context={"location": current_location.replace("_", " ").title()},
                unique_key=f"new_city_{current_location}",
            )

        if location_type == "dungeon":
            # Check if entering or exiting dungeon
            prev_type = self._classify_location(self._prev_location)

            if current_location not in self._visited_dungeons:
                self._visited_dungeons.add(current_location)
                return self.trigger(
                    ZoraAchievementType.NEW_DUNGEON,
                    context={"location": current_location.replace("_", " ").title()},
                    unique_key=f"new_dungeon_{current_location}",
                )
            elif prev_type == "dungeon" and location_type != "dungeon":
                # Just exited a dungeon
                return self.trigger(
                    ZoraAchievementType.DUNGEON_EXIT,
                    context={"location": self._prev_location.replace("_", " ").title()},
                    unique_key=f"dungeon_exit_{self._prev_location}_{self._cycles_since_last_post}",
                )

        return None

    def _check_new_species(
        self, current_party: List[Dict]
    ) -> Optional[ZoraAchievement]:
        """Check for first catch of a new species."""
        current_species = set()
        for pokemon in current_party:
            if pokemon:
                species = pokemon.get("name", "").upper()
                if species:
                    current_species.add(species)

        # Find newly caught species
        new_species = current_species - self._caught_species

        for species in new_species:
            self._caught_species.add(species)

            # Skip if this is likely the starter (first Pokemon)
            if len(self._caught_species) == 1:
                continue

            # Skip second Pokemon (handled by FIRST_CATCH)
            if len(self._caught_species) == 2:
                continue

            # Trigger first catch for this species
            return self.trigger(
                ZoraAchievementType.FIRST_CATCH_SPECIES,
                context={
                    "pokemon": species.title(),
                    "pokedex_count": len(self._caught_species),
                },
                unique_key=f"first_catch_species_{species}",
            )

        return None

    def _check_team_changes(
        self, current_party: List[Dict]
    ) -> Optional[ZoraAchievement]:
        """Check for team size changes."""
        current_size = len(current_party)

        # Check for full team (6 Pokemon)
        if current_size == 6 and self._prev_party_size < 6:
            key = "team_full"
            if key not in self.triggered:
                pokemon_names = [
                    p.get("name", "Unknown").title() for p in current_party if p
                ]
                return self.trigger(
                    ZoraAchievementType.TEAM_FULL,
                    context={"team": pokemon_names},
                    unique_key=key,
                )

        # Check for evolutions (non-starter)
        for i, pokemon in enumerate(current_party):
            if i >= len(self._prev_party):
                continue

            prev_pokemon = self._prev_party[i]
            current_species = pokemon.get("name", "").upper() if pokemon else ""
            prev_species = prev_pokemon.get("name", "").upper() if prev_pokemon else ""

            if current_species and prev_species and current_species != prev_species:
                # Evolution detected (same slot, different species)
                key = f"evolution_{current_species}"
                if key not in self.triggered:
                    return self.trigger(
                        ZoraAchievementType.POKEMON_EVOLVED,
                        context={
                            "pokemon": current_species.title(),
                            "evolved_from": prev_species.title(),
                        },
                        unique_key=key,
                    )

        return None

    def _check_pokedex_milestone(
        self, pokedex_caught: int
    ) -> Optional[ZoraAchievement]:
        """Check for Pokedex milestone (every 10 Pokemon)."""
        if pokedex_caught <= 0:
            return None

        # Check if we crossed a 10-Pokemon milestone
        prev_milestone = (self._prev_pokedex_count // 10) * 10
        current_milestone = (pokedex_caught // 10) * 10

        if current_milestone > prev_milestone and current_milestone > 0:
            key = f"pokedex_milestone_{current_milestone}"
            if key not in self.triggered:
                return self.trigger(
                    ZoraAchievementType.POKEDEX_MILESTONE,
                    context={"count": current_milestone},
                    unique_key=key,
                )

        return None

    def _check_fallback_checkpoint(self) -> Optional[ZoraAchievement]:
        """Check if we should trigger a fallback checkpoint post."""
        if self._cycles_since_last_post >= self.fallback_cycles:
            key = f"checkpoint_{self._post_counter + 1}"
            return self.trigger(
                ZoraAchievementType.CYCLE_CHECKPOINT,
                context={
                    "cycles": self._cycles_since_last_post,
                    "is_fallback": True,
                },
                unique_key=key,
            )
        return None

    def trigger_from_major_achievement(
        self,
        achievement_type_value: str,
        context: Optional[Dict[str, Any]] = None,
    ) -> Optional[ZoraAchievement]:
        """
        Trigger a Zora achievement from a major achievement (from AchievementTracker).

        This is called when the main AchievementTracker triggers a major achievement
        that should also be posted to Zora.

        Args:
            achievement_type_value: The achievement type value string
            context: Achievement context

        Returns:
            ZoraAchievement if triggered
        """
        try:
            zora_type = ZoraAchievementType(achievement_type_value)
        except ValueError:
            log.warning(f"Unknown achievement type for Zora: {achievement_type_value}")
            return None

        return self.trigger(zora_type, context=context)

    def get_stats(self) -> dict:
        """Get tracker statistics."""
        major_count = sum(
            1 for a in self.triggered.values() if a.tier == ZoraAchievementTier.MAJOR
        )
        minor_count = sum(
            1 for a in self.triggered.values() if a.tier == ZoraAchievementTier.MINOR
        )
        progress_count = sum(
            1 for a in self.triggered.values() if a.tier == ZoraAchievementTier.PROGRESS
        )

        return {
            "total_achievements": len(self.triggered),
            "major_achievements": major_count,
            "minor_achievements": minor_count,
            "progress_checkpoints": progress_count,
            "post_counter": self._post_counter,
            "caught_species": len(self._caught_species),
            "visited_routes": len(self._visited_routes),
            "visited_cities": len(self._visited_cities),
            "visited_dungeons": len(self._visited_dungeons),
            "cycles_since_last_post": self._cycles_since_last_post,
        }

Functions

__init__(storage_path='data/zora_achievements.json', fallback_cycles=60, min_post_interval=300)

Initialize the Zora achievement tracker.

Parameters:

Name	Type	Description	Default
storage_path	str	Path to persist Zora achievement state	'data/zora_achievements.json'
fallback_cycles	int	Cycles without achievements before checkpoint post	60
min_post_interval	int	Minimum seconds between Zora posts	300

Source code in trackers/zora_achievement_tracker.py

def __init__(
    self,
    storage_path: str = "data/zora_achievements.json",
    fallback_cycles: int = 60,
    min_post_interval: int = 300,  # 5 minutes minimum between posts
):
    """
    Initialize the Zora achievement tracker.

    Args:
        storage_path: Path to persist Zora achievement state
        fallback_cycles: Cycles without achievements before checkpoint post
        min_post_interval: Minimum seconds between Zora posts
    """
    self.storage_path = storage_path
    self.fallback_cycles = fallback_cycles
    self.min_post_interval = min_post_interval

    # Achievement tracking
    self.triggered: Dict[str, ZoraAchievement] = {}
    self._pending_achievement: Optional[ZoraAchievement] = None

    # Species catch tracking (derived from Pokedex)
    self._caught_species: Set[str] = set()

    # Location tracking
    self._visited_routes: Set[str] = set()
    self._visited_cities: Set[str] = set()
    self._visited_dungeons: Set[str] = set()
    self._current_location: str = ""
    self._prev_location: str = ""

    # Party tracking
    self._prev_party: List[Dict] = []
    self._prev_party_size: int = 0

    # Pokedex tracking
    self._prev_pokedex_count: int = 0

    # Post counter (will be set from Zora API)
    self._post_counter: int = 0

    # Timing
    self._last_post_time: float = 0
    self._cycles_since_last_post: int = 0

    # Ensure data directory exists
    Path(os.path.dirname(storage_path)).mkdir(parents=True, exist_ok=True)

    self._load()
    log.info(
        f"Zora achievement tracker initialized: "
        f"{len(self.triggered)} achievements, "
        f"post counter at {self._post_counter}"
    )

check_for_achievements(current_party, current_map, pokedex_caught=0, game_state=None)

Check current game state for new Zora achievements.

This checks for MINOR achievements only. MAJOR achievements are handled by the main AchievementTracker and forwarded here.

Parameters:

Name	Type	Description	Default
current_party	List[Dict]	Current party Pokemon list	required
current_map	str	Current map/location name	required
pokedex_caught	int	Number of Pokemon caught (from Pokedex)	0
game_state	Optional[Dict[str, Any]]	Full game state dict	None

Returns:

Type	Description
Optional[ZoraAchievement]	Newly triggered ZoraAchievement, or None

Source code in trackers/zora_achievement_tracker.py

def check_for_achievements(
    self,
    current_party: List[Dict],
    current_map: str,
    pokedex_caught: int = 0,
    game_state: Optional[Dict[str, Any]] = None,
) -> Optional[ZoraAchievement]:
    """
    Check current game state for new Zora achievements.

    This checks for MINOR achievements only. MAJOR achievements
    are handled by the main AchievementTracker and forwarded here.

    Args:
        current_party: Current party Pokemon list
        current_map: Current map/location name
        pokedex_caught: Number of Pokemon caught (from Pokedex)
        game_state: Full game state dict

    Returns:
        Newly triggered ZoraAchievement, or None
    """
    self._cycles_since_last_post += 1
    achievement = None

    # Normalize map name
    map_normalized = self._normalize_location(current_map)

    # Check location changes (route/city/dungeon)
    achievement = achievement or self._check_location_change(map_normalized)

    # Check for new Pokemon species in party
    achievement = achievement or self._check_new_species(current_party)

    # Check team size changes
    achievement = achievement or self._check_team_changes(current_party)

    # Check Pokedex milestones
    achievement = achievement or self._check_pokedex_milestone(pokedex_caught)

    # Check for fallback checkpoint
    if not achievement:
        achievement = self._check_fallback_checkpoint()

    # Update previous state
    self._prev_location = map_normalized
    self._prev_party = [
        p.copy() if isinstance(p, dict) else {} for p in current_party
    ]
    self._prev_party_size = len(current_party)
    self._prev_pokedex_count = pokedex_caught

    return achievement

clear_pending_achievement()

Clear the pending achievement after posting.

Source code in trackers/zora_achievement_tracker.py

def clear_pending_achievement(self) -> None:
    """Clear the pending achievement after posting."""
    self._pending_achievement = None

get_next_post_number()

Get the next post number (increments counter).

Source code in trackers/zora_achievement_tracker.py

def get_next_post_number(self) -> int:
    """Get the next post number (increments counter)."""
    self._post_counter += 1
    self._save()
    return self._post_counter

get_pending_achievement()

Get the pending achievement for Zora posting.

Source code in trackers/zora_achievement_tracker.py

def get_pending_achievement(self) -> Optional[ZoraAchievement]:
    """Get the pending achievement for Zora posting."""
    return self._pending_achievement

get_stats()

Get tracker statistics.

Source code in trackers/zora_achievement_tracker.py

def get_stats(self) -> dict:
    """Get tracker statistics."""
    major_count = sum(
        1 for a in self.triggered.values() if a.tier == ZoraAchievementTier.MAJOR
    )
    minor_count = sum(
        1 for a in self.triggered.values() if a.tier == ZoraAchievementTier.MINOR
    )
    progress_count = sum(
        1 for a in self.triggered.values() if a.tier == ZoraAchievementTier.PROGRESS
    )

    return {
        "total_achievements": len(self.triggered),
        "major_achievements": major_count,
        "minor_achievements": minor_count,
        "progress_checkpoints": progress_count,
        "post_counter": self._post_counter,
        "caught_species": len(self._caught_species),
        "visited_routes": len(self._visited_routes),
        "visited_cities": len(self._visited_cities),
        "visited_dungeons": len(self._visited_dungeons),
        "cycles_since_last_post": self._cycles_since_last_post,
    }

mark_posted(key, coin_address=None)

Mark an achievement as posted to Zora.

Source code in trackers/zora_achievement_tracker.py

def mark_posted(
    self,
    key: str,
    coin_address: Optional[str] = None,
) -> None:
    """Mark an achievement as posted to Zora."""
    if key in self.triggered:
        self.triggered[key].zora_posted = True
        self.triggered[key].zora_coin_address = coin_address
        self._save()

set_post_counter(count)

Set the post counter (called after querying Zora API on startup).

Parameters:

Name	Type	Description	Default
count	int	Current post count from Zora	required

Source code in trackers/zora_achievement_tracker.py

def set_post_counter(self, count: int):
    """
    Set the post counter (called after querying Zora API on startup).

    Args:
        count: Current post count from Zora
    """
    self._post_counter = count
    self._save()
    log.info(f"Zora post counter set to {count}")

trigger(achievement_type, context=None, unique_key=None)

Trigger a Zora achievement.

Parameters:

Name	Type	Description	Default
achievement_type	ZoraAchievementType	The type of achievement	required
context	Optional[Dict[str, Any]]	Additional context for the achievement	None
unique_key	Optional[str]	Optional unique key for achievements that can trigger multiple times (e.g., "first_catch_species_PIDGEY")	None

Returns:

Type	Description
Optional[ZoraAchievement]	ZoraAchievement if newly triggered, None if already exists

Source code in trackers/zora_achievement_tracker.py

def trigger(
    self,
    achievement_type: ZoraAchievementType,
    context: Optional[Dict[str, Any]] = None,
    unique_key: Optional[str] = None,
) -> Optional[ZoraAchievement]:
    """
    Trigger a Zora achievement.

    Args:
        achievement_type: The type of achievement
        context: Additional context for the achievement
        unique_key: Optional unique key for achievements that can trigger multiple times
                    (e.g., "first_catch_species_PIDGEY")

    Returns:
        ZoraAchievement if newly triggered, None if already exists
    """
    # Generate key for this achievement
    key = unique_key or achievement_type.value

    if key in self.triggered:
        log.debug(f"Zora achievement already triggered: {key}")
        return None

    tier = ACHIEVEMENT_TIERS.get(achievement_type, ZoraAchievementTier.MINOR)
    post_number = self.get_next_post_number()

    achievement = ZoraAchievement(
        achievement_type=achievement_type,
        tier=tier,
        triggered_at=datetime.now().isoformat(),
        context=context or {},
        post_number=post_number,
    )

    self.triggered[key] = achievement
    self._pending_achievement = achievement
    self._cycles_since_last_post = 0
    self._save()

    log.info(
        f"Zora achievement triggered: {achievement.get_title()} "
        f"(tier: {tier.value})"
    )
    return achievement

trigger_from_major_achievement(achievement_type_value, context=None)

Trigger a Zora achievement from a major achievement (from AchievementTracker).

This is called when the main AchievementTracker triggers a major achievement that should also be posted to Zora.

Parameters:

Name	Type	Description	Default
achievement_type_value	str	The achievement type value string	required
context	Optional[Dict[str, Any]]	Achievement context	None

Returns:

Type	Description
Optional[ZoraAchievement]	ZoraAchievement if triggered

Source code in trackers/zora_achievement_tracker.py

def trigger_from_major_achievement(
    self,
    achievement_type_value: str,
    context: Optional[Dict[str, Any]] = None,
) -> Optional[ZoraAchievement]:
    """
    Trigger a Zora achievement from a major achievement (from AchievementTracker).

    This is called when the main AchievementTracker triggers a major achievement
    that should also be posted to Zora.

    Args:
        achievement_type_value: The achievement type value string
        context: Achievement context

    Returns:
        ZoraAchievement if triggered
    """
    try:
        zora_type = ZoraAchievementType(achievement_type_value)
    except ValueError:
        log.warning(f"Unknown achievement type for Zora: {achievement_type_value}")
        return None

    return self.trigger(zora_type, context=context)

ZoraAchievementType

Bases: Enum

All achievement types for Zora posting.

Major achievements match the existing AchievementType values. Minor achievements are new and Zora-specific.

Source code in trackers/zora_achievement_tracker.py

class ZoraAchievementType(Enum):
    """
    All achievement types for Zora posting.

    Major achievements match the existing AchievementType values.
    Minor achievements are new and Zora-specific.
    """

    # === MAJOR (Twitter + Zora, ComfyUI image, with cross-promo footer) ===
    STREAM_START = "stream_start"
    BADGE_BOULDER = "badge_boulder"
    BADGE_CASCADE = "badge_cascade"
    BADGE_THUNDER = "badge_thunder"
    BADGE_RAINBOW = "badge_rainbow"
    BADGE_SOUL = "badge_soul"
    BADGE_MARSH = "badge_marsh"
    BADGE_VOLCANO = "badge_volcano"
    BADGE_EARTH = "badge_earth"
    STARTER_EVOLUTION_1 = "starter_evolution_1"
    STARTER_EVOLUTION_2 = "starter_evolution_2"
    LEGENDARY_ARTICUNO = "legendary_articuno"
    LEGENDARY_ZAPDOS = "legendary_zapdos"
    LEGENDARY_MOLTRES = "legendary_moltres"
    LEGENDARY_MEWTWO = "legendary_mewtwo"
    POKEMON_CHAMPION = "pokemon_champion"
    FIRST_POKEMON = "first_pokemon"
    FIRST_CATCH = "first_catch"

    # === MINOR (Zora only, screenshot, no footer) ===
    # Route/Location milestones
    NEW_ROUTE = "new_route"
    NEW_CITY = "new_city"
    NEW_DUNGEON = "new_dungeon"
    DUNGEON_EXIT = "dungeon_exit"

    # Pokemon collection (first catch of each species)
    FIRST_CATCH_SPECIES = "first_catch_species"
    POKEMON_EVOLVED = "pokemon_evolved"  # Non-starter evolutions
    TEAM_FULL = "team_full"
    POKEDEX_MILESTONE = "pokedex_milestone"  # Every 10 entries

    # Battle milestones
    RIVAL_DEFEATED = "rival_defeated"
    ROCKET_DEFEATED = "rocket_defeated"

    # Items/Progress
    HM_OBTAINED = "hm_obtained"
    KEY_ITEM = "key_item"

    # === PROGRESS (Zora only, gallery, with footer - fallback) ===
    CYCLE_CHECKPOINT = "cycle_checkpoint"

Functions

get_zora_achievement_tracker(storage_path='data/zora_achievements.json', fallback_cycles=60)

Get or create the singleton Zora achievement tracker.

Source code in trackers/zora_achievement_tracker.py

def get_zora_achievement_tracker(
    storage_path: str = "data/zora_achievements.json",
    fallback_cycles: int = 60,
) -> ZoraAchievementTracker:
    """Get or create the singleton Zora achievement tracker."""
    global _zora_tracker
    if _zora_tracker is None:
        _zora_tracker = ZoraAchievementTracker(
            storage_path=storage_path,
            fallback_cycles=fallback_cycles,
        )
    return _zora_tracker

Goal Tracker

trackers.goal_tracker

Goal Tracker Module for Pokemon LLM

Implements hierarchical goal stack that persists across summaries. Provides structured goal context for LLM decision making.

Classes

Goal dataclass

A single goal with priority and status

Source code in trackers/goal_tracker.py

@dataclass
class Goal:
    """A single goal with priority and status"""
    id: str
    description: str
    priority: int  # 1-4 from GoalPriority
    status: str = "pending"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    completed_at: Optional[str] = None
    parent_id: Optional[str] = None  # For sub-goals
    context: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict) -> 'Goal':
        return cls(**data)

GoalPriority

Bases: Enum

Goal priority levels - QUESTS are PRIMARY, healing is SECONDARY

Priority Order: 1. CRITICAL - Main story quests (deliver Oak's Parcel, beat Gyms) 2. HIGH - Quest item objectives (pick up items, find NPCs) 3. MEDIUM - Urgent needs (heal at Pokemon Center when low HP) 4. LOW - Optional (extra exploration, training)

Source code in trackers/goal_tracker.py

class GoalPriority(Enum):
    """
    Goal priority levels - QUESTS are PRIMARY, healing is SECONDARY

    Priority Order:
    1. CRITICAL - Main story quests (deliver Oak's Parcel, beat Gyms)
    2. HIGH - Quest item objectives (pick up items, find NPCs)
    3. MEDIUM - Urgent needs (heal at Pokemon Center when low HP)
    4. LOW - Optional (extra exploration, training)
    """
    CRITICAL = 1    # Main story quests (Oak's Parcel delivery, Gym battles)
    HIGH = 2        # Quest-related tasks (go to target location)
    MEDIUM = 3      # Urgent but not quest-critical (healing when HP < 30%)
    LOW = 4         # Optional (explore, train, catch Pokemon)

GoalStatus

Bases: Enum

Goal completion status

Source code in trackers/goal_tracker.py

class GoalStatus(Enum):
    """Goal completion status"""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

GoalTracker

Manages hierarchical goal stack for Pokemon gameplay.

Goal hierarchy: - Main Quest: Beat the Elite Four - Milestone: Get 8 badges - Current: Beat Pewter Gym - Micro: Navigate to Pewter City - Immediate: Exit current building

Source code in trackers/goal_tracker.py

class GoalTracker:
    """
    Manages hierarchical goal stack for Pokemon gameplay.

    Goal hierarchy:
    - Main Quest: Beat the Elite Four
      - Milestone: Get 8 badges
        - Current: Beat Pewter Gym
          - Micro: Navigate to Pewter City
            - Immediate: Exit current building
    """

    def __init__(self, storage_path: str = "data/game_goals.json"):
        self.storage_path = storage_path
        self.goals: Dict[str, Goal] = {}
        self.goal_stack: List[str] = []  # IDs in priority order
        self.failure_history: List[Dict] = []  # Recent failures for replay
        self._load()

    def add_goal(
        self, 
        description: str, 
        priority: GoalPriority = GoalPriority.MEDIUM,
        parent_id: Optional[str] = None,
        context: Optional[Dict] = None
    ) -> str:
        """Add a new goal to the tracker"""
        goal_id = f"goal_{len(self.goals) + 1}_{int(datetime.now().timestamp())}"

        goal = Goal(
            id=goal_id,
            description=description,
            priority=priority.value,
            parent_id=parent_id,
            context=context or {}
        )

        self.goals[goal_id] = goal
        self._update_stack()
        self._save()

        return goal_id

    def update_goal(self, goal_id: str, status: GoalStatus) -> bool:
        """Update goal status"""
        if goal_id not in self.goals:
            return False

        self.goals[goal_id].status = status.value
        if status == GoalStatus.COMPLETED:
            self.goals[goal_id].completed_at = datetime.now().isoformat()

        self._update_stack()
        self._save()
        return True

    def complete_goal(self, goal_id: str) -> bool:
        """Mark a goal as completed"""
        return self.update_goal(goal_id, GoalStatus.COMPLETED)

    def fail_goal(self, goal_id: str, reason: str = "") -> bool:
        """Mark a goal as failed and record for replay"""
        if goal_id not in self.goals:
            return False

        goal = self.goals[goal_id]
        self.failure_history.append({
            "goal": goal.description,
            "reason": reason,
            "timestamp": datetime.now().isoformat()
        })
        # Keep only last 5 failures
        self.failure_history = self.failure_history[-5:]

        return self.update_goal(goal_id, GoalStatus.FAILED)

    def get_current_goal(self) -> Optional[Goal]:
        """Get the highest priority active goal"""
        for goal_id in self.goal_stack:
            goal = self.goals.get(goal_id)
            if goal and goal.status in ["pending", "in_progress"]:
                return goal
        return None

    def get_goal_hierarchy(self) -> List[Goal]:
        """Get goals from highest to lowest priority (main quest → micro)"""
        hierarchy = []
        for goal_id in self.goal_stack:
            goal = self.goals.get(goal_id)
            if goal and goal.status in ["pending", "in_progress"]:
                hierarchy.append(goal)
        # Sort by priority (1=critical first)
        return sorted(hierarchy, key=lambda g: g.priority)[:4]

    def record_failure(self, action: str, position: tuple, reason: str):
        """Record a failed action for replay context"""
        self.failure_history.append({
            "action": action,
            "position": position,
            "reason": reason,
            "timestamp": datetime.now().isoformat()
        })
        self.failure_history = self.failure_history[-5:]
        self._save()

    def get_failure_context(self) -> str:
        """Get recent failures as context for LLM"""
        if not self.failure_history:
            return ""

        lines = ["⚠️ RECENT FAILURES (don't repeat these):"]
        for i, failure in enumerate(self.failure_history[-3:], 1):
            if "goal" in failure:
                lines.append(f"  {i}. Goal failed: {failure['goal']} - {failure.get('reason', 'unknown')}")
            else:
                lines.append(f"  {i}. Action '{failure['action']}' failed at {failure.get('position', '?')} - {failure.get('reason', 'unknown')}")

        return "\n".join(lines)

    def get_context_for_llm(self, team_size: int = 0, team_pokemon: list = None) -> str:
        """Generate compact goal context for LLM injection with progress tracking"""
        lines = ["🎯 GOAL STATUS:"]

        # Show team status first - this is critical for progress awareness
        if team_size > 0:
            pokemon_names = ", ".join(team_pokemon[:3]) if team_pokemon else "your Pokemon"
            lines.append(f"  ✓ OBTAINED STARTER: You have {team_size} Pokemon ({pokemon_names})")
            lines.append(f"  → The 'get first Pokemon' objective is COMPLETE - do NOT repeat it!")
        else:
            lines.append(f"  ○ NO POKEMON YET: You need to get your starter from Professor Oak")

        # Show recently completed goals (helps LLM remember progress)
        completed = self.get_completed_goals(limit=3)
        if completed:
            lines.append("  📋 COMPLETED:")
            for goal in completed:
                lines.append(f"    ✓ {goal.description}")

        # Show active goals hierarchy
        hierarchy = self.get_goal_hierarchy()
        if hierarchy:
            lines.append("  📍 ACTIVE GOALS:")
            labels = ["MAIN", "CURRENT", "MICRO", "IMMEDIATE"]
            for i, goal in enumerate(hierarchy):
                label = labels[i] if i < len(labels) else f"#{i+1}"
                status_icon = "→" if goal.status == "in_progress" else "○"
                lines.append(f"    {status_icon} {label}: {goal.description}")

        # Add failure context if any
        failure_ctx = self.get_failure_context()
        if failure_ctx:
            lines.append(failure_ctx)

        return "\n".join(lines)

    def get_completed_goals(self, limit: int = 5) -> List[Goal]:
        """Get recently completed goals"""
        completed = [
            g for g in self.goals.values() 
            if g.status == "completed"
        ]
        # Sort by completion time, most recent first
        completed.sort(key=lambda g: g.completed_at or "", reverse=True)
        return completed[:limit]

    def complete_goal_by_keyword(self, keyword: str) -> bool:
        """Find and complete a goal containing the keyword"""
        for goal_id, goal in self.goals.items():
            if keyword.lower() in goal.description.lower() and goal.status != "completed":
                self.complete_goal(goal_id)
                return True
        return False

    def add_quest_goal(
        self,
        quest_id: str,
        description: str,
        target_location: Optional[str] = None,
        target_npc: Optional[str] = None,
        priority: GoalPriority = GoalPriority.CRITICAL  # Quests are CRITICAL priority
    ) -> Optional[str]:
        """
        Add a goal derived from a quest item pickup.

        Quest goals are CRITICAL priority because the game's story progression
        depends on completing them. Healing can wait, but the Parcel must be delivered.
        """
        # Check if we already have a goal for this quest
        for goal in self.goals.values():
            if goal.context.get("quest_id") == quest_id and goal.status != "completed":
                return goal.id  # Already have this quest goal

        context = {
            "type": "quest",
            "quest_id": quest_id,
            "target_location": target_location,
            "target_npc": target_npc
        }

        # Get main quest ID as parent
        main_quest_id = self._get_main_quest_id()

        return self.add_goal(
            description=description,
            priority=priority,
            parent_id=main_quest_id,
            context=context
        )

    def _get_main_quest_id(self) -> Optional[str]:
        """Get the main quest goal ID."""
        for goal_id, goal in self.goals.items():
            if goal.context.get("type") == "main_quest":
                return goal_id
        return None

    def has_quest_goal(self, quest_id: str) -> bool:
        """Check if we already have an active goal for this quest."""
        for goal in self.goals.values():
            if (goal.context.get("quest_id") == quest_id and 
                goal.status in ["pending", "in_progress"]):
                return True
        return False

    def initialize_default_goals(self):
        """Set up initial game goals"""
        # Main quest
        main_id = self.add_goal(
            "Complete Pokemon Red - Become Champion",
            GoalPriority.LOW,
            context={"type": "main_quest"}
        )

        # Current milestone
        self.add_goal(
            "Get first Pokemon and start journey",
            GoalPriority.HIGH,
            parent_id=main_id,
            context={"type": "milestone"}
        )

    def _update_stack(self):
        """Rebuild priority stack"""
        active = [
            g for g in self.goals.values() 
            if g.status in ["pending", "in_progress"]
        ]
        # Sort by priority (lower number = higher priority)
        active.sort(key=lambda g: g.priority)
        self.goal_stack = [g.id for g in active]

    def _save(self):
        """Save goals to file"""
        try:
            data = {
                "goals": {gid: g.to_dict() for gid, g in self.goals.items()},
                "stack": self.goal_stack,
                "failures": self.failure_history
            }
            with open(self.storage_path, 'w') as f:
                json.dump(data, f, indent=2)
        except Exception as e:
            print(f"Error saving goals: {e}")

    def _load(self):
        """Load goals from file"""
        if not os.path.exists(self.storage_path):
            return

        try:
            with open(self.storage_path, 'r') as f:
                data = json.load(f)

            self.goals = {
                gid: Goal.from_dict(gdata) 
                for gid, gdata in data.get("goals", {}).items()
            }
            self.goal_stack = data.get("stack", [])
            self.failure_history = data.get("failures", [])
        except Exception as e:
            print(f"Error loading goals: {e}")

Functions

add_goal(description, priority=GoalPriority.MEDIUM, parent_id=None, context=None)

Add a new goal to the tracker

Source code in trackers/goal_tracker.py

def add_goal(
    self, 
    description: str, 
    priority: GoalPriority = GoalPriority.MEDIUM,
    parent_id: Optional[str] = None,
    context: Optional[Dict] = None
) -> str:
    """Add a new goal to the tracker"""
    goal_id = f"goal_{len(self.goals) + 1}_{int(datetime.now().timestamp())}"

    goal = Goal(
        id=goal_id,
        description=description,
        priority=priority.value,
        parent_id=parent_id,
        context=context or {}
    )

    self.goals[goal_id] = goal
    self._update_stack()
    self._save()

    return goal_id

add_quest_goal(quest_id, description, target_location=None, target_npc=None, priority=GoalPriority.CRITICAL)

Add a goal derived from a quest item pickup.

Quest goals are CRITICAL priority because the game's story progression depends on completing them. Healing can wait, but the Parcel must be delivered.

Source code in trackers/goal_tracker.py

def add_quest_goal(
    self,
    quest_id: str,
    description: str,
    target_location: Optional[str] = None,
    target_npc: Optional[str] = None,
    priority: GoalPriority = GoalPriority.CRITICAL  # Quests are CRITICAL priority
) -> Optional[str]:
    """
    Add a goal derived from a quest item pickup.

    Quest goals are CRITICAL priority because the game's story progression
    depends on completing them. Healing can wait, but the Parcel must be delivered.
    """
    # Check if we already have a goal for this quest
    for goal in self.goals.values():
        if goal.context.get("quest_id") == quest_id and goal.status != "completed":
            return goal.id  # Already have this quest goal

    context = {
        "type": "quest",
        "quest_id": quest_id,
        "target_location": target_location,
        "target_npc": target_npc
    }

    # Get main quest ID as parent
    main_quest_id = self._get_main_quest_id()

    return self.add_goal(
        description=description,
        priority=priority,
        parent_id=main_quest_id,
        context=context
    )

complete_goal(goal_id)

Mark a goal as completed

Source code in trackers/goal_tracker.py

def complete_goal(self, goal_id: str) -> bool:
    """Mark a goal as completed"""
    return self.update_goal(goal_id, GoalStatus.COMPLETED)

complete_goal_by_keyword(keyword)

Find and complete a goal containing the keyword

Source code in trackers/goal_tracker.py

def complete_goal_by_keyword(self, keyword: str) -> bool:
    """Find and complete a goal containing the keyword"""
    for goal_id, goal in self.goals.items():
        if keyword.lower() in goal.description.lower() and goal.status != "completed":
            self.complete_goal(goal_id)
            return True
    return False

fail_goal(goal_id, reason='')

Mark a goal as failed and record for replay

Source code in trackers/goal_tracker.py

def fail_goal(self, goal_id: str, reason: str = "") -> bool:
    """Mark a goal as failed and record for replay"""
    if goal_id not in self.goals:
        return False

    goal = self.goals[goal_id]
    self.failure_history.append({
        "goal": goal.description,
        "reason": reason,
        "timestamp": datetime.now().isoformat()
    })
    # Keep only last 5 failures
    self.failure_history = self.failure_history[-5:]

    return self.update_goal(goal_id, GoalStatus.FAILED)

get_completed_goals(limit=5)

Get recently completed goals

Source code in trackers/goal_tracker.py

def get_completed_goals(self, limit: int = 5) -> List[Goal]:
    """Get recently completed goals"""
    completed = [
        g for g in self.goals.values() 
        if g.status == "completed"
    ]
    # Sort by completion time, most recent first
    completed.sort(key=lambda g: g.completed_at or "", reverse=True)
    return completed[:limit]

get_context_for_llm(team_size=0, team_pokemon=None)

Generate compact goal context for LLM injection with progress tracking

Source code in trackers/goal_tracker.py

def get_context_for_llm(self, team_size: int = 0, team_pokemon: list = None) -> str:
    """Generate compact goal context for LLM injection with progress tracking"""
    lines = ["🎯 GOAL STATUS:"]

    # Show team status first - this is critical for progress awareness
    if team_size > 0:
        pokemon_names = ", ".join(team_pokemon[:3]) if team_pokemon else "your Pokemon"
        lines.append(f"  ✓ OBTAINED STARTER: You have {team_size} Pokemon ({pokemon_names})")
        lines.append(f"  → The 'get first Pokemon' objective is COMPLETE - do NOT repeat it!")
    else:
        lines.append(f"  ○ NO POKEMON YET: You need to get your starter from Professor Oak")

    # Show recently completed goals (helps LLM remember progress)
    completed = self.get_completed_goals(limit=3)
    if completed:
        lines.append("  📋 COMPLETED:")
        for goal in completed:
            lines.append(f"    ✓ {goal.description}")

    # Show active goals hierarchy
    hierarchy = self.get_goal_hierarchy()
    if hierarchy:
        lines.append("  📍 ACTIVE GOALS:")
        labels = ["MAIN", "CURRENT", "MICRO", "IMMEDIATE"]
        for i, goal in enumerate(hierarchy):
            label = labels[i] if i < len(labels) else f"#{i+1}"
            status_icon = "→" if goal.status == "in_progress" else "○"
            lines.append(f"    {status_icon} {label}: {goal.description}")

    # Add failure context if any
    failure_ctx = self.get_failure_context()
    if failure_ctx:
        lines.append(failure_ctx)

    return "\n".join(lines)

get_current_goal()

Get the highest priority active goal

Source code in trackers/goal_tracker.py

def get_current_goal(self) -> Optional[Goal]:
    """Get the highest priority active goal"""
    for goal_id in self.goal_stack:
        goal = self.goals.get(goal_id)
        if goal and goal.status in ["pending", "in_progress"]:
            return goal
    return None

get_failure_context()

Get recent failures as context for LLM

Source code in trackers/goal_tracker.py

def get_failure_context(self) -> str:
    """Get recent failures as context for LLM"""
    if not self.failure_history:
        return ""

    lines = ["⚠️ RECENT FAILURES (don't repeat these):"]
    for i, failure in enumerate(self.failure_history[-3:], 1):
        if "goal" in failure:
            lines.append(f"  {i}. Goal failed: {failure['goal']} - {failure.get('reason', 'unknown')}")
        else:
            lines.append(f"  {i}. Action '{failure['action']}' failed at {failure.get('position', '?')} - {failure.get('reason', 'unknown')}")

    return "\n".join(lines)

get_goal_hierarchy()

Get goals from highest to lowest priority (main quest → micro)

Source code in trackers/goal_tracker.py

def get_goal_hierarchy(self) -> List[Goal]:
    """Get goals from highest to lowest priority (main quest → micro)"""
    hierarchy = []
    for goal_id in self.goal_stack:
        goal = self.goals.get(goal_id)
        if goal and goal.status in ["pending", "in_progress"]:
            hierarchy.append(goal)
    # Sort by priority (1=critical first)
    return sorted(hierarchy, key=lambda g: g.priority)[:4]

has_quest_goal(quest_id)

Check if we already have an active goal for this quest.

Source code in trackers/goal_tracker.py

def has_quest_goal(self, quest_id: str) -> bool:
    """Check if we already have an active goal for this quest."""
    for goal in self.goals.values():
        if (goal.context.get("quest_id") == quest_id and 
            goal.status in ["pending", "in_progress"]):
            return True
    return False

initialize_default_goals()

Set up initial game goals

Source code in trackers/goal_tracker.py

def initialize_default_goals(self):
    """Set up initial game goals"""
    # Main quest
    main_id = self.add_goal(
        "Complete Pokemon Red - Become Champion",
        GoalPriority.LOW,
        context={"type": "main_quest"}
    )

    # Current milestone
    self.add_goal(
        "Get first Pokemon and start journey",
        GoalPriority.HIGH,
        parent_id=main_id,
        context={"type": "milestone"}
    )

record_failure(action, position, reason)

Record a failed action for replay context

Source code in trackers/goal_tracker.py

def record_failure(self, action: str, position: tuple, reason: str):
    """Record a failed action for replay context"""
    self.failure_history.append({
        "action": action,
        "position": position,
        "reason": reason,
        "timestamp": datetime.now().isoformat()
    })
    self.failure_history = self.failure_history[-5:]
    self._save()

update_goal(goal_id, status)

Update goal status

Source code in trackers/goal_tracker.py

def update_goal(self, goal_id: str, status: GoalStatus) -> bool:
    """Update goal status"""
    if goal_id not in self.goals:
        return False

    self.goals[goal_id].status = status.value
    if status == GoalStatus.COMPLETED:
        self.goals[goal_id].completed_at = datetime.now().isoformat()

    self._update_stack()
    self._save()
    return True

Coordinate Tracker

trackers.coordinate_tracker

Coordinate Tracker for Enhanced Navigation

Tracks player world coordinates over time to detect loops and measure progress toward navigation targets. Persists to disk for run continuity.

Classes

CoordinateTracker

Tracks player world coordinates to detect loops and measure navigation progress.

Key features: - Maintains history of last N positions - Detects when player visits same tile 3+ times in 10 cycles (loop) - Tracks distance to target coordinate to measure progress - Persists to disk for run continuity

Source code in trackers/coordinate_tracker.py

class CoordinateTracker:
    """
    Tracks player world coordinates to detect loops and measure navigation progress.

    Key features:
    - Maintains history of last N positions
    - Detects when player visits same tile 3+ times in 10 cycles (loop)
    - Tracks distance to target coordinate to measure progress
    - Persists to disk for run continuity
    """

    def __init__(self, storage_path: str = "data/coordinate_history.json", 
                 max_history: int = 10, reset_on_start: bool = False):
        """
        Initialize the coordinate tracker.

        Args:
            storage_path: Path to JSON file for persistence
            max_history: Number of cycles to keep in history
            reset_on_start: Whether to clear history on initialization
        """
        self.storage_path = storage_path
        self.max_history = max_history

        # Position history (newest first)
        self.history: List[PositionRecord] = []

        # Current navigation target
        self.target: Optional[NavigationTarget] = None

        # Load from disk if not resetting
        if not reset_on_start and os.path.exists(storage_path):
            self._load()
            log.info(f"📍 Coordinate tracker loaded ({len(self.history)} positions, target: {self.target is not None})")
        else:
            log.info(f"📍 Coordinate tracker initialized (fresh start)")

    def _load(self) -> None:
        """Load state from disk."""
        try:
            with open(self.storage_path, 'r') as f:
                data = json.load(f)

            # Load history
            self.history = [
                PositionRecord(**rec) for rec in data.get("history", [])
            ]

            # Load target
            if data.get("target"):
                self.target = NavigationTarget(**data["target"])

        except Exception as e:
            log.error(f"Failed to load coordinate history: {e}")
            self.history = []
            self.target = None

    def _save(self) -> None:
        """Save state to disk."""
        try:
            data = {
                "history": [asdict(rec) for rec in self.history],
                "target": asdict(self.target) if self.target else None,
                "last_updated": datetime.now().isoformat()
            }
            with open(self.storage_path, 'w') as f:
                json.dump(data, f, indent=2)
        except Exception as e:
            log.error(f"Failed to save coordinate history: {e}")

    def add_position(self, cycle: int, x: int, y: int, map_name: str) -> None:
        """
        Record current player position.

        Args:
            cycle: Current cycle number
            x: World X coordinate
            y: World Y coordinate  
            map_name: Current map name
        """
        record = PositionRecord(
            cycle=cycle,
            x=x,
            y=y,
            map_name=map_name,
            timestamp=datetime.now().isoformat()
        )

        # Add to front, trim to max
        self.history.insert(0, record)
        while len(self.history) > self.max_history:
            self.history.pop()

        # Update target progress if we have one
        if self.target and self.target.map_name == map_name:
            current_dist = self._calculate_distance(x, y, self.target.x, self.target.y)

            if self.target.last_distance is not None:
                if current_dist >= self.target.last_distance:
                    self.target.cycles_without_progress += 1
                else:
                    self.target.cycles_without_progress = 0

            self.target.last_distance = current_dist

        log.debug(f"📍 Position recorded: cycle {cycle} -> ({x}, {y}) on {map_name}")
        self._save()

    def _calculate_distance(self, x1: int, y1: int, x2: int, y2: int) -> float:
        """Calculate Manhattan distance between two points."""
        return abs(x2 - x1) + abs(y2 - y1)

    def detect_loop(self) -> Optional[str]:
        """
        Detect if player is moving in circles.

        Returns description if loop detected (same tile 3+ times in history),
        None otherwise.
        """
        if len(self.history) < 3:
            return None

        # Count occurrences of each (map, x, y) tuple
        tile_counts = Counter(
            (rec.map_name, rec.x, rec.y) for rec in self.history
        )

        # Find any tiles visited 3+ times
        loops = [(tile, count) for tile, count in tile_counts.items() if count >= 3]

        if loops:
            # Sort by count, report worst loop
            loops.sort(key=lambda x: -x[1])
            (map_name, x, y), count = loops[0]
            return f"LOOP DETECTED: Visited ({x}, {y}) on {map_name} {count} times in last {len(self.history)} cycles"

        return None

    def get_progress_toward_target(self) -> Dict[str, Any]:
        """
        Get information about progress toward current target.

        Returns dict with:
        - has_target: bool
        - target_info: Target details if set
        - distance: Current distance to target
        - progress: "approaching", "stuck", "no_progress", or "wrong_map"
        - recommendation: What to do
        """
        if not self.target:
            return {
                "has_target": False,
                "recommendation": "Set a navigation target to track progress"
            }

        if not self.history:
            return {
                "has_target": True,
                "target_info": f"({self.target.x}, {self.target.y}) on {self.target.map_name}: {self.target.reason}",
                "recommendation": "Position data not yet available"
            }

        current = self.history[0]

        # Check if on wrong map
        if current.map_name != self.target.map_name:
            return {
                "has_target": True,
                "target_info": f"({self.target.x}, {self.target.y}) on {self.target.map_name}: {self.target.reason}",
                "progress": "wrong_map",
                "current_map": current.map_name,
                "target_map": self.target.map_name,
                "recommendation": f"Need to travel to {self.target.map_name} first"
            }

        distance = self._calculate_distance(current.x, current.y, self.target.x, self.target.y)

        # Determine progress status
        if self.target.cycles_without_progress >= 5:
            progress = "stuck"
            recommendation = f"STUCK for {self.target.cycles_without_progress} cycles. Consider invalidating target and trying a different route."
        elif self.target.cycles_without_progress >= 3:
            progress = "no_progress"
            recommendation = "Not making progress. Try going around obstacles or taking a different path."
        else:
            progress = "approaching"
            recommendation = f"Continue toward target ({distance} tiles away)"

        return {
            "has_target": True,
            "target_info": f"({self.target.x}, {self.target.y}) on {self.target.map_name}: {self.target.reason}",
            "distance": distance,
            "cycles_without_progress": self.target.cycles_without_progress,
            "progress": progress,
            "recommendation": recommendation
        }

    def set_target(self, x: int, y: int, map_name: str, reason: str, cycle: int) -> None:
        """
        Set a new navigation target.

        Args:
            x: Target X coordinate
            y: Target Y coordinate
            map_name: Target map name
            reason: Why we're going there
            cycle: Current cycle when target was set
        """
        self.target = NavigationTarget(
            x=x,
            y=y,
            map_name=map_name,
            reason=reason,
            set_at_cycle=cycle,
            set_at_time=datetime.now().isoformat(),
            last_distance=None,
            cycles_without_progress=0
        )
        log.info(f"🎯 Navigation target set: ({x}, {y}) on {map_name} - {reason}")
        self._save()

    def invalidate_target(self, reason: str) -> None:
        """
        Clear the current target when stuck or reached.

        Args:
            reason: Why the target is being invalidated
        """
        if self.target:
            log.info(f"❌ Target invalidated: {reason} (was: ({self.target.x}, {self.target.y}) on {self.target.map_name})")
        self.target = None
        self._save()

    def check_target_reached(self, threshold: int = 2) -> bool:
        """
        Check if we've reached the target (within threshold tiles).

        Returns True and invalidates target if reached.
        """
        if not self.target or not self.history:
            return False

        current = self.history[0]
        if current.map_name != self.target.map_name:
            return False

        distance = self._calculate_distance(
            current.x, current.y, 
            self.target.x, self.target.y
        )

        if distance <= threshold:
            log.info(f"✅ Target reached! Distance: {distance}")
            self.invalidate_target("Target reached")
            return True

        return False

    def get_context_summary(self) -> str:
        """
        Return formatted context string for LLM.

        Includes:
        - Recent position history
        - Loop detection results
        - Target progress info
        """
        lines = []

        # Position history
        if self.history:
            lines.append("COORDINATE HISTORY (last 10 cycles):")
            for rec in self.history[:10]:
                lines.append(f"  Cycle {rec.cycle}: ({rec.x}, {rec.y}) on {rec.map_name}")

        # Loop detection
        loop = self.detect_loop()
        if loop:
            lines.append(f"\n⚠️ {loop}")

        # Target progress
        progress = self.get_progress_toward_target()
        if progress.get("has_target"):
            lines.append(f"\n🎯 NAVIGATION TARGET: {progress['target_info']}")
            if "distance" in progress:
                lines.append(f"   Distance: {progress['distance']} tiles")
            lines.append(f"   Status: {progress.get('progress', 'unknown')}")
            lines.append(f"   {progress['recommendation']}")

        return "\n".join(lines)

    def get_position_at_cycle(self, cycle: int) -> Optional[Tuple[int, int, str]]:
        """Get position at a specific cycle if available."""
        for rec in self.history:
            if rec.cycle == cycle:
                return (rec.x, rec.y, rec.map_name)
        return None

Functions

__init__(storage_path='data/coordinate_history.json', max_history=10, reset_on_start=False)

Initialize the coordinate tracker.

Parameters:

Name	Type	Description	Default
storage_path	str	Path to JSON file for persistence	'data/coordinate_history.json'
max_history	int	Number of cycles to keep in history	10
reset_on_start	bool	Whether to clear history on initialization	False

Source code in trackers/coordinate_tracker.py

def __init__(self, storage_path: str = "data/coordinate_history.json", 
             max_history: int = 10, reset_on_start: bool = False):
    """
    Initialize the coordinate tracker.

    Args:
        storage_path: Path to JSON file for persistence
        max_history: Number of cycles to keep in history
        reset_on_start: Whether to clear history on initialization
    """
    self.storage_path = storage_path
    self.max_history = max_history

    # Position history (newest first)
    self.history: List[PositionRecord] = []

    # Current navigation target
    self.target: Optional[NavigationTarget] = None

    # Load from disk if not resetting
    if not reset_on_start and os.path.exists(storage_path):
        self._load()
        log.info(f"📍 Coordinate tracker loaded ({len(self.history)} positions, target: {self.target is not None})")
    else:
        log.info(f"📍 Coordinate tracker initialized (fresh start)")

add_position(cycle, x, y, map_name)

Record current player position.

Parameters:

Name	Type	Description	Default
cycle	int	Current cycle number	required
x	int	World X coordinate	required
y	int	World Y coordinate	required
map_name	str	Current map name	required

Source code in trackers/coordinate_tracker.py

def add_position(self, cycle: int, x: int, y: int, map_name: str) -> None:
    """
    Record current player position.

    Args:
        cycle: Current cycle number
        x: World X coordinate
        y: World Y coordinate  
        map_name: Current map name
    """
    record = PositionRecord(
        cycle=cycle,
        x=x,
        y=y,
        map_name=map_name,
        timestamp=datetime.now().isoformat()
    )

    # Add to front, trim to max
    self.history.insert(0, record)
    while len(self.history) > self.max_history:
        self.history.pop()

    # Update target progress if we have one
    if self.target and self.target.map_name == map_name:
        current_dist = self._calculate_distance(x, y, self.target.x, self.target.y)

        if self.target.last_distance is not None:
            if current_dist >= self.target.last_distance:
                self.target.cycles_without_progress += 1
            else:
                self.target.cycles_without_progress = 0

        self.target.last_distance = current_dist

    log.debug(f"📍 Position recorded: cycle {cycle} -> ({x}, {y}) on {map_name}")
    self._save()

check_target_reached(threshold=2)

Check if we've reached the target (within threshold tiles).

Returns True and invalidates target if reached.

Source code in trackers/coordinate_tracker.py

def check_target_reached(self, threshold: int = 2) -> bool:
    """
    Check if we've reached the target (within threshold tiles).

    Returns True and invalidates target if reached.
    """
    if not self.target or not self.history:
        return False

    current = self.history[0]
    if current.map_name != self.target.map_name:
        return False

    distance = self._calculate_distance(
        current.x, current.y, 
        self.target.x, self.target.y
    )

    if distance <= threshold:
        log.info(f"✅ Target reached! Distance: {distance}")
        self.invalidate_target("Target reached")
        return True

    return False

detect_loop()

Detect if player is moving in circles.

Returns description if loop detected (same tile 3+ times in history), None otherwise.

Source code in trackers/coordinate_tracker.py

def detect_loop(self) -> Optional[str]:
    """
    Detect if player is moving in circles.

    Returns description if loop detected (same tile 3+ times in history),
    None otherwise.
    """
    if len(self.history) < 3:
        return None

    # Count occurrences of each (map, x, y) tuple
    tile_counts = Counter(
        (rec.map_name, rec.x, rec.y) for rec in self.history
    )

    # Find any tiles visited 3+ times
    loops = [(tile, count) for tile, count in tile_counts.items() if count >= 3]

    if loops:
        # Sort by count, report worst loop
        loops.sort(key=lambda x: -x[1])
        (map_name, x, y), count = loops[0]
        return f"LOOP DETECTED: Visited ({x}, {y}) on {map_name} {count} times in last {len(self.history)} cycles"

    return None

get_context_summary()

Return formatted context string for LLM.

Includes: - Recent position history - Loop detection results - Target progress info

Source code in trackers/coordinate_tracker.py

def get_context_summary(self) -> str:
    """
    Return formatted context string for LLM.

    Includes:
    - Recent position history
    - Loop detection results
    - Target progress info
    """
    lines = []

    # Position history
    if self.history:
        lines.append("COORDINATE HISTORY (last 10 cycles):")
        for rec in self.history[:10]:
            lines.append(f"  Cycle {rec.cycle}: ({rec.x}, {rec.y}) on {rec.map_name}")

    # Loop detection
    loop = self.detect_loop()
    if loop:
        lines.append(f"\n⚠️ {loop}")

    # Target progress
    progress = self.get_progress_toward_target()
    if progress.get("has_target"):
        lines.append(f"\n🎯 NAVIGATION TARGET: {progress['target_info']}")
        if "distance" in progress:
            lines.append(f"   Distance: {progress['distance']} tiles")
        lines.append(f"   Status: {progress.get('progress', 'unknown')}")
        lines.append(f"   {progress['recommendation']}")

    return "\n".join(lines)

get_position_at_cycle(cycle)

Get position at a specific cycle if available.

Source code in trackers/coordinate_tracker.py

def get_position_at_cycle(self, cycle: int) -> Optional[Tuple[int, int, str]]:
    """Get position at a specific cycle if available."""
    for rec in self.history:
        if rec.cycle == cycle:
            return (rec.x, rec.y, rec.map_name)
    return None

get_progress_toward_target()

Get information about progress toward current target.

Returns dict with: - has_target: bool - target_info: Target details if set - distance: Current distance to target - progress: "approaching", "stuck", "no_progress", or "wrong_map" - recommendation: What to do

Source code in trackers/coordinate_tracker.py

def get_progress_toward_target(self) -> Dict[str, Any]:
    """
    Get information about progress toward current target.

    Returns dict with:
    - has_target: bool
    - target_info: Target details if set
    - distance: Current distance to target
    - progress: "approaching", "stuck", "no_progress", or "wrong_map"
    - recommendation: What to do
    """
    if not self.target:
        return {
            "has_target": False,
            "recommendation": "Set a navigation target to track progress"
        }

    if not self.history:
        return {
            "has_target": True,
            "target_info": f"({self.target.x}, {self.target.y}) on {self.target.map_name}: {self.target.reason}",
            "recommendation": "Position data not yet available"
        }

    current = self.history[0]

    # Check if on wrong map
    if current.map_name != self.target.map_name:
        return {
            "has_target": True,
            "target_info": f"({self.target.x}, {self.target.y}) on {self.target.map_name}: {self.target.reason}",
            "progress": "wrong_map",
            "current_map": current.map_name,
            "target_map": self.target.map_name,
            "recommendation": f"Need to travel to {self.target.map_name} first"
        }

    distance = self._calculate_distance(current.x, current.y, self.target.x, self.target.y)

    # Determine progress status
    if self.target.cycles_without_progress >= 5:
        progress = "stuck"
        recommendation = f"STUCK for {self.target.cycles_without_progress} cycles. Consider invalidating target and trying a different route."
    elif self.target.cycles_without_progress >= 3:
        progress = "no_progress"
        recommendation = "Not making progress. Try going around obstacles or taking a different path."
    else:
        progress = "approaching"
        recommendation = f"Continue toward target ({distance} tiles away)"

    return {
        "has_target": True,
        "target_info": f"({self.target.x}, {self.target.y}) on {self.target.map_name}: {self.target.reason}",
        "distance": distance,
        "cycles_without_progress": self.target.cycles_without_progress,
        "progress": progress,
        "recommendation": recommendation
    }

invalidate_target(reason)

Clear the current target when stuck or reached.

Parameters:

Name	Type	Description	Default
reason	str	Why the target is being invalidated	required

Source code in trackers/coordinate_tracker.py

def invalidate_target(self, reason: str) -> None:
    """
    Clear the current target when stuck or reached.

    Args:
        reason: Why the target is being invalidated
    """
    if self.target:
        log.info(f"❌ Target invalidated: {reason} (was: ({self.target.x}, {self.target.y}) on {self.target.map_name})")
    self.target = None
    self._save()

set_target(x, y, map_name, reason, cycle)

Set a new navigation target.

Parameters:

Name	Type	Description	Default
x	int	Target X coordinate	required
y	int	Target Y coordinate	required
map_name	str	Target map name	required
reason	str	Why we're going there	required
cycle	int	Current cycle when target was set	required

Source code in trackers/coordinate_tracker.py

def set_target(self, x: int, y: int, map_name: str, reason: str, cycle: int) -> None:
    """
    Set a new navigation target.

    Args:
        x: Target X coordinate
        y: Target Y coordinate
        map_name: Target map name
        reason: Why we're going there
        cycle: Current cycle when target was set
    """
    self.target = NavigationTarget(
        x=x,
        y=y,
        map_name=map_name,
        reason=reason,
        set_at_cycle=cycle,
        set_at_time=datetime.now().isoformat(),
        last_distance=None,
        cycles_without_progress=0
    )
    log.info(f"🎯 Navigation target set: ({x}, {y}) on {map_name} - {reason}")
    self._save()

NavigationTarget dataclass

Target coordinate the agent is trying to reach.

Source code in trackers/coordinate_tracker.py

@dataclass 
class NavigationTarget:
    """Target coordinate the agent is trying to reach."""
    x: int
    y: int
    map_name: str
    reason: str
    set_at_cycle: int
    set_at_time: str
    last_distance: Optional[float] = None
    cycles_without_progress: int = 0

PositionRecord dataclass

Record of player position at a cycle.

Source code in trackers/coordinate_tracker.py

@dataclass
class PositionRecord:
    """Record of player position at a cycle."""
    cycle: int
    x: int
    y: int
    map_name: str
    timestamp: str

Exploration Tracker

trackers.exploration_tracker

Exploration Tracking System for Pokemon LLM Agent

Tracks visited tiles per map to enable systematic exploration and avoid repetitive navigation patterns.

Classes

ExplorationTracker

Tracks which tiles have been visited on each map.

Enables the agent to: 1. Know what % of a map has been explored 2. Find unexplored areas to prioritize 3. Suggest next exploration targets

Source code in trackers/exploration_tracker.py

class ExplorationTracker:
    """
    Tracks which tiles have been visited on each map.

    Enables the agent to:
    1. Know what % of a map has been explored
    2. Find unexplored areas to prioritize
    3. Suggest next exploration targets
    """

    def __init__(
        self,
        storage_path: str = "data/exploration_data.json",
        reset_on_start: bool = False,
    ):
        self.storage_path = storage_path
        self.maps: Dict[int, MapExploration] = {}  # map_id -> MapExploration

        if reset_on_start:
            self._save()
            log.info("🗺️ Exploration tracker reset for fresh start")
        else:
            self._load()
            log.info(f"🗺️ Exploration tracker loaded: {len(self.maps)} maps tracked")

    def record_visit(
        self, map_id: int, map_name: str, x: int, y: int, total_walkable: int = 0
    ) -> None:
        """
        Record that a tile has been visited.

        Args:
            map_id: Game map ID
            map_name: Human-readable map name
            x: Tile X coordinate
            y: Tile Y coordinate
            total_walkable: Total walkable tiles on this map (optional, for % calc)
        """
        if map_id not in self.maps:
            self.maps[map_id] = MapExploration(
                map_id=map_id, map_name=map_name, total_walkable=total_walkable
            )

        exploration = self.maps[map_id]
        exploration.visited_tiles.add((x, y))
        exploration.last_visited = datetime.now().isoformat()

        # Update total walkable if provided and larger
        if total_walkable > exploration.total_walkable:
            exploration.total_walkable = total_walkable

        # Auto-save periodically (every 10 visits)
        if len(exploration.visited_tiles) % 10 == 0:
            self._save()

    def get_exploration_summary(self, map_id: int) -> str:
        """
        Get a human-readable exploration summary for a map.

        Returns:
            e.g., "PALLET_TOWN: 45% explored (18/40 tiles)"
        """
        if map_id not in self.maps:
            return "Map not yet explored"

        exp = self.maps[map_id]
        visited = len(exp.visited_tiles)

        if exp.total_walkable > 0:
            return f"{exp.map_name}: {exp.exploration_pct:.0f}% explored ({visited}/{exp.total_walkable} tiles)"
        else:
            return f"{exp.map_name}: {visited} tiles visited"

    def get_unexplored_directions(
        self, map_id: int, player_x: int, player_y: int, minimap_2d: str
    ) -> List[str]:
        """
        Analyze minimap to find directions with unexplored walkable tiles.

        Args:
            map_id: Current map
            player_x: Player X position (in minimap center coords)
            player_y: Player Y position (in minimap center coords)
            minimap_2d: Semicolon-separated minimap string

        Returns:
            List of directions with unexplored tiles, e.g., ["EAST", "NORTH"]
        """
        if not minimap_2d:
            return []

        # Parse minimap
        rows = minimap_2d.split(";")
        if not rows:
            return []

        grid_size = len(rows)
        center = grid_size // 2  # Player is at center

        # Get visited tiles for this map
        visited = self.maps.get(map_id, MapExploration(map_id, "")).visited_tiles

        # Check each quadrant for unexplored walkable tiles
        unexplored_dirs = []

        # Define quadrant bounds (relative to center)
        quadrants = {
            "NORTH": (0, center, 0, grid_size),  # Top half
            "SOUTH": (center, grid_size, 0, grid_size),  # Bottom half
            "WEST": (0, grid_size, 0, center),  # Left half
            "EAST": (0, grid_size, center, grid_size),  # Right half
        }

        for direction, (row_start, row_end, col_start, col_end) in quadrants.items():
            unexplored_count = 0

            for row_idx in range(row_start, min(row_end, len(rows))):
                row = rows[row_idx]
                for col_idx in range(col_start, min(col_end, len(row))):
                    tile = row[col_idx]
                    # Walkable or special tiles that aren't visited
                    if tile in ["W", "O"]:
                        # Convert minimap coords to world coords (rough estimate)
                        world_x = player_x + (col_idx - center)
                        world_y = player_y + (row_idx - center)

                        if (world_x, world_y) not in visited:
                            unexplored_count += 1

            if unexplored_count > 3:  # Threshold to consider direction worth exploring
                unexplored_dirs.append(f"{direction} ({unexplored_count} unexplored)")

        return unexplored_dirs

    def suggest_exploration_target(
        self, map_id: int, player_x: int, player_y: int, minimap_2d: str
    ) -> Optional[Tuple[int, int]]:
        """
        Suggest the nearest unexplored tile to explore.

        Returns:
            (x, y) coordinates of suggested target, or None if fully explored
        """
        if not minimap_2d:
            return None

        rows = minimap_2d.split(";")
        if not rows:
            return None

        grid_size = len(rows)
        center = grid_size // 2

        visited = self.maps.get(map_id, MapExploration(map_id, "")).visited_tiles

        # Find nearest unexplored walkable tile (BFS-like expanding search)
        for distance in range(1, center + 1):
            for dy in range(-distance, distance + 1):
                for dx in range(-distance, distance + 1):
                    if abs(dx) != distance and abs(dy) != distance:
                        continue  # Only check perimeter of current distance

                    row_idx = center + dy
                    col_idx = center + dx

                    if 0 <= row_idx < len(rows) and 0 <= col_idx < len(rows[row_idx]):
                        tile = rows[row_idx][col_idx]
                        if tile in ["W", "O"]:
                            world_x = player_x + dx
                            world_y = player_y + dy

                            if (world_x, world_y) not in visited:
                                return (world_x, world_y)

        return None  # All visible tiles explored

    def get_context_for_llm(
        self, map_id: int, map_name: str, player_x: int, player_y: int, minimap_2d: str
    ) -> str:
        """
        Generate exploration context for LLM injection.

        Returns:
            Multi-line string with exploration status and suggestions
        """
        parts = []

        # Exploration summary
        summary = self.get_exploration_summary(map_id)
        parts.append(f"EXPLORATION: {summary}")

        # Unexplored directions with counts
        unexplored = self.get_unexplored_directions(
            map_id, player_x, player_y, minimap_2d
        )
        if unexplored:
            # Sort by count (highest first) to prioritize direction with most unexplored
            parts.append(f"UNEXPLORED AREAS: {', '.join(unexplored)}")

            # Find direction with most unexplored tiles
            best_dir = unexplored[0].split(" ")[0] if unexplored else None
            if best_dir:
                # Give strong directional command
                dir_to_button = {"NORTH": "U", "SOUTH": "D", "EAST": "R", "WEST": "L"}
                button = dir_to_button.get(best_dir, "U")
                parts.append(
                    f"⚡ PRIORITY: Move {best_dir} (chain 5x {button};{button};{button};{button};{button};) to reach unexplored area"
                )

        # Suggested target
        target = self.suggest_exploration_target(map_id, player_x, player_y, minimap_2d)
        if target:
            dx = target[0] - player_x
            dy = target[1] - player_y
            direction = []
            if dy < 0:
                direction.append("NORTH")
            elif dy > 0:
                direction.append("SOUTH")
            if dx > 0:
                direction.append("EAST")
            elif dx < 0:
                direction.append("WEST")

            dir_str = "-".join(direction) if direction else "NEARBY"
            # Calculate distance for step suggestion
            distance = abs(dx) + abs(dy)
            steps = min(5, max(3, distance))
            parts.append(
                f"SUGGESTED TARGET: Move {dir_str} toward unexplored area ({distance} tiles away, use {steps}+ moves)"
            )

        return "\n".join(parts) if parts else ""

    def _save(self) -> None:
        """Save exploration data to file."""
        try:
            data = {str(map_id): exp.to_dict() for map_id, exp in self.maps.items()}
            with open(self.storage_path, "w") as f:
                json.dump(data, f, indent=2)
        except Exception as e:
            log.error(f"Failed to save exploration data: {e}")

    def _load(self) -> None:
        """Load exploration data from file."""
        if not os.path.exists(self.storage_path):
            return

        try:
            with open(self.storage_path, "r") as f:
                data = json.load(f)

            self.maps = {
                int(map_id): MapExploration.from_dict(exp_data)
                for map_id, exp_data in data.items()
            }
        except Exception as e:
            log.error(f"Failed to load exploration data: {e}")
            self.maps = {}

    def save(self) -> None:
        """Public save method."""
        self._save()

Functions

get_context_for_llm(map_id, map_name, player_x, player_y, minimap_2d)

Generate exploration context for LLM injection.

Returns:

Type	Description
str	Multi-line string with exploration status and suggestions

Source code in trackers/exploration_tracker.py

def get_context_for_llm(
    self, map_id: int, map_name: str, player_x: int, player_y: int, minimap_2d: str
) -> str:
    """
    Generate exploration context for LLM injection.

    Returns:
        Multi-line string with exploration status and suggestions
    """
    parts = []

    # Exploration summary
    summary = self.get_exploration_summary(map_id)
    parts.append(f"EXPLORATION: {summary}")

    # Unexplored directions with counts
    unexplored = self.get_unexplored_directions(
        map_id, player_x, player_y, minimap_2d
    )
    if unexplored:
        # Sort by count (highest first) to prioritize direction with most unexplored
        parts.append(f"UNEXPLORED AREAS: {', '.join(unexplored)}")

        # Find direction with most unexplored tiles
        best_dir = unexplored[0].split(" ")[0] if unexplored else None
        if best_dir:
            # Give strong directional command
            dir_to_button = {"NORTH": "U", "SOUTH": "D", "EAST": "R", "WEST": "L"}
            button = dir_to_button.get(best_dir, "U")
            parts.append(
                f"⚡ PRIORITY: Move {best_dir} (chain 5x {button};{button};{button};{button};{button};) to reach unexplored area"
            )

    # Suggested target
    target = self.suggest_exploration_target(map_id, player_x, player_y, minimap_2d)
    if target:
        dx = target[0] - player_x
        dy = target[1] - player_y
        direction = []
        if dy < 0:
            direction.append("NORTH")
        elif dy > 0:
            direction.append("SOUTH")
        if dx > 0:
            direction.append("EAST")
        elif dx < 0:
            direction.append("WEST")

        dir_str = "-".join(direction) if direction else "NEARBY"
        # Calculate distance for step suggestion
        distance = abs(dx) + abs(dy)
        steps = min(5, max(3, distance))
        parts.append(
            f"SUGGESTED TARGET: Move {dir_str} toward unexplored area ({distance} tiles away, use {steps}+ moves)"
        )

    return "\n".join(parts) if parts else ""

get_exploration_summary(map_id)

Get a human-readable exploration summary for a map.

Returns:

Type	Description
str	e.g., "PALLET_TOWN: 45% explored (18/40 tiles)"

Source code in trackers/exploration_tracker.py

def get_exploration_summary(self, map_id: int) -> str:
    """
    Get a human-readable exploration summary for a map.

    Returns:
        e.g., "PALLET_TOWN: 45% explored (18/40 tiles)"
    """
    if map_id not in self.maps:
        return "Map not yet explored"

    exp = self.maps[map_id]
    visited = len(exp.visited_tiles)

    if exp.total_walkable > 0:
        return f"{exp.map_name}: {exp.exploration_pct:.0f}% explored ({visited}/{exp.total_walkable} tiles)"
    else:
        return f"{exp.map_name}: {visited} tiles visited"

get_unexplored_directions(map_id, player_x, player_y, minimap_2d)

Analyze minimap to find directions with unexplored walkable tiles.

Parameters:

Name	Type	Description	Default
map_id	int	Current map	required
player_x	int	Player X position (in minimap center coords)	required
player_y	int	Player Y position (in minimap center coords)	required
minimap_2d	str	Semicolon-separated minimap string	required

Returns:

Type	Description
List[str]	List of directions with unexplored tiles, e.g., ["EAST", "NORTH"]

Source code in trackers/exploration_tracker.py

def get_unexplored_directions(
    self, map_id: int, player_x: int, player_y: int, minimap_2d: str
) -> List[str]:
    """
    Analyze minimap to find directions with unexplored walkable tiles.

    Args:
        map_id: Current map
        player_x: Player X position (in minimap center coords)
        player_y: Player Y position (in minimap center coords)
        minimap_2d: Semicolon-separated minimap string

    Returns:
        List of directions with unexplored tiles, e.g., ["EAST", "NORTH"]
    """
    if not minimap_2d:
        return []

    # Parse minimap
    rows = minimap_2d.split(";")
    if not rows:
        return []

    grid_size = len(rows)
    center = grid_size // 2  # Player is at center

    # Get visited tiles for this map
    visited = self.maps.get(map_id, MapExploration(map_id, "")).visited_tiles

    # Check each quadrant for unexplored walkable tiles
    unexplored_dirs = []

    # Define quadrant bounds (relative to center)
    quadrants = {
        "NORTH": (0, center, 0, grid_size),  # Top half
        "SOUTH": (center, grid_size, 0, grid_size),  # Bottom half
        "WEST": (0, grid_size, 0, center),  # Left half
        "EAST": (0, grid_size, center, grid_size),  # Right half
    }

    for direction, (row_start, row_end, col_start, col_end) in quadrants.items():
        unexplored_count = 0

        for row_idx in range(row_start, min(row_end, len(rows))):
            row = rows[row_idx]
            for col_idx in range(col_start, min(col_end, len(row))):
                tile = row[col_idx]
                # Walkable or special tiles that aren't visited
                if tile in ["W", "O"]:
                    # Convert minimap coords to world coords (rough estimate)
                    world_x = player_x + (col_idx - center)
                    world_y = player_y + (row_idx - center)

                    if (world_x, world_y) not in visited:
                        unexplored_count += 1

        if unexplored_count > 3:  # Threshold to consider direction worth exploring
            unexplored_dirs.append(f"{direction} ({unexplored_count} unexplored)")

    return unexplored_dirs

record_visit(map_id, map_name, x, y, total_walkable=0)

Record that a tile has been visited.

Parameters:

Name	Type	Description	Default
map_id	int	Game map ID	required
map_name	str	Human-readable map name	required
x	int	Tile X coordinate	required
y	int	Tile Y coordinate	required
total_walkable	int	Total walkable tiles on this map (optional, for % calc)	0

Source code in trackers/exploration_tracker.py

def record_visit(
    self, map_id: int, map_name: str, x: int, y: int, total_walkable: int = 0
) -> None:
    """
    Record that a tile has been visited.

    Args:
        map_id: Game map ID
        map_name: Human-readable map name
        x: Tile X coordinate
        y: Tile Y coordinate
        total_walkable: Total walkable tiles on this map (optional, for % calc)
    """
    if map_id not in self.maps:
        self.maps[map_id] = MapExploration(
            map_id=map_id, map_name=map_name, total_walkable=total_walkable
        )

    exploration = self.maps[map_id]
    exploration.visited_tiles.add((x, y))
    exploration.last_visited = datetime.now().isoformat()

    # Update total walkable if provided and larger
    if total_walkable > exploration.total_walkable:
        exploration.total_walkable = total_walkable

    # Auto-save periodically (every 10 visits)
    if len(exploration.visited_tiles) % 10 == 0:
        self._save()

save()

Public save method.

Source code in trackers/exploration_tracker.py

def save(self) -> None:
    """Public save method."""
    self._save()

suggest_exploration_target(map_id, player_x, player_y, minimap_2d)

Suggest the nearest unexplored tile to explore.

Returns:

Type	Description
Optional[Tuple[int, int]]	(x, y) coordinates of suggested target, or None if fully explored

Source code in trackers/exploration_tracker.py

def suggest_exploration_target(
    self, map_id: int, player_x: int, player_y: int, minimap_2d: str
) -> Optional[Tuple[int, int]]:
    """
    Suggest the nearest unexplored tile to explore.

    Returns:
        (x, y) coordinates of suggested target, or None if fully explored
    """
    if not minimap_2d:
        return None

    rows = minimap_2d.split(";")
    if not rows:
        return None

    grid_size = len(rows)
    center = grid_size // 2

    visited = self.maps.get(map_id, MapExploration(map_id, "")).visited_tiles

    # Find nearest unexplored walkable tile (BFS-like expanding search)
    for distance in range(1, center + 1):
        for dy in range(-distance, distance + 1):
            for dx in range(-distance, distance + 1):
                if abs(dx) != distance and abs(dy) != distance:
                    continue  # Only check perimeter of current distance

                row_idx = center + dy
                col_idx = center + dx

                if 0 <= row_idx < len(rows) and 0 <= col_idx < len(rows[row_idx]):
                    tile = rows[row_idx][col_idx]
                    if tile in ["W", "O"]:
                        world_x = player_x + dx
                        world_y = player_y + dy

                        if (world_x, world_y) not in visited:
                            return (world_x, world_y)

    return None  # All visible tiles explored

MapExploration dataclass

Tracks exploration state for a single map.

Source code in trackers/exploration_tracker.py

@dataclass
class MapExploration:
    """Tracks exploration state for a single map."""

    map_id: int
    map_name: str
    visited_tiles: Set[Tuple[int, int]] = field(default_factory=set)
    total_walkable: int = 0
    last_visited: str = ""

    @property
    def exploration_pct(self) -> float:
        """Percentage of walkable tiles that have been visited."""
        if self.total_walkable == 0:
            return 0.0
        return (len(self.visited_tiles) / self.total_walkable) * 100

    def to_dict(self) -> dict:
        """Serialize for JSON storage."""
        return {
            "map_id": self.map_id,
            "map_name": self.map_name,
            "visited_tiles": list(self.visited_tiles),  # Convert set to list for JSON
            "total_walkable": self.total_walkable,
            "last_visited": self.last_visited,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "MapExploration":
        """Deserialize from JSON storage."""
        return cls(
            map_id=data["map_id"],
            map_name=data["map_name"],
            visited_tiles=set(tuple(t) for t in data.get("visited_tiles", [])),
            total_walkable=data.get("total_walkable", 0),
            last_visited=data.get("last_visited", ""),
        )

Attributes

exploration_pct property

Percentage of walkable tiles that have been visited.

Functions

from_dict(data) classmethod

Deserialize from JSON storage.

Source code in trackers/exploration_tracker.py

@classmethod
def from_dict(cls, data: dict) -> "MapExploration":
    """Deserialize from JSON storage."""
    return cls(
        map_id=data["map_id"],
        map_name=data["map_name"],
        visited_tiles=set(tuple(t) for t in data.get("visited_tiles", [])),
        total_walkable=data.get("total_walkable", 0),
        last_visited=data.get("last_visited", ""),
    )

to_dict()

Serialize for JSON storage.

Source code in trackers/exploration_tracker.py

def to_dict(self) -> dict:
    """Serialize for JSON storage."""
    return {
        "map_id": self.map_id,
        "map_name": self.map_name,
        "visited_tiles": list(self.visited_tiles),  # Convert set to list for JSON
        "total_walkable": self.total_walkable,
        "last_visited": self.last_visited,
    }

History Tracker

trackers.history_tracker

Screenshot History Tracker

Tracks screenshot history between cycles. Maintains history for cleanup and reference.

Classes

ScreenshotHistoryTracker

Tracks screenshot paths across agent cycles.

Source code in trackers/history_tracker.py

class ScreenshotHistoryTracker:
    """
    Tracks screenshot paths across agent cycles.
    """

    def __init__(self, snapshot_dir: str = "snapshots", max_history: int = 2):
        """
        Initialize the tracker.

        Args:
            snapshot_dir: Directory where snapshots are stored
            max_history: Number of cycles to keep (default 2)
        """
        self.snapshot_dir = snapshot_dir
        self.max_history = max_history
        # List of (cycle_num, path) tuples, newest first
        self.history: List[Tuple[int, str]] = []

        # Ensure snapshot directory exists
        os.makedirs(snapshot_dir, exist_ok=True)
        log.info(
            f"📸 Screenshot history tracker initialized (dir: {snapshot_dir}, max: {max_history})"
        )

    def add_screenshot(self, cycle_num: int, path: str) -> None:
        """
        Register a new screenshot for the given cycle.

        Adds to front of history, trims to max_history, and cleans up old files.

        Args:
            cycle_num: Current cycle number
            path: Path to the screenshot file
        """
        # Add new entry at front
        self.history.insert(0, (cycle_num, path))

        log.debug(f"📸 Registered screenshot: cycle {cycle_num} -> {path}")

        # Cleanup entries beyond max_history
        while len(self.history) > self.max_history:
            old_cycle, old_path = self.history.pop()
            if old_path and os.path.exists(old_path):
                try:
                    os.remove(old_path)
                    # Also clean up any combined versions
                    combined_path = old_path.replace(".png", "_with_minimap.png")
                    if os.path.exists(combined_path):
                        os.remove(combined_path)
                    log.debug(
                        f"🧹 Cleaned up old snapshot: {old_path} (cycle {old_cycle})"
                    )
                except OSError as e:
                    log.warning(f"Failed to cleanup old snapshot {old_path}: {e}")

    def get_current_screenshot(self) -> Optional[str]:
        """
        Get the path to the current cycle's screenshot (N).

        Returns:
            Path to current screenshot if available and exists, None otherwise
        """
        if self.history and os.path.exists(self.history[0][1]):
            return self.history[0][1]
        return None

    def get_previous_screenshot(self) -> Optional[str]:
        """
        Get the path to the previous cycle's screenshot (N-1).

        Returns:
            Path to previous screenshot if available and exists, None otherwise
        """
        if len(self.history) >= 2 and os.path.exists(self.history[1][1]):
            return self.history[1][1]
        return None

Functions

__init__(snapshot_dir='snapshots', max_history=2)

Initialize the tracker.

Parameters:

Name	Type	Description	Default
snapshot_dir	str	Directory where snapshots are stored	'snapshots'
max_history	int	Number of cycles to keep (default 2)	2

Source code in trackers/history_tracker.py

def __init__(self, snapshot_dir: str = "snapshots", max_history: int = 2):
    """
    Initialize the tracker.

    Args:
        snapshot_dir: Directory where snapshots are stored
        max_history: Number of cycles to keep (default 2)
    """
    self.snapshot_dir = snapshot_dir
    self.max_history = max_history
    # List of (cycle_num, path) tuples, newest first
    self.history: List[Tuple[int, str]] = []

    # Ensure snapshot directory exists
    os.makedirs(snapshot_dir, exist_ok=True)
    log.info(
        f"📸 Screenshot history tracker initialized (dir: {snapshot_dir}, max: {max_history})"
    )

add_screenshot(cycle_num, path)

Register a new screenshot for the given cycle.

Adds to front of history, trims to max_history, and cleans up old files.

Parameters:

Name	Type	Description	Default
cycle_num	int	Current cycle number	required
path	str	Path to the screenshot file	required

Source code in trackers/history_tracker.py

def add_screenshot(self, cycle_num: int, path: str) -> None:
    """
    Register a new screenshot for the given cycle.

    Adds to front of history, trims to max_history, and cleans up old files.

    Args:
        cycle_num: Current cycle number
        path: Path to the screenshot file
    """
    # Add new entry at front
    self.history.insert(0, (cycle_num, path))

    log.debug(f"📸 Registered screenshot: cycle {cycle_num} -> {path}")

    # Cleanup entries beyond max_history
    while len(self.history) > self.max_history:
        old_cycle, old_path = self.history.pop()
        if old_path and os.path.exists(old_path):
            try:
                os.remove(old_path)
                # Also clean up any combined versions
                combined_path = old_path.replace(".png", "_with_minimap.png")
                if os.path.exists(combined_path):
                    os.remove(combined_path)
                log.debug(
                    f"🧹 Cleaned up old snapshot: {old_path} (cycle {old_cycle})"
                )
            except OSError as e:
                log.warning(f"Failed to cleanup old snapshot {old_path}: {e}")

get_current_screenshot()

Get the path to the current cycle's screenshot (N).

Returns:

Type	Description
Optional[str]	Path to current screenshot if available and exists, None otherwise

Source code in trackers/history_tracker.py

def get_current_screenshot(self) -> Optional[str]:
    """
    Get the path to the current cycle's screenshot (N).

    Returns:
        Path to current screenshot if available and exists, None otherwise
    """
    if self.history and os.path.exists(self.history[0][1]):
        return self.history[0][1]
    return None

get_previous_screenshot()

Get the path to the previous cycle's screenshot (N-1).

Returns:

Type	Description
Optional[str]	Path to previous screenshot if available and exists, None otherwise

Source code in trackers/history_tracker.py

def get_previous_screenshot(self) -> Optional[str]:
    """
    Get the path to the previous cycle's screenshot (N-1).

    Returns:
        Path to previous screenshot if available and exists, None otherwise
    """
    if len(self.history) >= 2 and os.path.exists(self.history[1][1]):
        return self.history[1][1]
    return None