=== Model: muse-glimmer-30b-q4_k_xl-local | Task: code_gen_long | Rep: 1 === === Score: 5/5 | Wall: 32.223s | Cost: $0.000000 === === Pass: True | Finish: stop | In-reasoning-only: False === --- JUDGE --- Class compiles, has __init__/try_acquire/time_until_available with proper type hints, uses threading.Lock for thread safety, and includes a __main__ demo. All criteria met. --- RESPONSE --- ```python import threading import time 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 = float(capacity) self._last = time.monotonic() self._lock = threading.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.""" if tokens <= 0: return True 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.""" if tokens <= 0: return 0.0 with self._lock: self._refill() if self._tokens >= tokens: return 0.0 needed = tokens - self._tokens return needed / self._rate if self._rate > 0 else float('inf') if __name__ == "__main__": limiter = RateLimiter(rate=2.0, capacity=2) # allow print("allow:", limiter.try_acquire()) # True # consume remaining token limiter.try_acquire() # deny print("deny:", limiter.try_acquire()) # False # time until next token print("wait s:", round(limiter.time_until_available(), 3)) # ~0.5 ``` The class is thread-safe via `threading.Lock`, refills tokens based on `time.monotonic()`, and provides the required `try_acquire` and `time_until_available` APIs. --- 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, max capacity. Need to track last refill time, current tokens. Use time.monotonic() for time. __init__: rate, capacity. self.rate = rate, self.capacity = capacity, self.tokens = float(capacity), self.last = time.monotonic(), lock = threading.Lock() _method _refill(): now = monotonic(), elapsed = now - last, add = elapsed * rate, tokens = min(capacity, tokens + add), 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.0 else needed = requested - tokens, time = needed / rate. Return time. Edge cases: rate zero? Might divide by zero. Assume rate >0. Thread-safe. Docstrings. Main block: demo. We need 3 demonstration calls showing allow, deny, and time-until-available helper. Example: if __name__ == "__main__": limiter = RateLimiter(rate=2.0, capacity=5) print(limiter.try_acquire()) # True ... etc. Maybe show deny after exhausting. Let's craft ~60 lines. Make sure type hints. Potential nuance: tokens is int but internal float. Implementation: import threading, time class RateLimiter: def __init__(self, rate: float, capacity: int): ... Provide docstring. Make sure __main__ block with 3 demonstration calls. Possibly show: limiter = RateLimiter(rate=1.0, capacity=2) print(limiter.try_acquire()) # allow print(limiter.try_acquire()) # allow print(limiter.try_acquire()) # deny print(limiter.time_until_available()) # >0 That's 3 calls? Actually 4. Maybe combine. Requirement: 3 demonstration calls showing allow, deny, and time-until-available helper. Could be three prints. Let's do: limiter = RateLimiter(rate=2, capacity=2) print(limiter.try_acquire()) # True allow limiter.try_acquire() # consume second print(limiter.try_acquire()) # False deny print(limiter.time_until_available()) # ~0.5 That's 3 prints. Make sure code is clean. Add type hints for methods. Potential improvements: use self._tokens: float. Make sure lock. Let's produce final answer with code block. Aim for ~60 lines. Let's count roughly. Lines: 1 import threading 2 import time 3 4 class RateLimiter: 5 def __init__(self, rate: float, capacity: int): 6 ... 7 ... 8 9 def _refill(self): 10 ... 11 12 def try_acquire(self, tokens: int = 1) -> bool: 13 ... 14 15 def time_until_available(self, tokens: int = 1) -> float: 16 ... 17 18 if __name__ == "__main__": 19 ... That's fine. Make docstrings. Ensure thread-safe. Potential edge: capacity int but tokens float. Return float for time_until_available. Make sure time_until_available returns 0 if already available. Implementation details for _refill: need to update self.last = now after refill. Implementation: def _refill(self): now = time.monotonic() elapsed = now - self._last if elapsed > 0: self._tokens = min(self._capacity, self._tokens + elapsed * self._rate) self._last = now But if we call refill multiple times quickly, elapsed small. In try_acquire, after refill, check. Potential issue: if rate is zero, division by zero in time_until_available. Could guard. Simplify. Provide final code. Make sure docstring for each method. Let's produce final output. Maybe add __repr__? Not needed. Ensure ~60 lines. Let's produce final answer. Potential style: use self._lock = threading.Lock() Ok. Let's output code. Make sure main block demonstrates. Possibly use time.sleep to show refill? Not needed. We can just show deny then time_until_available. Ok. Provide final. Potential nuance: tokens parameter should be int >0. Assume. Let's produce final answer with code block. Ensure no external dependencies. Ok. Let's output.