=== Model: muse-glimmer-30b-q4_k_xl-local | Task: code_gen_long | Rep: 5 === === Score: 5/5 | Wall: 34.201s | Cost: $0.000000 === === Pass: True | Finish: stop | In-reasoning-only: False === --- JUDGE --- Class compiles, contains __init__, try_acquire, and time_until_available with proper type hints (plus a private _refill helper), uses threading.Lock for thread safety, and includes a __main__ demo blo --- RESPONSE --- ```python import time from threading import Lock class RateLimiter: def __init__(self, rate: float, capacity: int): """Create a token bucket limiter. rate: tokens added per second capacity: maximum bucket size """ self.rate = float(rate) self.capacity = float(capacity) self._tokens = self.capacity self._last = time.monotonic() self._lock = Lock() def _refill(self) -> None: now = time.monotonic() elapsed = now - self._last if elapsed > 0: self._tokens = min(self.capacity, self._tokens + elapsed * self.rate) self._last = now def try_acquire(self, tokens: int = 1) -> bool: """Return True and consume tokens if available, else False.""" with self._lock: self._refill() if self._tokens >= tokens: self._tokens -= tokens return True return False def time_until_available(self, tokens: int = 1) -> float: """Seconds until `tokens` can be acquired. 0 if already available.""" with self._lock: self._refill() if self._tokens >= tokens: return 0.0 needed = float(tokens) - self._tokens return needed / self.rate if __name__ == "__main__": limiter = RateLimiter(rate=2.0, capacity=3) # allow print("allow:", limiter.try_acquire()) # True print("allow:", limiter.try_acquire()) # True # deny - bucket is empty print("deny:", limiter.try_acquire(3)) # False # helper print("wait:", round(limiter.time_until_available(3), 2), "s") ``` The class is thread-safe via `Lock`, refills tokens based on `time.monotonic()`, and provides `try_acquire` and `time_until_available` with full type hints and docstrings. --- REASONING --- Write a complete Python class `RateLimiter` (token-bucket algorithm) with: - `__init__(self, rate: float, capacity: int)` where rate is tokens per second and capacity is max bucket size - `try_acquire(self, tokens: int = 1) -> bool` that returns True if tokens are available (and decrements), False otherwise - `time_until_available(self, tokens: int = 1) -> float` that returns seconds until `tokens` are available (0 if already available) - Thread-safe (use threading.Lock) - Full type hints and a docstring for each method - A `__main__` block with 3 demonstration calls showing allow, deny, and the time-until-available helper Aim for ~60 lines. No external dependencies beyond stdlib. We need to produce code. Probably about 60 lines. Provide class. Implementation details: token bucket: tokens refill over time at rate per second, capacity max. Need to track last refill time, current tokens. Use time.monotonic() for time. __init__: self.rate, self.capacity, self.tokens = capacity? Usually start full. Probably start full. Could start full. Let's start full. self.last = time.monotonic() _refill: compute elapsed = now - last, add tokens = min(capacity, current + elapsed*rate). Update last. try_acquire: with lock, refill, if tokens >= requested, subtract, return True else False. time_until_available: with lock, refill, if tokens >= requested return 0 else need = requested - tokens, time = need / rate. Return time. Thread-safe. Docstrings. Main block: demo. Need 3 demonstration calls showing allow, deny, and time-until-available helper. Example: limiter = RateLimiter(rate=2, capacity=5) print(limiter.try_acquire()) -> True ... acquire many to deny print(limiter.time_until_available()) Make sure ~60 lines. Provide code. Potential edge cases: rate >0. capacity int. Type hints. Use from threading import Lock. Make sure imports. Let's produce final answer with code block. Probably want to ensure tokens is float. Implementation: class RateLimiter: def __init__(self, rate: float, capacity: int): ... Make docstring. Let's produce ~60 lines. Count lines roughly. Let's draft: import time from threading import Lock class RateLimiter: def __init__(self, rate: float, capacity: int): ... That's about 60 lines. Provide __main__. Make sure to use if __name__ == "__main__": ... Demonstration calls: limiter = RateLimiter(rate=1.0, capacity=3) print(limiter.try_acquire()) # True print(limiter.try_acquire()) # True print(limiter.try_acquire()) # True print(limiter.try_acquire()) # False deny print(limiter.time_until_available()) # >0 Maybe show allow, deny, time. Ok. Potential nuance: time_until_available should not modify last? It should refill first. That's fine. Make sure thread-safe. Return float. Add type hints for methods. Let's produce final output. Maybe add __repr__? Not needed. Ensure code is clean. Potential improvement: use self._tokens: float. Ok. Let's produce final answer. Make sure docstring for each method. Provide final code. Probably want to ensure capacity is int but tokens float. Edge: tokens requested > capacity? Could return False? Might still compute time. We'll assume tokens <= capacity. Ok. Let's output. Probably use Python 3. Make sure no external dependencies. Provide final. Let's produce final answer with code block. Maybe add comments. Aim for ~60 lines. Let's count approximate. 1 import time 2 from threading import Lock 3 4 class RateLimiter: 5 def __init__(self, rate: float, capacity: int): 6 ... 7 self.rate = rate 8 self.capacity = float(capacity) 9 self._tokens = self.capacity 10 self._last = time.monotonic() 11 self._lock = Lock() 12 13 def _refill(self): 14 now = time.monotonic() 15 elapsed = now - self._last 16 if elapsed > 0: 17 self._tokens = min(self.capacity, self._tokens + elapsed * self.rate) 18 self._last = now 19 20 def try_acquire(self, tokens: int = 1) -> bool: 21 ... 22 with self._lock: 23 self._refill() 24 if self._tokens >= tokens: 25 self._tokens -= tokens 26 return True 27 return False 28 29 def time_until_available(self, tokens: int = 1) -> float: 30 ... 31 with self._lock: 32 self._refill() 33 if self._tokens >= tokens: 34 return 0.0 35 needed = tokens - self._tokens 36 return needed / self.rate 37 38 if __name__ == "__main__": 39 limiter = RateLimiter(rate=2.0, capacity=3) 40 print("allow:", limiter.try_acquire()) # True 41 print("allow:", limiter.try_acquire()) # True 42 print("allow:", limiter.try_acquire()) # True 43 print("deny:", limiter.try_acquire()) # False 44 print("wait:", limiter.time_until_available(), "s") 45 That's ~45 lines. Good. Maybe add more demo. Ok. Provide final answer. Potential nuance: capacity int but we store float. Make docstrings multi-line. Let's produce final output with proper docstrings. Ensure type hints. Ok. Let's output.