Py-Hanabi/database.py

284 lines
9.6 KiB
Python
Raw Normal View History

import json
import psycopg2
from typing import Optional, Dict
## global connection
conn = psycopg2.connect("dbname=hanab-live user=postgres")
## cursor
cur = conn.cursor()
# cur.execute("DROP TABLE games;")
# conn.commit()
# exit(0)
## check if table exists, else create it
2023-03-02 15:18:08 +01:00
def create_games_table():
tablename = "games"
cur.execute(
"SELECT EXISTS (SELECT FROM pg_tables WHERE schemaname = 'public' AND tablename = '{}');".format(tablename))
2023-03-02 15:18:08 +01:00
a = cur.fetchone()
if a[0] is False:
print("Creating table '{}'".format(tablename))
cur.execute(
"CREATE TABLE {} ("
"id INT PRIMARY KEY,"
"num_players SMALLINT NOT NULL,"
"score SMALLINT NOT NULL,"
"seed TEXT NOT NULL,"
"variant_id SMALLINT NOT NULL,"
"deck_plays BOOLEAN,"
"one_extra_card BOOLEAN,"
"one_less_card BOOLEAN,"
"all_or_nothing BOOLEAN,"
"num_turns SMALLINT,"
"actions TEXT"
")".format(tablename))
2023-03-02 15:18:08 +01:00
conn.commit()
2023-03-02 15:18:08 +01:00
# else:
# print("table already exists")
2023-03-02 15:18:08 +01:00
def create_seeds_table():
tablename = 'seeds'
cur.execute(
"SELECT EXISTS (SELECT FROM pg_tables WHERE schemaname = 'public' AND tablename = '{}');".format(tablename))
2023-03-02 15:18:08 +01:00
a = cur.fetchone()
if a[0] is False:
print("Creating table '{}'".format(tablename))
cur.execute(
"CREATE TABLE {} ("
"seed TEXT NOT NULL PRIMARY KEY,"
"num_players SMALLINT NOT NULL,"
"variant_id SMALLINT NOT NULL,"
"feasible BOOLEAN," # theoretical solvability
"max_score_theoretical SMALLINT," # if infeasible, max score
"deck VARCHAR(60)"
")".format(tablename))
2023-03-02 15:18:08 +01:00
conn.commit()
def init_static_tables():
# check if table already exists
create = False
tables = ['suits', 'colors', 'suit_colors', 'variants', 'variant_suits']
for table in tables:
cur.execute(
"SELECT EXISTS (SELECT FROM pg_tables WHERE schemaname = 'public' AND tablename = '{}');".format(table)
)
a = cur.fetchone()
if a[0] is False:
create = True
if not create:
2023-05-12 23:15:53 +02:00
pass
# init tables in database
with open("variant_suits_schema.sql", "r") as f:
cur.execute(f.read())
with open("suits.json", "r") as f:
suits: Dict = json.loads(f.read())
with open('variants.json', 'r') as f:
variants = json.loads(f.read())
suits_to_reverse = set()
for var in variants:
for suit in var['suits']:
if 'Reversed' in suit:
suits_to_reverse.add(suit.replace(' Reversed', ''))
for suit in suits:
name: str = suit['name']
display_name: str = suit.get('displayName', name)
abbreviation = suit.get('abbreviation', name[0].upper())
all_colors = suit.get('allClueColors', False)
no_color_clues = suit.get('noClueColors', False)
all_ranks = suit.get('allClueRanks', False)
no_rank_clues = suit.get('noClueRanks', False)
prism = suit.get('prism', False)
dark = suit.get('oneOfEach', False)
assert([all_colors, no_color_clues, prism].count(True) <= 1)
assert(not all([no_rank_clues, all_ranks]))
2023-05-12 23:15:53 +02:00
color_clues = 2 if all_colors else (0 if no_color_clues else 1)
rank_clues = 2 if all_ranks else (0 if no_rank_clues else 1)
2023-05-12 23:15:53 +02:00
clue_colors = suit.get('clueColors', [name] if (color_clues == 1 and not prism) else [])
for rev in [False, True]:
if rev is True and name not in suits_to_reverse:
break
suit_name = name
suit_name += ' Reversed' if rev else ''
cur.execute(
2023-05-12 23:15:53 +02:00
"INSERT INTO suits (name, display_name, abbreviation, rank_clues, color_clues, dark, reversed, prism)"
"VALUES"
"(%s, %s, %s, %s, %s, %s, %s, %s)",
2023-05-12 23:15:53 +02:00
(suit_name, display_name, abbreviation, rank_clues, color_clues, dark, rev, prism)
)
cur.execute(
"SELECT id FROM suits WHERE name = %s",
(suit_name,)
)
suit_id = cur.fetchone()
for color in clue_colors:
2023-05-12 15:36:00 +02:00
if not rev:
cur.execute(
"INSERT INTO colors (name) VALUES (%s)"
"ON CONFLICT (name) DO NOTHING",
(color,)
)
cur.execute(
"SELECT id FROM colors WHERE name = %s",
(color,)
)
color_id = cur.fetchone()
cur.execute(
"INSERT INTO suit_colors (suit_id, color_id) VALUES"
"(%s, %s)",
(suit_id, color_id)
)
for var in variants:
var_id = var['id']
name = var['name']
clue_starved = var.get('clueStarved', False)
throw_it_in_a_hole = var.get('throwItInHole', False)
alternating_clues = var.get('alternatingClues', False)
synesthesia = var.get('synesthesia', False)
chimneys = var.get('chimneys', False)
funnels = var.get('funnels', False)
no_color_clues = var.get('clueColors', None) == []
no_rank_clues = var.get('clueRanks', None) == []
empty_color_clues = var.get('colorCluesTouchNothing', False)
empty_rank_clues = var.get('rankCluesTouchNothing', False)
odds_and_evens = var.get('oddsAndEvens', False)
up_or_down = var.get('upOrDown', False)
critical_fours = var.get('criticalFours', False)
suits = var['suits']
num_suits = len(suits)
special_rank_no_ranks = var.get('specialNoClueRanks', False)
special_rank_all_ranks = var.get('specialAllClueRanks', False)
special_rank_no_colors = var.get('specialNoClueColors', False)
special_rank_all_colors = var.get('specialAllClueColors', False)
special_rank = var.get('specialRank', None)
2023-05-13 00:01:35 +02:00
special_deceptive = var.get('specialDeceptive', False)
assert(not all([special_rank_all_ranks, special_rank_no_ranks]))
assert(not all([special_rank_all_colors, special_rank_no_colors]))
special_rank_ranks = 2 if special_rank_all_ranks else (0 if special_rank_no_ranks else 1)
special_rank_colors = 2 if special_rank_all_colors else (0 if special_rank_no_colors else 1)
cur.execute(
"INSERT INTO variants ("
"id, name, clue_starved, throw_it_in_a_hole, alternating_clues, synesthesia, chimneys, funnels,"
"no_color_clues, no_rank_clues, empty_color_clues, empty_rank_clues, odds_and_evens, up_or_down,"
2023-05-13 00:01:35 +02:00
"critical_fours, num_suits, special_rank, special_rank_ranks, special_rank_colors, special_deceptive"
")"
"VALUES"
2023-05-13 00:01:35 +02:00
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
var_id, name, clue_starved, throw_it_in_a_hole, alternating_clues, synesthesia, chimneys, funnels,
no_color_clues, no_rank_clues, empty_color_clues, empty_rank_clues, odds_and_evens, up_or_down,
2023-05-13 00:01:35 +02:00
critical_fours, num_suits, special_rank, special_rank_ranks, special_rank_colors, special_deceptive
)
)
for index, suit in enumerate(suits):
cur.execute(
"SELECT id FROM suits WHERE name = %s",
(suit,)
)
suit_id = cur.fetchone()
if suit_id is None:
print(suit)
cur.execute(
"INSERT INTO variant_suits (variant_id, suit_id, index) VALUES (%s, %s, %s)",
(var_id, suit_id, index)
)
conn.commit()
2023-03-02 15:18:08 +01:00
create_games_table()
create_seeds_table()
init_static_tables()
class Game():
def __init__(self, info=None):
self.id = -1
self.num_players = -1
self.score = -1
self.seed = ""
self.variant_id = -1
self.deck_plays = None
self.one_extra_card = None
self.one_less_card = None
self.all_or_nothing = None
self.num_turns = None
if type(info) == dict:
self.__dict__.update(info)
@staticmethod
def from_tuple(t):
g = Game()
g.id = t[0]
g.num_players = t[1]
g.score = t[2]
g.seed = t[3]
g.variant_id = t[4]
g.deck_plays = t[5]
g.one_extra_card = t[6]
g.one_less_card = t[7]
g.all_or_nothing = t[8]
g.num_turns = t[9]
return g
def __eq__(self, other):
return self.__dict__ == other.__dict__
def load(game_id: int) -> Optional[Game]:
cur.execute("SELECT * from games WHERE id = {};".format(game_id))
a = cur.fetchone()
if a is None:
return None
else:
return Game.from_tuple(a)
def store(game: Game):
stored = load(game.id)
if stored is None:
# print("inserting game with id {} into DB".format(game.id))
cur.execute(
"INSERT INTO games"
"(id, num_players, score, seed, variant_id)"
"VALUES"
"(%s, %s, %s, %s, %s);",
(game.id, game.num_players, game.score, game.seed, game.variant_id)
)
else:
if not stored == game:
print("Already stored game with id {}, aborting".format(game.id))
print("Stored game is: {}".format(stored.__dict__))
print("New game is: {}".format(game.__dict__))
def commit():
conn.commit()