Skip to content

fix(buff): implement QueueLogic for single-stack duration boons #244

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Roddygithub merged 3 commits into from
Aug 18, 2026
Merged

fix(buff): implement QueueLogic for single-stack duration boons #244

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
948161c
fix(buff): implement QueueLogic for single-stack duration boons
Roddygithub Aug 18, 2026
b5cf08f
style(buff): ruff format
Roddygithub Aug 18, 2026
7e9af87
fix(buff): repair QueueLogic duration/expiry semantics + restore inte…
Roddygithub Aug 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 167 additions & 46 deletions libs/gw2_analytics/src/gw2_analytics/buff_state.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@

from __future__ import annotations

import math
from collections import defaultdict

from pydantic import BaseModel, ConfigDict
Expand Down Expand Up @@ -67,18 +66,41 @@
#: drop shortest on overflow, graft extensions by closest TotalDuration).
_OVERRIDE_LOGIC_BUFFS: frozenset[str] = frozenset({"might", "stability"})

#: Single-stack duration boons that use Elite Insights' QueueLogic (capacity 9,
#: only front stack counts toward uptime, drop shortest on overflow,
#: graft extensions by closest duration, added_active moves to front).
#: Regeneration uses its own special queue logic; might/stability use OverrideLogic.
_QUEUE_LOGIC_BUFFS: frozenset[str] = frozenset(
{
"fury",
"quickness",
"alacrity",
"protection",
"vigor",
"aegis",
"swiftness",
"resistance",
"resolution",
"superspeed",
"stealth",
}
)

_CAPACITIES = {
"might": 25,
"stability": 25,
"regeneration": 5,
"stealth": 5,
"fury": 99,
"quickness": 99,
"protection": 99,
"vigor": 99,
"swiftness": 99,
"resistance": 99,
"resolution": 99,
"stealth": 9,
"fury": 9,
"quickness": 9,
"alacrity": 9,
"protection": 9,
"vigor": 9,
"aegis": 9,
"swiftness": 9,
"resistance": 9,
"resolution": 9,
"superspeed": 9,
}
# All other boons default to 1 stack max (handled in compute logic).

Expand Down Expand Up @@ -215,29 +237,37 @@ def _get_stack(self, agent_id: int, buff_name: str) -> _BuffStack:
return agent[buff_name]

@staticmethod
def _advance(stack: _BuffStack, new_time_ms: int) -> None:
"""Accumulate stack-time through expirations up to ``new_time_ms``."""
if _max_stacks_for(stack.name) == 1:
elapsed = new_time_ms - stack.last_time_ms
while elapsed > 0 and stack.expirations:
remaining = stack.expirations[0]
if remaining is None:
stack.cumulative_stack_ms += elapsed
break
active = min(elapsed, remaining)
stack.cumulative_stack_ms += active
elapsed -= active
remaining -= active
if remaining == 0:
stack.expirations.pop(0)
if stack.total_durations:
stack.total_durations.pop(0)
stack.stack_ids.pop(0)
stack.healing_scores.pop(0)
else:
stack.expirations[0] = remaining
stack.last_time_ms = new_time_ms
return
def _advance_single(stack: _BuffStack, new_time_ms: int) -> None:
"""Accumulate stack-time for single-stack (max_stacks=1) through expirations."""
elapsed = new_time_ms - stack.last_time_ms
while elapsed > 0 and stack.expirations:
remaining = stack.expirations[0]
if remaining is None:
stack.cumulative_stack_ms += elapsed
break
active = min(elapsed, remaining)
stack.cumulative_stack_ms += active
elapsed -= active
remaining -= active
if remaining == 0:
stack.expirations.pop(0)
if stack.total_durations:
stack.total_durations.pop(0)
stack.stack_ids.pop(0)
stack.healing_scores.pop(0)
else:
stack.expirations[0] = remaining
stack.last_time_ms = new_time_ms

@staticmethod
def _advance_queue(stack: _BuffStack, new_time_ms: int) -> None:
"""Accumulate stack-time for QueueLogic (only front stack counts)."""
# QueueLogic uses the same front-stack consumption logic as single-stack
BuffStateTracker._advance_single(stack, new_time_ms)

@staticmethod
def _advance_intensity(stack: _BuffStack, new_time_ms: int) -> None:
"""Accumulate stack-time for intensity buffs (max_stacks > 1)."""
while True:
next_expiry = min(
(expiry for expiry in stack.expirations if expiry is not None),
Expand All @@ -256,6 +286,17 @@ def _advance(stack: _BuffStack, new_time_ms: int) -> None:
stack.cumulative_stack_ms += len(stack.expirations) * (new_time_ms - stack.last_time_ms)
stack.last_time_ms = new_time_ms

@staticmethod
def _advance(stack: _BuffStack, new_time_ms: int) -> None:
"""Accumulate stack-time through expirations up to ``new_time_ms``."""
if _max_stacks_for(stack.name) == 1:
if stack.name in _QUEUE_LOGIC_BUFFS:
BuffStateTracker._advance_queue(stack, new_time_ms)
else:
BuffStateTracker._advance_single(stack, new_time_ms)
return
BuffStateTracker._advance_intensity(stack, new_time_ms)

def _regen_overstack_hint(self, agent_id: int, time_ms: int) -> tuple[int, int] | None:
"""The displaced-stack hint arcdps recorded just before this apply."""
hints = self._regen_overstacks.get(agent_id)
Expand Down Expand Up @@ -358,7 +399,38 @@ def process( # noqa: PLR0912, PLR0915
self._advance(target_tracker, time_ms)

if event.kind == "apply":
if _max_stacks_for(buff_name) > 1:
# QueueLogic buffs (single-stack duration boons with queue behavior)
# are handled separately even though they have max_stacks=1.
if buff_name in _QUEUE_LOGIC_BUFFS:
# QueueLogic (EI): capacity 9, only front stack counts.
# Add new stack to end; if at capacity, drop shortest (not front).
duration = event.duration_ms if event.duration_ms > 0 else None
if len(target_tracker.expirations) >= _capacity_for(buff_name):
# Find shortest duration (excluding front which is active)
# EI drops the shortest duration stack, not the front
if len(target_tracker.expirations) > 1:
# Find shortest among non-front stacks
min_idx = 1
min_dur = target_tracker.expirations[1]
for i in range(2, len(target_tracker.expirations)):
dur = target_tracker.expirations[i]
if dur is not None and (min_dur is None or dur < min_dur):
min_dur = dur
min_idx = i
target_tracker.expirations.pop(min_idx)
target_tracker.stack_ids.pop(min_idx)
target_tracker.healing_scores.pop(min_idx)
else:
# Only one stack, replace it
target_tracker.expirations.pop(0)
target_tracker.stack_ids.pop(0)
target_tracker.healing_scores.pop(0)
target_tracker.expirations.append(duration)
target_tracker.stack_ids.append(event.stack_id)
target_tracker.healing_scores.append(
self._healing_by_agent.get(event.source_agent_id, 0)
)
elif _max_stacks_for(buff_name) > 1:
if buff_name in _OVERRIDE_LOGIC_BUFFS:
# OverrideLogic (EI): sort by TotalDuration (shortest first),
# remove index 0 when at capacity.
Expand Down Expand Up @@ -467,7 +539,30 @@ def process( # noqa: PLR0912, PLR0915
del target_tracker.stack_ids[_capacity_for(buff_name) :]
del target_tracker.healing_scores[_capacity_for(buff_name) :]
elif event.kind == "remove_single":
if _max_stacks_for(buff_name) == 1 and target_tracker.expirations:
if buff_name in _QUEUE_LOGIC_BUFFS and target_tracker.expirations:
# QueueLogic: find stack by stack_id or closest duration
stack_index = next(
(
i
for i, stack_id in enumerate(target_tracker.stack_ids)
if event.stack_id is not None and stack_id == event.stack_id
),
None,
)
if stack_index is None:
stack_index = next(
(
i
for i, duration in enumerate(target_tracker.expirations)
if duration is not None and abs(duration - event.duration_ms) < 15
),
None,
)
if stack_index is not None:
target_tracker.expirations.pop(stack_index)
target_tracker.stack_ids.pop(stack_index)
target_tracker.healing_scores.pop(stack_index)
elif _max_stacks_for(buff_name) == 1 and target_tracker.expirations:
stack_index = next(
(
i
Expand Down Expand Up @@ -517,7 +612,7 @@ def process( # noqa: PLR0912, PLR0915
event.duration_ms * event.stacks
)

def _process_buff_apply(self, event: BuffApplyEvent) -> None:
def _process_buff_apply(self, event: BuffApplyEvent) -> None: # noqa: PLR0915
"""Process a ``BuffApplyEvent`` (CBTS_BUFFAPPLY statechange).

These are initial-stack snapshots: ``skill_id`` is the buff ID,
Expand All @@ -536,7 +631,23 @@ def _process_buff_apply(self, event: BuffApplyEvent) -> None:
time_ms = self._relative_time(event.time_ms)
self._advance(target_tracker, time_ms)
expiry = time_ms + event.duration_ms if event.duration_ms > 0 else None
if _max_stacks_for(buff_name) > 1:
# QueueLogic buffs (single-stack duration boons with queue behavior)
if buff_name in _QUEUE_LOGIC_BUFFS:
# QueueLogic: add initial stacks as queue
duration = event.duration_ms if event.duration_ms > 0 else None
target_tracker.expirations.extend([duration] * event.stacks)
target_tracker.stack_ids.extend([event.stack_id] * event.stacks)
target_tracker.healing_scores.extend([0] * event.stacks)
# No total_durations for QueueLogic buffs
# Keep only up to capacity (drop oldest if over capacity)
if len(target_tracker.expirations) > _capacity_for(buff_name):
excess = len(target_tracker.expirations) - _capacity_for(buff_name)
target_tracker.expirations = target_tracker.expirations[excess:]
target_tracker.stack_ids = target_tracker.stack_ids[excess:]
target_tracker.healing_scores = target_tracker.healing_scores[excess:]

# Intensity buffs with max_stacks > 1 (might, stability)
elif _max_stacks_for(buff_name) > 1:
if buff_name in _OVERRIDE_LOGIC_BUFFS:
# OverrideLogic: track TotalDuration for each stack
total_dur = event.duration_ms
Expand All @@ -561,23 +672,26 @@ def _process_buff_apply(self, event: BuffApplyEvent) -> None:
target_tracker.stack_ids = [sid for _, _, sid, _ in pairs]
target_tracker.healing_scores = [h for _, _, _, h in pairs]
else:
# Other intensity buffs (stability): sort by expiration
target_tracker.expirations.extend([expiry] * event.stacks)
target_tracker.stack_ids.extend([event.stack_id] * event.stacks)
target_tracker.healing_scores.extend([0] * event.stacks)
pairs_3: list[tuple[int | None, int, int]] = sorted(
target_tracker.healing_scores.extend(
[self._healing_by_agent.get(event.source_agent_id, 0)] * event.stacks
)
stability_pairs = sorted(
zip(
target_tracker.expirations,
target_tracker.stack_ids,
target_tracker.healing_scores,
strict=True,
),
key=lambda pair: pair[0] if pair[0] is not None else math.inf,
)[-_capacity_for(buff_name) :]
target_tracker.expirations = [value for value, _, _ in pairs_3]
target_tracker.stack_ids = [stack_id for _, stack_id, _ in pairs_3]
target_tracker.healing_scores = [healing for _, _, healing in pairs_3]
if target_tracker.total_durations:
target_tracker.total_durations.clear()
)
)
stability_pairs = stability_pairs[-_capacity_for(buff_name) :]
target_tracker.expirations = [expiry for expiry, _, _ in stability_pairs]
target_tracker.stack_ids = [stack_id for _, stack_id, _ in stability_pairs]
target_tracker.healing_scores = [healing for _, _, healing in stability_pairs]

# Single-stack duration boons (fury, quickness, etc.) - original behavior
else:
target_tracker.expirations.append(event.duration_ms or None)
target_tracker.stack_ids.append(event.stack_id)
Expand Down Expand Up @@ -611,6 +725,13 @@ def _process_buff_extension(self, event: BuffExtensionEvent) -> None:
target_tracker.expirations[index] or 0
) + event.extended_duration_ms
return
if buff_name in _QUEUE_LOGIC_BUFFS and target_tracker.expirations:
# QueueLogic: graft extension onto stack with duration closest to old_duration
candidates = [(i, e) for i, e in enumerate(target_tracker.expirations) if e is not None]
if candidates:
index, remaining = min(candidates, key=lambda pair: abs(pair[1] - old_duration))
target_tracker.expirations[index] = remaining + event.extended_duration_ms
return
if target_tracker.expirations and (
old_duration > 0 or len(target_tracker.expirations) >= _capacity_for(buff_name)
):
Expand Down
Loading