Poker bot
A Texas Hold'em dealer with play chips, one bot per room. Each player's two cards go to that player alone, and the deck never leaves the bot, so there is nothing in the room to read. Play a hand below, or take the code.
Works today With @pikoman/ws 1.0.0-rc.11 from npm and Pikoman as it runs now.
A folder that runs as it is, in Node and Python: poker.mjs, hand.mjs, poker.py, hand.py, .env.example, README.md, package.json, page.js, pikoman_bot.py, requirements.txt.
Play it
A real table, with a dealer running on our server. Start one, send the link to your friends, and everyone picks a place. Play chips only, and the table is gone when the room is empty.
Dealer · bot
Pick a free place to sit down.
What the code does
Members write !sit to take a seat and !deal to start. The bot shuffles, deals, takes the blinds, asks each player in turn, puts out the flop, the turn and the river, works out the side pots and pays the winners. Then it deals the next hand, until one player has all the chips.
- player's move
- poker bot
- the table
- each player's cards, alone
Play chips only: no real money and nothing that can be cashed in. Playing for money is gambling, and needs a license.
The key it needs
Rights subscribe and client_events on the room prefix: it may join the poker rooms and speak in them, and nothing else.
- Start from
- Chosen: Bot
- Name
- Poker bot
- Rights
- Not chosen:
publishNot chosen:statsNot chosen:clientsNot chosen:disconnectNot chosen:disconnect_allChosen:subscribeChosen:client_eventsNot chosen:kickNot chosen:ban - Channel kinds
- Not chosen: public Not chosen: private Chosen: presence
- Channels
- Chosen: Only some channelsprefix
presence-poker- - Addresses
- Chosen: Only these addressesthe server the bot runs on
- Expires
- Never, or in 90 days for a trial
Set up
The docs cover the key and connecting; everything about the bot itself is on this page.
- The Pro plan. A key that joins channels is a bot’s key, and bot keys come with Pro.
- The key above. In the dashboard: your app, New API key, filled in as shown. How each choice works: Keys and rights. The secret,
sk_k_….…, is shown once (what it looks like). - Your app id and the secret in
.env, copied from.env.example. Nowhere else: not in a page, not in git. - Connecting. The bot signs its own join tokens with the key, and Pikoman shows it as a bot: bots in presence channels.
- Your players join with tokens your backend signs, with their name and user id in the member data,
{"user_id": "42", "name": "Ada"}: the token endpoint. The user id is what gives a player their seat.
How it works
Cards go to one player
room.sendTo({ userId }, "cards", …)reaches that user's tabs and no other connection, so another player has nothing to read, not even in the browser's network tab. A page opened in the middle of a hand gets the cards again.The deck stays with the bot
client-tablesays what everyone may know: the board, the bets, the pot and whose turn it is. It never carries a card from a hand. The shuffle uses the system's secure random numbers, which cannot be predicted from earlier cards.The server says who moved
A move goes to the bot alone with
sendTo, and the server adds the sender's user id. Nothing a player writes can move for someone else. Each move names the turn it answers, so a doubled click does nothing.The bot's clock decides
A player who does not answer in 30 seconds checks if they can, and otherwise folds. Someone who has left loses the seat at the next hand.
Hands stay hidden unless they must show
At a showdown
client-showdownshows the hands still in. Folded hands are never sent, and neither is a hand that won because everyone else folded.It keeps to the rate limit
A connection may send ten client events a second, and a deal is one event per player. Everything the bot sends waits in a queue, nine a second, with the game before the chat lines, so the next turn shows at once.
What you still have to trust: the bot. Whoever runs it can read every card. That is fine for play chips among friends, and it is one reason this bot is not for money.
The source
These are the files in the zip, as they are: copy one, download one, or take the whole folder.
Node: poker.mjs
// poker.mjs: a Texas Hold'em dealer, one bot per room, with play chips. Each player's two cards go
// to that player alone; the deck stays with the bot, so there is nothing in the room to read.
// Members write "!sit" to take a seat and "!deal" to start. Node 22+. npm install, then npm start
// (reads .env).
//
// Its key (dashboard, New API key): rights subscribe and client_events, kind presence,
// channels: prefix presence-poker-.
//
// Players are told apart by the user_id your backend signed into their join token, which the server
// passes on with each event: {"user_id": "42", "name": "Ada"}. Nothing a player sends says who they are.
//
// Play chips only: no real money and nothing that can be cashed in. Playing for money is gambling,
// and needs a license.
import { Pikoman, TokenSigner } from "@pikoman/ws";
import { bestHand, compare, shuffledDeck } from "./hand.mjs";
const ROOM = process.env.ROOM || "presence-poker-1";
const START_CHIPS = Number(process.env.START_CHIPS || 1000);
const BIG_BLIND = Number(process.env.BIG_BLIND || 20);
const SMALL_BLIND = Math.floor(BIG_BLIND / 2);
const TURN_SECONDS = Number(process.env.TURN_SECONDS || 30);
const PAUSE_SECONDS = Number(process.env.PAUSE_SECONDS || 5); // between two hands
const MAX_SEATS = 8;
const STREETS = { preflop: 3, flop: 1, turn: 1 }; // how many cards each next street puts on the board
const pikoman = new Pikoman(process.env.PIKOMAN_APP, {
ws_address: process.env.PIKOMAN_WS || "wss://next.pikoman.se",
});
const auth = TokenSigner.fromSecret(process.env.PIKOMAN_API_KEY, { name: "Dealer" }, 3);
const room = pikoman.listenPresence(ROOM, auth);
// A connection may send ten client events a second, so everything waits its turn, nine a second.
// The game goes first and the chat lines after it, so a player sees the next turn at once.
const game = [];
const chat = [];
let waiting = false;
function pump() {
if (waiting) return;
const next = game.shift() ?? chat.shift();
if (!next) return;
next();
waiting = true;
setTimeout(() => {
waiting = false;
pump();
}, 110);
}
const inTurn = (lane, fn) => {
lane.push(fn);
queueMicrotask(pump); // after the handler that called it: its table goes before its chat line
};
const tell = (event, data) => inTurn(game, () => room.send(event, data));
const tellUser = (user, event, data) => inTurn(game, () => room.sendTo({ userId: user }, event, data));
const say = (text) => inTurn(chat, () => room.send("message", { text }));
// A seat: { user, name, chips, bet, total, folded, cards }. bet is this street, total this hand.
let seats = [];
let dealer = -1;
let playing = false; // a hand is on
let stage = null;
let deck = [];
let board = [];
let handNumber = 0;
let toCall = 0; // what everyone must have bet on this street
let minRaise = 0;
let toAct = new Set(); // the seats that still have to answer
let turn = -1;
let moveNumber = 0; // a move names the one it answers, so a late or doubled click does nothing
let deadline = 0;
let timer = null;
const live = () => seats.filter((s) => !s.folded);
const connected = (user) => [...room.getMembers().values()].some((m) => String(m?.user_id) === user);
/** What everyone may know, and nothing else: never the deck, never a player's cards. */
function sendTable() {
tell("table", table());
}
function table() {
return {
hand: handNumber,
stage,
board,
pot: seats.reduce((sum, s) => sum + s.total, 0),
dealer: seats[dealer]?.user ?? null,
turn: playing && toAct.size ? seats[turn]?.user ?? null : null,
move: moveNumber,
toCall,
minRaise: toCall + minRaise,
seconds: Math.max(0, Math.ceil((deadline - Date.now()) / 1000)),
seats: seats.map(({ user, name, chips, bet, folded }) => ({ user, name, chips, bet, folded })),
};
}
function pay(seat, amount) {
const paid = Math.min(amount, seat.chips);
seat.chips -= paid;
seat.bet += paid;
seat.total += paid;
}
function startHand() {
// Whoever is out of chips or has left the room loses the seat. The button moves to the next one that stays.
const stays = (s) => s.chips > 0 && connected(s.user);
let next = dealer;
for (let i = 0; i < seats.length; i++) {
next = (next + 1) % seats.length;
if (stays(seats[next])) break;
}
const button = seats[next];
seats = seats.filter(stays);
if (seats.length < 2) {
playing = false;
stage = null;
board = [];
for (const s of seats) Object.assign(s, { bet: 0, total: 0, cards: null, folded: true });
say("Waiting for two players: write !sit, then !deal.");
return sendTable();
}
dealer = seats.indexOf(button);
playing = true;
handNumber++;
stage = "preflop";
deck = shuffledDeck();
board = [];
for (const s of seats) Object.assign(s, { bet: 0, total: 0, folded: false, cards: [deck.pop(), deck.pop()] });
// Two players: the button is the small blind and acts first before the flop.
const at = (offset) => (dealer + offset) % seats.length;
const small = seats.length === 2 ? dealer : at(1);
const big = at(seats.length === 2 ? 1 : 2);
pay(seats[small], SMALL_BLIND);
pay(seats[big], BIG_BLIND);
toCall = BIG_BLIND;
minRaise = BIG_BLIND;
toAct = new Set(seats.filter((s) => s.chips > 0));
turn = big;
for (const s of seats) tellUser(s.user, "cards", { hand: handNumber, cards: s.cards });
say(`Hand ${handNumber}: ${seats[dealer].name} has the button.`);
advance();
}
/** The next thing that happens: the next player's turn, the next street, or the end of the hand. */
function advance() {
clearTimeout(timer);
if (live().length === 1) return finish();
for (const s of toAct) if (s.chips === 0 || s.folded) toAct.delete(s);
if (toAct.size === 0) return nextStreet();
do turn = (turn + 1) % seats.length;
while (!toAct.has(seats[turn]));
moveNumber++;
deadline = Date.now() + TURN_SECONDS * 1000;
// The bot's clock decides. A player who is gone or asleep checks if they can, and otherwise folds.
const seat = seats[turn];
timer = setTimeout(() => {
say(`${seat.name} ran out of time.`);
apply(seat, seat.bet === toCall ? "check" : "fold");
}, TURN_SECONDS * 1000);
sendTable();
}
function nextStreet() {
for (const s of seats) s.bet = 0;
toCall = 0;
minRaise = BIG_BLIND;
const cards = STREETS[stage];
if (!cards) return finish();
board.push(...deck.splice(-cards));
stage = { preflop: "flop", flop: "turn", turn: "river" }[stage];
turn = dealer;
toAct = new Set(live().filter((s) => s.chips > 0));
// At most one player can still bet: nobody to bet against, so the rest of the board just comes.
if (toAct.size < 2) {
toAct.clear();
sendTable();
timer = setTimeout(nextStreet, 1500);
return;
}
advance();
}
function apply(seat, action, amount) {
clearTimeout(timer);
if (action === "fold") {
seat.folded = true;
say(`${seat.name} folds.`);
} else if (action === "check") {
say(`${seat.name} checks.`);
} else if (action === "call") {
pay(seat, toCall - seat.bet);
say(`${seat.name} calls.`);
} else {
// A raise is the total to bet this street. A smaller one than minRaise is fine only all in.
if (amount - toCall >= minRaise) minRaise = amount - toCall;
pay(seat, amount - seat.bet);
toCall = amount;
toAct = new Set(live().filter((s) => s.chips > 0));
say(`${seat.name} raises to ${amount}.`);
}
toAct.delete(seat);
advance();
}
/** The pots, then the cards of those still in. Folded hands stay hidden, and so does a hand nobody called. */
function finish() {
clearTimeout(timer);
const players = live();
const showdown = players.length > 1;
const best = new Map(showdown ? players.map((s) => [s, bestHand([...s.cards, ...board])]) : []);
// One pot per all-in size: a player wins only as much from each other player as they put in.
const won = new Map();
const levels = [...new Set(players.map((s) => s.total))].sort((a, b) => a - b);
let below = 0;
levels.forEach((level, i) => {
const top = i === levels.length - 1 ? Infinity : level; // the last pot takes what folded players left above it
const pot = seats.reduce((sum, s) => sum + Math.min(s.total, top) - Math.min(s.total, below), 0);
below = level;
const entitled = players.filter((s) => s.total >= level);
let winners = [entitled[0]];
for (const s of entitled.slice(1)) {
const order = compare(best.get(s).score, best.get(winners[0]).score);
if (order > 0) winners = [s];
else if (order === 0) winners.push(s);
}
// A split pot's odd chips go to the first winner after the button, one each.
const afterButton = (s) => (seats.indexOf(s) - dealer - 1 + seats.length) % seats.length;
winners.sort((a, b) => afterButton(a) - afterButton(b));
winners.forEach((s, n) => {
const share = Math.floor(pot / winners.length) + (n < pot % winners.length ? 1 : 0);
won.set(s, (won.get(s) ?? 0) + share);
});
});
for (const [s, chips] of won) s.chips += chips;
tell("showdown", {
hand: handNumber,
board,
players: showdown ? players.map((s) => ({ user: s.user, name: s.name, cards: s.cards, hand: best.get(s).name })) : [],
winners: [...won].map(([s, chips]) => ({ user: s.user, name: s.name, won: chips })),
});
for (const [s, chips] of won) {
say(showdown ? `${s.name} wins ${chips} with ${best.get(s).name}.` : `${s.name} wins ${chips}.`);
}
for (const s of seats) Object.assign(s, { bet: 0, total: 0, cards: null, folded: true });
toAct.clear();
toCall = 0;
stage = "between";
sendTable();
timer = setTimeout(startHand, PAUSE_SECONDS * 1000);
}
// "!sit" and "!deal" in the room. A seat taken during a hand plays from the next one.
room.onClient("message", (from, data, info) => {
const text = String(data?.text ?? "").trim();
if ((text !== "!sit" && text !== "!deal") || room.isBot(from)) return;
const user = info?.userId;
if (user === undefined) return tell("message", { text: "Your join token needs a user_id to play." });
const seated = seats.find((s) => s.user === user);
if (text === "!sit") {
if (seated) return;
if (seats.length >= MAX_SEATS) return say("The table is full.");
const name = String(room.getMember(from)?.name ?? "Someone").slice(0, 40);
seats.push({ user, name, chips: START_CHIPS, bet: 0, total: 0, folded: true, cards: null });
say(`${name} sits down with ${START_CHIPS} chips.`);
sendTable();
} else if (seated && !playing) {
startHand();
}
});
// A move, sent to the bot alone: room.sendTo(bot, "move", { move: 7, action: "raise", amount: 60 }).
room.onClient("move", (from, data, info) => {
const seat = seats[turn];
if (!playing || !toAct.has(seat) || info?.userId !== seat.user || data?.move !== moveNumber) return;
const { action, amount } = data;
const most = seat.bet + seat.chips;
if (action === "fold") apply(seat, "fold");
else if (action === "check" && seat.bet === toCall) apply(seat, "check");
else if (action === "call" && seat.bet < toCall) apply(seat, "call");
else if (action === "raise" && Number.isInteger(amount) && amount > toCall && amount <= most) {
if (amount - toCall >= minRaise || amount === most) apply(seat, "raise", amount);
}
});
// Someone joins, or comes back with a new page: the table, and their own cards if they have any.
room.subscribe((_type, _count, id, data) => {
if (!playing && seats.length === 0) return;
inTurn(game, () => room.sendTo(id, "table", table()));
const seat = seats.find((s) => s.user === String(data?.user_id));
if (seat?.cards) inTurn(game, () => room.sendTo(id, "cards", { hand: handNumber, cards: seat.cards }));
});
room.refused((code) => console.error(`${ROOM}: ${code}`));
console.log(`Dealer in ${ROOM}: ${START_CHIPS} chips, blinds ${SMALL_BLIND}/${BIG_BLIND}`);
pikoman.revoked(() => {
console.error("The bot's key was deleted.");
process.exit(1);
});The cards, in Node: hand.mjs
// hand.mjs: the cards. A card is two letters, rank and suit: "As" is the ace of spades, "Td" the ten
// of diamonds. A hand's strength is a list of numbers, compared from the left.
import { randomInt } from "node:crypto";
const RANKS = "23456789TJQKA";
const SUITS = "shdc";
const NAMES = ["high card", "a pair", "two pair", "three of a kind", "a straight", "a flush", "a full house", "four of a kind", "a straight flush"];
/** A new deck, shuffled with the system's secure random numbers: Math.random can be predicted. */
export function shuffledDeck() {
const deck = [...RANKS].flatMap((rank) => [...SUITS].map((suit) => rank + suit));
for (let i = deck.length - 1; i > 0; i--) {
const j = randomInt(i + 1);
[deck[i], deck[j]] = [deck[j], deck[i]];
}
return deck;
}
/** The strength of exactly five cards: [category, then the ranks that break a tie]. */
function strength(cards) {
const ranks = cards.map((card) => RANKS.indexOf(card[0])).sort((a, b) => b - a);
const count = new Map();
for (const rank of ranks) count.set(rank, (count.get(rank) ?? 0) + 1);
// The biggest group first, then the highest rank: a full house reads [three's rank, pair's rank].
const groups = [...count].sort((a, b) => b[1] - a[1] || b[0] - a[0]);
const shape = groups.map(([, n]) => n).join("");
const order = groups.map(([rank]) => rank);
const flush = cards.every((card) => card[1] === cards[0][1]);
let high = -1; // the top card of a straight
if (groups.length === 5) {
if (ranks[0] - ranks[4] === 4) high = ranks[0];
else if (ranks.join() === "12,3,2,1,0") high = 3; // A-2-3-4-5: the ace counts low, the five is top
}
if (flush && high >= 0) return [8, high];
if (shape === "41") return [7, ...order];
if (shape === "32") return [6, ...order];
if (flush) return [5, ...ranks];
if (high >= 0) return [4, high];
if (shape === "311") return [3, ...order];
if (shape === "221") return [2, ...order];
if (shape === "2111") return [1, ...order];
return [0, ...ranks];
}
/** Above zero when strength a beats b, zero for a tie. */
export function compare(a, b) {
for (let i = 0; i < a.length; i++) {
if (a[i] !== b[i]) return a[i] - b[i];
}
return 0;
}
/** The best five of a player's two cards and the board: { score, name }. */
export function bestHand(cards) {
let score = null;
for (let skipA = 0; skipA < cards.length; skipA++) {
for (let skipB = skipA + 1; skipB < cards.length; skipB++) {
const five = cards.filter((_, i) => i !== skipA && i !== skipB);
const candidate = strength(five);
if (!score || compare(candidate, score) > 0) score = candidate;
}
}
return { score, name: NAMES[score[0]] };
}Python: poker.py
"""poker.py: a Texas Hold'em dealer, one bot per room, with play chips, in Python. Each player's two
cards go to that player alone; the deck stays with the bot, so there is nothing in the room to read.
Members write "!sit" to take a seat and "!deal" to start. Python 3.10+.
pip install -r requirements.txt, then python poker.py (reads .env).
Its key (dashboard, New API key): rights subscribe and client_events, kind presence,
channels: prefix presence-poker-.
Players are told apart by the user_id your backend signed into their join token, which the server
passes on with each event: {"user_id": "42", "name": "Ada"}. Nothing a player sends says who they are.
Play chips only: no real money and nothing that can be cashed in. Playing for money is gambling,
and needs a license.
"""
import asyncio
import time
from collections import deque
from hand import best_hand, shuffled_deck
from pikoman_bot import Bot, env
ROOM = env("ROOM", "presence-poker-1")
START_CHIPS = int(env("START_CHIPS", "1000"))
BIG_BLIND = int(env("BIG_BLIND", "20"))
SMALL_BLIND = BIG_BLIND // 2
TURN_SECONDS = float(env("TURN_SECONDS", "30"))
PAUSE_SECONDS = float(env("PAUSE_SECONDS", "5")) # between two hands
MAX_SEATS = 8
STREETS = {"preflop": (3, "flop"), "flop": (1, "turn"), "turn": (1, "river")} # the next street, and its new cards
bot = Bot("Dealer", access=3)
room = bot.join(ROOM)
# A connection may send ten client events a second, so everything waits its turn, nine a second.
# The game goes first and the chat lines after it, so a player sees the next turn at once.
moves = deque()
chat = deque()
pumping = None # the task that sends, kept so it is not collected while it runs
async def pump():
global pumping
while moves or chat:
send = moves.popleft() if moves else chat.popleft()
await send()
await asyncio.sleep(0.11)
pumping = None
def in_turn(lane, send):
global pumping
lane.append(send)
if pumping is None:
pumping = asyncio.create_task(pump())
def tell(event, data):
in_turn(moves, lambda: room.send(event, data))
def tell_to(member, event, data):
in_turn(moves, lambda: room.send_to(member, event, data))
def say(text):
in_turn(chat, lambda: room.send("message", {"text": text}))
class Seat:
def __init__(self, user, name):
self.user, self.name = user, name
self.chips = START_CHIPS
self.clear()
def clear(self):
self.bet = 0 # this street
self.total = 0 # this hand
self.folded = True
self.cards = None
def pay(self, amount):
paid = min(amount, self.chips)
self.chips -= paid
self.bet += paid
self.total += paid
class Game:
def __init__(self):
self.seats = []
self.dealer = -1
self.playing = False # a hand is on
self.stage = None
self.deck = []
self.board = []
self.hand = 0
self.to_call = 0 # what everyone must have bet on this street
self.min_raise = 0
self.to_act = set() # the seats that still have to answer
self.turn = -1
self.move = 0 # a move names the one it answers, so a late or doubled click does nothing
self.deadline = 0
self.timer = None
def live(self):
return [s for s in self.seats if not s.folded]
def table(self):
"""What everyone may know, and nothing else: never the deck, never a player's cards."""
turn = self.seats[self.turn].user if self.playing and self.to_act else None
return {
"hand": self.hand,
"stage": self.stage,
"board": self.board,
"pot": sum(s.total for s in self.seats),
"dealer": self.seats[self.dealer].user if 0 <= self.dealer < len(self.seats) else None,
"turn": turn,
"move": self.move,
"toCall": self.to_call,
"minRaise": self.to_call + self.min_raise,
"seconds": max(0, round(self.deadline - time.monotonic())),
"seats": [{"user": s.user, "name": s.name, "chips": s.chips, "bet": s.bet, "folded": s.folded} for s in self.seats],
}
def later(self, seconds, fn):
self.cancel()
self.timer = asyncio.get_running_loop().call_later(seconds, fn)
def cancel(self):
if self.timer:
self.timer.cancel()
self.timer = None
def start_hand(self):
# Whoever is out of chips or has left the room loses the seat. The button moves to the next one that stays.
here = {str(m.get("user_id")) for m in room.members.values() if isinstance(m, dict)}
def stays(s):
return s.chips > 0 and s.user in here
button = None
for i in range(1, len(self.seats) + 1):
seat = self.seats[(self.dealer + i) % len(self.seats)]
if stays(seat):
button = seat
break
self.seats = [s for s in self.seats if stays(s)]
if len(self.seats) < 2:
self.playing = False
self.stage = None
self.board = []
for s in self.seats:
s.clear()
say("Waiting for two players: write !sit, then !deal.")
return tell("table", self.table())
self.dealer = self.seats.index(button)
self.playing = True
self.hand += 1
self.stage = "preflop"
self.deck = shuffled_deck()
self.board = []
for s in self.seats:
s.clear()
s.folded = False
s.cards = [self.deck.pop(), self.deck.pop()]
# Two players: the button is the small blind and acts first before the flop.
count = len(self.seats)
small = self.dealer if count == 2 else (self.dealer + 1) % count
big = (self.dealer + (1 if count == 2 else 2)) % count
self.seats[small].pay(SMALL_BLIND)
self.seats[big].pay(BIG_BLIND)
self.to_call = BIG_BLIND
self.min_raise = BIG_BLIND
self.to_act = {s for s in self.seats if s.chips > 0}
self.turn = big
for s in self.seats:
tell_to({"user_id": s.user}, "cards", {"hand": self.hand, "cards": s.cards})
say(f"Hand {self.hand}: {self.seats[self.dealer].name} has the button.")
self.advance()
def advance(self):
"""The next thing that happens: the next player's turn, the next street, or the end of the hand."""
self.cancel()
if len(self.live()) == 1:
return self.finish()
self.to_act = {s for s in self.to_act if s.chips > 0 and not s.folded}
if not self.to_act:
return self.next_street()
self.turn = (self.turn + 1) % len(self.seats)
while self.seats[self.turn] not in self.to_act:
self.turn = (self.turn + 1) % len(self.seats)
self.move += 1
self.deadline = time.monotonic() + TURN_SECONDS
# The bot's clock decides. A player who is gone or asleep checks if they can, and otherwise folds.
seat = self.seats[self.turn]
def out_of_time():
say(f"{seat.name} ran out of time.")
self.apply(seat, "check" if seat.bet == self.to_call else "fold")
self.later(TURN_SECONDS, out_of_time)
tell("table", self.table())
def next_street(self):
for s in self.seats:
s.bet = 0
self.to_call = 0
self.min_raise = BIG_BLIND
if self.stage not in STREETS:
return self.finish()
cards, self.stage = STREETS[self.stage]
self.board = self.board + [self.deck.pop() for _ in range(cards)]
self.turn = self.dealer
self.to_act = {s for s in self.live() if s.chips > 0}
# At most one player can still bet: nobody to bet against, so the rest of the board just comes.
if len(self.to_act) < 2:
self.to_act = set()
tell("table", self.table())
return self.later(1.5, self.next_street)
self.advance()
def apply(self, seat, action, amount=0):
self.cancel()
if action == "fold":
seat.folded = True
say(f"{seat.name} folds.")
elif action == "check":
say(f"{seat.name} checks.")
elif action == "call":
seat.pay(self.to_call - seat.bet)
say(f"{seat.name} calls.")
else:
# A raise is the total to bet this street. A smaller one than min_raise is fine only all in.
if amount - self.to_call >= self.min_raise:
self.min_raise = amount - self.to_call
seat.pay(amount - seat.bet)
self.to_call = amount
self.to_act = {s for s in self.live() if s.chips > 0}
say(f"{seat.name} raises to {amount}.")
self.to_act.discard(seat)
self.advance()
def finish(self):
"""The pots, then the cards of those still in. Folded hands stay hidden, and so does a hand nobody called."""
self.cancel()
players = self.live()
showdown = len(players) > 1
best = {s: best_hand(s.cards + self.board) if len(self.board) == 5 else (None, "") for s in players}
def after_button(s):
return (self.seats.index(s) - self.dealer - 1) % len(self.seats)
# One pot per all-in size: a player wins only as much from each other player as they put in.
won = {}
levels = sorted({s.total for s in players})
below = 0
for i, level in enumerate(levels):
top = float("inf") if i == len(levels) - 1 else level # the last pot takes what folded players left above it
pot = sum(min(s.total, top) - min(s.total, below) for s in self.seats)
below = level
entitled = [s for s in players if s.total >= level]
if showdown:
high = max(best[s][0] for s in entitled)
winners = [s for s in entitled if best[s][0] == high]
else:
winners = entitled
# A split pot's odd chips go to the first winner after the button, one each.
winners.sort(key=after_button)
for n, s in enumerate(winners):
won[s] = won.get(s, 0) + pot // len(winners) + (1 if n < pot % len(winners) else 0)
for s, chips in won.items():
s.chips += chips
tell("showdown", {
"hand": self.hand,
"board": self.board,
"players": [{"user": s.user, "name": s.name, "cards": s.cards, "hand": best[s][1]} for s in players] if showdown else [],
"winners": [{"user": s.user, "name": s.name, "won": chips} for s, chips in won.items()],
})
for s, chips in won.items():
say(f"{s.name} wins {chips} with {best[s][1]}." if showdown else f"{s.name} wins {chips}.")
for s in self.seats:
s.clear()
self.to_act = set()
self.to_call = 0
self.stage = "between"
tell("table", self.table())
self.later(PAUSE_SECONDS, self.start_hand)
game = Game()
# "!sit" and "!deal" in the room. A seat taken during a hand plays from the next one.
@room.on("message")
def message(sender, data, info):
text = str((data or {}).get("text", "")).strip()
if text not in ("!sit", "!deal") or room.is_bot(sender):
return
user = info.get("user_id")
if user is None:
return say("Your join token needs a user_id to play.")
seated = next((s for s in game.seats if s.user == user), None)
if text == "!sit":
if seated:
return
if len(game.seats) >= MAX_SEATS:
return say("The table is full.")
name = str(room.member(sender).get("name", "Someone"))[:40]
game.seats.append(Seat(user, name))
say(f"{name} sits down with {START_CHIPS} chips.")
tell("table", game.table())
elif seated and not game.playing:
game.start_hand()
# A move, sent to the bot alone: room.sendTo(bot, "move", { move: 7, action: "raise", amount: 60 }).
@room.on("move")
def move(sender, data, info):
if not game.playing or not game.to_act or not isinstance(data, dict):
return
seat = game.seats[game.turn]
if seat not in game.to_act or info.get("user_id") != seat.user or data.get("move") != game.move:
return
action, amount = data.get("action"), data.get("amount")
most = seat.bet + seat.chips
if action == "fold":
game.apply(seat, "fold")
elif action == "check" and seat.bet == game.to_call:
game.apply(seat, "check")
elif action == "call" and seat.bet < game.to_call:
game.apply(seat, "call")
elif action == "raise" and type(amount) is int and game.to_call < amount <= most:
if amount - game.to_call >= game.min_raise or amount == most:
game.apply(seat, "raise", amount)
# Someone joins, or comes back with a new page: the table, and their own cards if they have any.
@room.on_join
def joined(connection, data):
if not game.playing and not game.seats:
return
tell_to(connection, "table", game.table())
user = str(data.get("user_id")) if isinstance(data, dict) else None
seat = next((s for s in game.seats if s.user == user), None)
if seat and seat.cards:
tell_to(connection, "cards", {"hand": game.hand, "cards": seat.cards})
print(f"Dealer in {ROOM}: {START_CHIPS} chips, blinds {SMALL_BLIND}/{BIG_BLIND}")
bot.start()The cards, in Python: hand.py
"""hand.py: the cards. A card is two letters, rank and suit: "As" is the ace of spades, "Td" the ten
of diamonds. A hand's strength is a tuple of numbers, and Python compares tuples from the left.
"""
import secrets
from collections import Counter
from itertools import combinations
RANKS = "23456789TJQKA"
SUITS = "shdc"
NAMES = ["high card", "a pair", "two pair", "three of a kind", "a straight", "a flush", "a full house", "four of a kind", "a straight flush"]
def shuffled_deck():
"""A new deck, shuffled with the system's secure random numbers: random.shuffle can be predicted."""
deck = [rank + suit for rank in RANKS for suit in SUITS]
secrets.SystemRandom().shuffle(deck)
return deck
def strength(cards):
"""The strength of exactly five cards: (category, then the ranks that break a tie)."""
ranks = sorted((RANKS.index(card[0]) for card in cards), reverse=True)
# The biggest group first, then the highest rank: a full house reads (three's rank, pair's rank).
groups = sorted(Counter(ranks).items(), key=lambda group: (-group[1], -group[0]))
shape = [n for _, n in groups]
order = [rank for rank, _ in groups]
flush = all(card[1] == cards[0][1] for card in cards)
high = -1 # the top card of a straight
if len(groups) == 5:
if ranks[0] - ranks[4] == 4:
high = ranks[0]
elif ranks == [12, 3, 2, 1, 0]: # A-2-3-4-5: the ace counts low, the five is top
high = 3
if flush and high >= 0:
return (8, high)
if shape == [4, 1]:
return (7, *order)
if shape == [3, 2]:
return (6, *order)
if flush:
return (5, *ranks)
if high >= 0:
return (4, high)
if shape == [3, 1, 1]:
return (3, *order)
if shape == [2, 2, 1]:
return (2, *order)
if shape == [2, 1, 1, 1]:
return (1, *order)
return (0, *ranks)
def best_hand(cards):
"""The best five of a player's two cards and the board: (score, name)."""
score = max(strength(five) for five in combinations(cards, 5))
return score, NAMES[score[0]]The player’s page: page.js
// page.js: the poker table on a player's page. Your own cards come to you alone; everything else
// is what the bot tells the room. Trust only the bot: room.isBot(from) is set by the server, never
// by a member. `room` is the presence channel your page already joined:
// pikoman.listenPresence("presence-poker-1"). `me` is the user_id your backend signed for this player.
const SUITS = { s: "♠", h: "♥", d: "♦", c: "♣" };
const show = (card) => (card[0] === "T" ? "10" : card[0]) + SUITS[card[1]];
export function playPoker(room, box, me) {
const line = () => box.appendChild(document.createElement("p"));
let mine = [];
let last = null;
const bot = () => [...room.getBots().keys()][0];
const button = (text, parent, click) => {
const b = parent.appendChild(Object.assign(document.createElement("button"), { type: "button", textContent: text }));
b.addEventListener("click", click);
return b;
};
const move = (action, amount) => room.sendTo(bot(), "move", { move: last.move, action, amount });
function draw(t) {
last = t;
box.replaceChildren();
line().textContent = `Board: ${t.board.map(show).join(" ") || "none yet"}. Pot: ${t.pot}.`;
const seats = box.appendChild(document.createElement("ul"));
for (const s of t.seats) {
const marks = [s.user === t.dealer && "button", s.user === t.turn && "to act", s.folded && t.stage !== "between" && "folded"];
seats.appendChild(document.createElement("li")).textContent =
`${s.name}: ${s.chips} chips, bet ${s.bet} ${marks.filter(Boolean).join(", ")}`;
}
const seat = t.seats.find((s) => s.user === me);
line().textContent = mine.length ? `Your cards: ${mine.map(show).join(" ")}` : "";
if (!seat) return button("Take a seat", box, () => room.send("message", { text: "!sit" }));
if (t.stage === null) return button("Deal", box, () => room.send("message", { text: "!deal" }));
if (t.turn !== me) return;
const owed = t.toCall - seat.bet;
button("Fold", box, () => move("fold"));
if (owed > 0) button(`Call ${Math.min(owed, seat.chips)}`, box, () => move("call"));
else button("Check", box, () => move("check"));
const most = seat.bet + seat.chips;
if (most > t.toCall) {
const amount = box.appendChild(Object.assign(document.createElement("input"), {
type: "number",
min: Math.min(t.minRaise, most),
max: most,
value: Math.min(t.minRaise, most),
}));
amount.setAttribute("aria-label", "Raise to");
button("Raise to", box, () => move("raise", Number(amount.value)));
}
}
room.onClient("table", (from, table) => {
if (room.isBot(from)) draw(table);
});
room.onClient("cards", (from, { cards }, info) => {
if (!room.isBot(from) || !info?.direct) return;
mine = cards;
if (last) draw(last);
});
room.onClient("showdown", (from, { players, winners }) => {
if (!room.isBot(from)) return;
mine = [];
if (last) draw(last);
for (const p of players) line().textContent = `${p.name}: ${p.cards.map(show).join(" ")}, ${p.hand}`;
for (const w of winners) line().textContent = `${w.name} wins ${w.won}.`;
});
}pikoman_bot.py: what every Python bot here shares, the tokens, the websocket and the rooms
"""pikoman_bot.py: what a Pikoman bot needs in Python, in one file next to the bot.
- Rest: signed HTTP requests (publish, to one member, stats), each with a token made for that
one request, as KEYS.md says: the method, the path and a hash of the exact body.
- Bot and Room: a websocket that joins presence channels with join tokens signed by the bot's
own API key, keeps the member list (and which members are bots), hands client events to your
handlers, sends, sends to one member and kicks. It reconnects by itself.
Python 3.10+, pip install -r requirements.txt (websockets, BSD-3-Clause; cryptography,
Apache-2.0 OR BSD-3-Clause). Reads PIKOMAN_APP, PIKOMAN_API_KEY and, for tests, PIKOMAN_WS and
PIKOMAN_HTTP from the environment or from a .env file beside the bot.
"""
import asyncio
import base64
import hashlib
import inspect
import json
import os
import random
import signal
import time
import urllib.error
import urllib.parse
import urllib.request
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed, InvalidHandshake
LOCAL = ("localhost", "127.0.0.1", "[::1]")
def load_env(path=".env"):
"""KEY=value lines from .env; what the environment already has wins."""
try:
with open(path, encoding="utf-8") as file:
lines = file.read().splitlines()
except FileNotFoundError:
return
for line in lines:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
name, value = (part.strip() for part in line.split("=", 1))
value = value.strip("\"'")
if value and name not in os.environ:
os.environ[name] = value
load_env()
def env(name, default=""):
"""A setting from the environment; empty counts as not set."""
return os.environ.get(name) or default
def b64(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def checked(address: str, secure: str) -> str:
"""https:// or wss://; plain http:// or ws:// only to this computer, so keys never go in the clear."""
parsed = urllib.parse.urlsplit(address)
if parsed.scheme != secure and not (parsed.scheme == secure[:-1] and parsed.hostname in ("localhost", "127.0.0.1", "::1")):
raise ValueError(f"{address}: use {secure}:// (plain {secure[:-1]}:// only for localhost)")
return address.rstrip("/")
class PikomanError(Exception):
def __init__(self, status, text):
super().__init__(f"{status}: {text}")
self.status = status
class Key:
"""An API key's secret, sk_<key id>.<seed>: the id goes in each token's header, the seed signs it."""
def __init__(self, secret: str):
if not secret.startswith("sk_") or "." not in secret:
raise ValueError("PIKOMAN_API_KEY looks like sk_k_….…: the secret the dashboard showed once")
key_id, seed = secret[3:].split(".", 1)
self.id = key_id
self._key = Ed25519PrivateKey.from_private_bytes(base64.urlsafe_b64decode(seed + "=" * (-len(seed) % 4)))
def sign(self, claims: dict) -> str:
"""An EdDSA JWT with the key id in its header."""
header = {"alg": "EdDSA", "typ": "JWT", "kid": self.id}
signed = b64(json.dumps(header).encode()) + "." + b64(json.dumps(claims).encode())
return signed + "." + b64(self._key.sign(signed.encode()))
def __repr__(self):
return f"Key({self.id}, [hidden])"
def target(member) -> dict:
"""A connection id ("c_…"), or {"user_id": "42"} for every connection of that user."""
if isinstance(member, dict):
return {"to_user": str(member["user_id"])}
return {"to": member}
class Rest:
"""The HTTP API, every request signed with the API key and bound to that one request."""
def __init__(self, app=None, api_key=None, address=None):
self.app = app or env("PIKOMAN_APP")
if not self.app:
raise ValueError("Set PIKOMAN_APP: your app id, app-…")
self.key = Key(api_key or env("PIKOMAN_API_KEY"))
self.address = checked(address or env("PIKOMAN_HTTP", "https://next-api.pikoman.se"), "https")
def request(self, method: str, path: str, data=None) -> dict:
body = b"" if data is None else json.dumps(data).encode()
now = int(time.time())
token = self.key.sign({
"iss": self.app, "iat": now, "exp": now + 60, "htm": method, "htu": path,
"bh": b64(hashlib.sha256(body).digest()),
})
request = urllib.request.Request(self.address + path, data=body if data is not None else None, method=method, headers={
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
})
try:
with urllib.request.urlopen(request, timeout=10) as response:
text = response.read()
return json.loads(text) if text else {}
except urllib.error.HTTPError as e:
raise PikomanError(e.code, e.read().decode(errors="replace")) from None
def publish(self, channel: str, event: str, data=None, to=None) -> dict:
"""POST /{channel}/{event}; with `to`, to one member only."""
path = f"/{urllib.parse.quote(channel, safe='=@,.;')}/{urllib.parse.quote(event, safe='=@,.;')}"
if to is not None:
path += "?" + urllib.parse.urlencode(target(to))
return self.request("POST", path, data)
def stats(self, channel: str) -> dict:
return self.request("GET", f"/{urllib.parse.quote(channel, safe='=@,.;')}/stats")
async def call(fn, *args):
"""Runs a handler, plain or async; an error in it is printed and the bot goes on."""
try:
result = fn(*args)
if inspect.isawaitable(result):
await result
except Exception as e: # noqa: BLE001: one handler must never stop the bot
print(f"handler {getattr(fn, '__name__', fn)}: {e!r}")
def client_name(event: str) -> str:
return event if event.startswith("client-") else "client-" + event
class Room:
"""A presence channel the bot is in: its members, which of them are bots, and its events."""
def __init__(self, bot, name: str):
self.bot = bot
self.name = name
self.members = {} # connection id -> member data, as their join token gave it
self.bots = {} # connection id -> the API key id a bot joined with: only the server says
self._handlers = {}
self._any = []
self._joins = []
self._leaves = []
self._errors = []
def on(self, event: str):
"""@room.on("message"): fn(sender, data, info) for that client event (client-message)."""
def add(fn):
self._handlers.setdefault(client_name(event), []).append(fn)
return fn
return add
def on_any(self, fn):
"""@room.on_any: fn(sender, event, data, info) for every client event."""
self._any.append(fn)
return fn
def on_join(self, fn):
"""@room.on_join: fn(connection, data) when a member joins."""
self._joins.append(fn)
return fn
def on_leave(self, fn):
"""@room.on_leave: fn(connection, data) when a member leaves."""
self._leaves.append(fn)
return fn
def on_error(self, fn):
"""@room.on_error: fn(code), such as join_refused, not_allowed or rate_limited."""
self._errors.append(fn)
return fn
def is_bot(self, connection: str) -> bool:
return connection in self.bots
def member(self, connection: str) -> dict:
data = self.members.get(connection)
return data if isinstance(data, dict) else {}
async def send(self, event: str, data=None):
"""A client event to the other members (at most 10 a second per connection)."""
await self.bot.send({"op": "send", "channel": self.name, "event": client_name(event), "data": data, "echo": False})
async def send_to(self, member, event: str, data=None):
"""To one member only: a connection id, or {"user_id": "42"}."""
await self.bot.send({"op": "send", "channel": self.name, "event": client_name(event), "data": data, "echo": False, **target(member)})
async def kick(self, member, reason=None, ban_seconds=None):
"""Out of this room: needs the kick right (ban, with ban_seconds)."""
message = {"op": "kick", "channel": self.name, **target(member)}
if reason:
message["reason"] = reason
if ban_seconds:
message["ban_seconds"] = ban_seconds
await self.bot.send(message)
async def handle(self, message: dict):
op = message.get("op")
if op == "members":
self.members = dict(message.get("members") or {})
self.bots = dict(message.get("bots") or {})
elif op == "member_joined":
self.members[message["id"]] = message.get("data")
if message.get("bot"):
self.bots[message["id"]] = message["bot"]
for fn in self._joins:
await call(fn, message["id"], message.get("data"))
elif op == "member_left":
data = self.members.pop(message["id"], None)
self.bots.pop(message["id"], None)
for fn in self._leaves:
await call(fn, message["id"], data)
elif op == "client_event":
sender, event, data = message.get("from"), message.get("event", ""), message.get("data")
# Set by the server, never by the sender: direct (to this bot alone), user_id, bot.
info = {k: message[k] for k in ("direct", "user_id", "bot") if k in message}
for fn in self._handlers.get(event, []):
await call(fn, sender, data, info)
for fn in self._any:
await call(fn, sender, event, data, info)
elif op == "error":
print(f"{self.name}: {message.get('code')}")
for fn in self._errors:
await call(fn, message.get("code"))
class Bot:
"""One websocket for the bot, in as many rooms as it joins. await bot.run() keeps it going."""
def __init__(self, name: str, access: int = 3, app=None, api_key=None, address=None):
self.app = app or env("PIKOMAN_APP")
if not self.app:
raise ValueError("Set PIKOMAN_APP: your app id, app-…")
self.key = Key(api_key or env("PIKOMAN_API_KEY"))
self.address = checked(address or env("PIKOMAN_WS", "wss://next.pikoman.se"), "wss")
self.data = {"name": name} # what the members see next to "bot"
self.access = access # 3: may speak (the key needs client_events); 0: listens
self.rooms = {}
self.id = None
self._ws = None
self._ready = []
def join(self, channel: str) -> Room:
room = self.rooms[channel] = Room(self, channel)
return room
def on_ready(self, fn):
"""@bot.on_ready: fn() each time the bot is connected and has asked to join its rooms."""
self._ready.append(fn)
return fn
def token(self, channel: str) -> str:
"""A join token for this connection and channel, signed with the bot's API key."""
return self.key.sign({
"connection": self.id, "channel": channel, "access": self.access,
"data": self.data, "exp": int(time.time()) + 300,
})
async def send(self, message: dict):
if self._ws is not None:
await self._ws.send(json.dumps(message))
def start(self, *tasks):
"""Runs the bot (and any other coroutines, such as a web server) until Ctrl+C or SIGTERM."""
def stop(*_):
raise KeyboardInterrupt
signal.signal(signal.SIGTERM, stop) # as from systemd or podman stop: the same as Ctrl+C
async def main():
await asyncio.gather(self.run(), *tasks)
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Stopped.")
async def run(self):
delay = 1
paused = 60 # after 4402: the first wait, doubling to five minutes
while True:
try:
async with connect(f"{self.address}/{self.app}") as ws:
self._ws = ws
async for raw in ws:
message = json.loads(raw)
if message.get("op") == "connected":
delay = 1
self.id = message["id"]
for name in self.rooms:
await self.send({"op": "join", "channel": name, "token": self.token(name)})
for fn in self._ready:
await call(fn)
elif message.get("channel") in self.rooms:
await self.rooms[message["channel"]].handle(message)
elif message.get("op") == "error":
print("pikoman:", message.get("code"))
except ConnectionClosed as e:
code = e.rcvd.code if e.rcvd is not None else None
if code == 4403:
raise SystemExit("The bot's key was deleted.") from None
if code == 4401:
raise SystemExit("The app disconnected the bot.") from None
if code == 4402:
# The key's bot rights are paused (the app's owner is off Pro): not for good.
# Try again in minutes, and the bot is back on its own when Pro is.
self._ws = None
self.id = None
print(f"The bot's key is paused until the app is on Pro again; trying again in about {paused // 60} min.")
await asyncio.sleep(random.uniform(paused / 2, paused))
paused = min(paused * 2, 300)
continue
print("connection closed:", code if code is not None else "no close frame")
except (OSError, InvalidHandshake) as e:
# Refused at the door too, such as HTTP 429 when too many are connected: try again.
print("could not connect:", e)
self._ws = None
self.id = None
paused = 60
# Back again after a random wait, longer each time it fails, up to 30 s.
await asyncio.sleep(random.uniform(0.5, delay))
delay = min(delay * 2, 30)# Copy to .env and fill in. Never commit .env.
# Your app id, app-…
PIKOMAN_APP=
# The bot's own API key, sk_k_….… (rights subscribe and client_events, prefix presence-poker-)
PIKOMAN_API_KEY=
# The room to deal in (default presence-poker-1)
ROOM=
# Chips each player starts with (default 1000)
START_CHIPS=
# The big blind; the small blind is half (default 20)
BIG_BLIND=
# Seconds a player has to move (default 30)
TURN_SECONDS=
# Seconds between two hands (default 5)
PAUSE_SECONDS=
# Another server than wss://next.pikoman.se, for tests
PIKOMAN_WS={
"name": "pikoman-poker-bot",
"version": "1.0.0",
"private": true,
"type": "module",
"description": "A Texas Hold'em dealer for a Pikoman room, with play chips: each player's cards to that player alone, side pots, the bot's own clock.",
"license": "MIT",
"scripts": {
"start": "node --env-file=.env poker.mjs"
},
"engines": {
"node": ">=22"
},
"dependencies": {
"@pikoman/ws": "1.0.0-rc.11"
}
}websockets==15.0.1
cryptography==46.0.3Run it
# unpack poker-bot.zip, then in the folder:
npm install
cp .env.example .env # fill in PIKOMAN_APP and PIKOMAN_API_KEY
npm startpython -m venv .venv && . .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -r requirements.txt
python poker.pyIdeas to take it further
- A TV view: the table big on a screen, each phone showing only its own two cards.
- Keep the chips between games: a leaderboard in your own database, still play chips.
- Other games: blackjack, Uno or a word game use the same three parts, one event to everyone, one to each player alone, and moves to the bot alone.