File size: 14,214 Bytes
6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 3116fdf 6cb8604 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 | import asyncio
import json
import random
import aiohttp
import websockets
# base = "araeynn-glorified.hf.space" # For production
base = "localhost:7860" # For local testing
API_URL = f"https://{base}"
if base.startswith("localhost"):
API_URL = f"http://{base}"
WS_URL_ROULETTE = f"ws://{base}/ws/roulette"
WS_URL_SPINNER = f"ws://{base}/ws/spinner"
WS_URL_CRASH = f"ws://{base}/ws/crash"
WS_URL_MINES = f"ws://{base}/ws/mines"
WS_URL_TOWER = f"ws://{base}/ws/tower"
else:
WS_URL_ROULETTE = f"wss://{base}/ws/roulette"
WS_URL_SPINNER = f"wss://{base}/ws/spinner"
WS_URL_CRASH = f"wss://{base}/ws/crash"
WS_URL_MINES = f"wss://{base}/ws/mines"
WS_URL_TOWER = f"wss://{base}/ws/tower"
USERNAME = "testuser"
PASSWORD = "testpass"
TOKEN = None
async def register_user():
global TOKEN
async with aiohttp.ClientSession() as session:
data = {"username": USERNAME, "password": PASSWORD}
async with session.post(f"{API_URL}/register", data=data) as resp:
# If already registered, the endpoint will return an error.
if resp.status != 200:
text = await resp.text()
print(f"Register failed (maybe already registered): {text}")
# Try login if already registered
async with session.post(f"{API_URL}/login", data=data) as login_resp:
if login_resp.status == 200:
res = await login_resp.json()
TOKEN = res.get("token")
print(f"Login response: {res}")
else:
print(f"Login failed: {await login_resp.text()}")
else:
res = await resp.json()
TOKEN = res.get("token")
print(f"Register response: {res}")
async def play_roulette():
async with websockets.connect(WS_URL_ROULETTE) as websocket:
# Send authentication
auth_msg = {"token": TOKEN}
await websocket.send(json.dumps(auth_msg))
# Receive welcome message with balance
welcome = await websocket.recv()
data = json.loads(welcome)
print(f"Initial connection: {data}")
# Wait 2 seconds
await asyncio.sleep(1)
# Bet 1 on red
bet_msg = {"bet": {"red": 1}}
print("Placing bet of 1 on red...")
await websocket.send(json.dumps(bet_msg))
response = await websocket.recv()
print(f"Bet response: {json.loads(response)}")
# Wait 2 seconds again
await asyncio.sleep(2)
# Bet 1 on red again
bet_msg = {"bet": {"red": 1}}
print("Placing another bet of 1 on red...")
await websocket.send(json.dumps(bet_msg))
response = await websocket.recv()
print(f"Bet response: {json.loads(response)}")
async def play_spinner():
# First get current balance to bet all-in
async with aiohttp.ClientSession() as session:
params = {k: v for k, v in {"token": TOKEN}.items() if v is not None}
filtered_params = {
k: str(v) for k, v in params.items()
} # Ensure all values are strings
async with session.get(f"{API_URL}/glares", params=filtered_params) as resp:
balance_data = await resp.json()
balance = balance_data.get("glares", 0)
print(f"Current balance: {balance}")
# Now connect to spinner and bet all-in
async with websockets.connect(WS_URL_SPINNER) as websocket:
# Send authentication
auth_msg = {"token": TOKEN}
await websocket.send(json.dumps(auth_msg))
# Receive welcome message
welcome = await websocket.recv()
data = json.loads(welcome)
print(f"Spinner connection: {data}")
# Get current balance for all-in bet
current_balance = data.get("glares", balance)
# Wait 2 seconds
await asyncio.sleep(2)
# All-in bet
bet_msg = {"bet": int(current_balance)}
print(f"Placing ALL-IN bet of {int(current_balance)} on spinner...")
await websocket.send(json.dumps(bet_msg))
# Get result
response = await websocket.recv()
result = json.loads(response)
print(f"Spinner result: {result}")
if "error" in result:
print(f"Error in spinner: {result['error']}")
else:
multiplier = result.get("multiplier", 0)
earnings = result.get("earnings", 0)
print(f"Multiplier: {multiplier}x")
print(f"Earnings: {earnings}")
async def play_crash():
"""Play the crash game and cash out after 2 seconds"""
# Get balance first
async with aiohttp.ClientSession() as session:
params = {k: v for k, v in {"token": TOKEN}.items() if v is not None}
filtered_params = {k: v for k, v in params.items() if v is not None}
async with session.get(f"{API_URL}/glares", params=filtered_params) as resp:
balance_data = await resp.json()
balance = balance_data.get("glares", 0)
print(f"Current balance before crash: {balance}")
# Connect to crash game
async with websockets.connect(WS_URL_CRASH) as websocket:
# Send authentication
auth_msg = {"token": TOKEN}
await websocket.send(json.dumps(auth_msg))
# Receive welcome message
welcome = await websocket.recv()
data = json.loads(welcome)
print(f"Crash game connection: {data}")
# Wait 1 second
await asyncio.sleep(1)
# Place bet (half of balance)
bet_amount = max(int(balance / 2), 1) # At least 1
bet_msg = {"bet": bet_amount}
print(f"Placing bet of {bet_amount} on crash game...")
await websocket.send(json.dumps(bet_msg))
# Set up a task to cash out after 2 seconds
async def cashout_after_delay():
await asyncio.sleep(2)
print("Cashing out after 2 seconds...")
await websocket.send(json.dumps({"action": "cashout"}))
cashout_task = asyncio.create_task(cashout_after_delay())
# Listen for messages
last_multiplier = 1.0
try:
while True:
response = await websocket.recv()
data = json.loads(response)
# Track multipliers
if "multiplier" in data:
last_multiplier = data["multiplier"]
print(f"Current multiplier: {last_multiplier}x", end="\r")
# Check for crash
if "crash" in data:
print(f"\nGame crashed at {data['crash']}x!")
break
# Check for cash out
if "cashed_out" in data:
print(f"\nCashed out at {data['cashed_out']}x!")
# Check for final message
if "message" in data and "earnings" in data:
print(f"Result: {data['message']}")
print(f"Earnings: {data['earnings']}")
break
except Exception as e:
print(f"Error in crash game: {e}")
finally:
if not cashout_task.done():
cashout_task.cancel()
async def play_mines():
"""Play the mines game with random placements and cash out after 5 reveals"""
# Get balance first
async with aiohttp.ClientSession() as session:
params = {"token": TOKEN}
filtered_params = {k: v for k, v in params.items() if v is not None}
async with session.get(f"{API_URL}/glares", params=filtered_params) as resp:
balance_data = await resp.json()
balance = balance_data.get("glares", 0)
print(f"Current balance before mines: {balance}")
# Connect to mines game
async with websockets.connect(WS_URL_MINES) as websocket:
# Send authentication
auth_msg = {"token": TOKEN}
await websocket.send(json.dumps(auth_msg))
# Receive welcome message
welcome = await websocket.recv()
data = json.loads(welcome)
print(f"Mines game connection: {data}")
# Wait 1 second
await asyncio.sleep(1)
# Place bet (half of balance)
bet_amount = max(int(balance / 2), 1) # At least 1
mines_count = 3 # Using 5 mines (medium difficulty)
bet_msg = {"bet": bet_amount, "mines": mines_count}
print(f"Placing bet of {bet_amount} on mines game with {mines_count} mines...")
await websocket.send(json.dumps(bet_msg))
# Keep track of revealed positions
revealed_positions = set()
available_positions = set(range(25))
num_reveals = 0
max_reveals = 5 # Cash out after 5 reveals
try:
while True:
# Choose a random position that hasn't been revealed yet
available = list(available_positions - revealed_positions)
if not available or num_reveals >= max_reveals:
# Cash out after revealing max_reveals positions
print(f"Cashing out after {num_reveals} reveals...")
await websocket.send(json.dumps({"action": "cashout"}))
break
position = random.choice(available)
print(f"Revealing position {position}...")
await websocket.send(
json.dumps({"action": "reveal", "position": position})
)
# Get result
response = await websocket.recv()
data = json.loads(response)
if "error" in data:
print(f"Error in mines game: {data['error']}")
break
if "multiplier" in data and "revealed" in data:
revealed_positions = set(data["revealed"])
multiplier = data["multiplier"]
print(f"Current multiplier: {multiplier}x")
num_reveals += 1
continue
# If we get here with other data, the game might be over
if "multiplier" in data and "earnings" in data:
print(f"Game finished with multiplier: {data['multiplier']}x")
print(f"Earnings: {data['earnings']}")
if "safe_positions" in data:
print(f"Safe positions were: {data['safe_positions']}")
break
except Exception as e:
print(f"Error in mines game: {e}")
async def play_tower():
"""Play the tower game with random tile selections and cash out after a random number of levels"""
# Get balance first
async with aiohttp.ClientSession() as session:
params = {"token": TOKEN}
filtered_params = {k: v for k, v in params.items() if v is not None}
async with session.get(f"{API_URL}/glares", params=filtered_params) as resp:
balance_data = await resp.json()
balance = balance_data.get("glares", 0)
print(f"Current balance before tower: {balance}")
# Connect to tower game
async with websockets.connect(WS_URL_TOWER) as websocket:
# Send authentication
auth_msg = {"token": TOKEN}
await websocket.send(json.dumps(auth_msg))
# Receive welcome message
welcome = await websocket.recv()
data = json.loads(welcome)
print(f"Tower game connection: {data}")
# Wait 1 second
await asyncio.sleep(1)
# Place bet (half of balance)
bet_amount = max(int(balance / 2), 1) # At least 1
bet_msg = {"bet": bet_amount}
print(f"Placing bet of {bet_amount} on tower game...")
await websocket.send(json.dumps(bet_msg))
# Choose a random number of levels to attempt (1-8)
target_levels = random.randint(1, 8)
print(f"Will attempt {target_levels} levels before cashing out")
current_level = 0
try:
while True:
# Check if we've reached our target number of levels
if current_level >= target_levels:
print(f"Reached target of {target_levels} levels. Cashing out...")
await websocket.send(json.dumps({"action": "cashout"}))
break
# Choose a random position (0-3) for the next level
position = random.randint(0, 3)
print(f"Selecting position {position} for level {current_level + 1}...")
await websocket.send(
json.dumps({"action": "reveal", "position": position})
)
# Get result
response = await websocket.recv()
data = json.loads(response)
if "error" in data:
print(f"Error in tower game: {data['error']}")
break
if "level" in data and "multiplier" in data:
current_level = data["level"]
multiplier = data["multiplier"]
print(
f"Advanced to level {current_level} with multiplier: {multiplier}x"
)
continue
# If we get here with other data, the game might be over
if "multiplier" in data and "earnings" in data:
print(f"Game finished with multiplier: {data['multiplier']}x")
print(f"Earnings: {data['earnings']}")
if "safe_positions" in data:
print(f"Safe positions were: {data['safe_positions']}")
break
except Exception as e:
print(f"Error in tower game: {e}")
async def main():
await register_user()
await play_roulette()
# Play spinner all-in
# await play_spinner()
# Play crash game with cashout after 2 seconds
await play_crash()
# Play mines game with random placements
await play_mines()
# Play tower game with random levels
await play_tower()
if __name__ == "__main__":
asyncio.run(main())
|