346 lines
11 KiB
Python
Executable File
346 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import sys
|
|
import sqlite3 as sql
|
|
import os
|
|
from os.path import isfile as fileexists
|
|
import subprocess
|
|
import json
|
|
from gomoku import Gomoku
|
|
from isola import Isola
|
|
from brandubh import Brandubh
|
|
|
|
version = '0.8.1'
|
|
home_folder = os.path.expanduser('~')
|
|
username = os.path.split(home_folder)[-1]
|
|
database = '/var/lib/oberon/oberon.sqlite'
|
|
basefile = '/var/lib/oberon/'
|
|
|
|
|
|
def print_header():
|
|
print('\nO B E R O N v{}\n'.format(version))
|
|
|
|
|
|
##############################
|
|
######------------------#######
|
|
###### Top Lvl Cmds ########
|
|
######------------------#######
|
|
##############################
|
|
|
|
# Binds to the HISTORY command
|
|
def show_game_record():
|
|
q = "SELECT rowid, p1, p2, winner, game_type FROM games WHERE winner is not NULL and (p1 = ? or p2 = ?)"
|
|
v = (username, username)
|
|
res = db_do(q, v)
|
|
if res == False:
|
|
print('Database error. Contact the sysadmin or try another time')
|
|
print_header()
|
|
print('{:^8} {:^10} {:^10} {:^10} {:^12}'.format('ID', 'Player 1', 'Player 2', 'Winner', 'Game Type'))
|
|
print('-------- ---------- ---------- ---------- ------------')
|
|
for x in res:
|
|
print('{:^8} {:^10} {:^10} {:^10} {:^12}'.format(x[0], x[1], x[2], x[3], x[4]))
|
|
if not len(res):
|
|
print('You have not completed any games yet...')
|
|
print('\nTo view a historical game, run "oberon play [game_id]"')
|
|
return True
|
|
|
|
|
|
# Binds to the PLAY command
|
|
def play_game(gid):
|
|
q = "SELECT rowid, p1, p2, game_board, turn, game_type, winner FROM games WHERE rowid = ?"
|
|
v = (gid,)
|
|
res = db_do(q,v)
|
|
if not res:
|
|
print('Invalid game id')
|
|
return False
|
|
elif res[0][1] != username and res[0][2] != username:
|
|
print('Invalid selection. You are not playing game {}'.format(gid))
|
|
return False
|
|
|
|
state = {
|
|
'id': res[0][0],
|
|
'p1': res[0][1],
|
|
'p2': res[0][2],
|
|
'board': json.loads(res[0][3]),
|
|
'turn_number': res[0][4],
|
|
'game_type': res[0][5],
|
|
'winner': res[0][6],
|
|
'current_ps_turn': res[0][2] if res[0][4] % 2 == 0 else res[0][1],
|
|
'piece_index': 1 if res[0][4] % 2 == 0 else 0
|
|
}
|
|
|
|
if state['game_type'] == 'gomoku':
|
|
game = Gomoku(state)
|
|
elif state['game_type'] == 'isola':
|
|
game = Isola(state)
|
|
elif state['game_type'] == 'brandubh':
|
|
game = Brandubh(state)
|
|
|
|
print_header()
|
|
game.print_challenge_text()
|
|
game.print_board()
|
|
|
|
if not state['current_ps_turn'] == username:
|
|
print('\nIt is your opponents turn.\n')
|
|
return True
|
|
|
|
if game.winner:
|
|
print('You are viewing a game that has already finished')
|
|
return True
|
|
|
|
play = game.get_input()
|
|
|
|
if not play:
|
|
sys.exit(0)
|
|
else:
|
|
q2 = "UPDATE games SET game_board = ?, turn = ?, winner = ? WHERE rowid = ?"
|
|
v2 = (json.dumps(play['board']), state['turn_number'] + 1, play['winner'], gid)
|
|
res2 = db_do(q2, v2, True)
|
|
if play['winner']:
|
|
enemy = state['p1'] if state['p1'] != username else state['p2']
|
|
msg_path = basefile + '{}'.format(enemy)
|
|
writemode = 'a' if fileexists(msg_path) else 'w'
|
|
with open(msg_path, writemode) as f:
|
|
f.write('{} defeated you in {} game {}\n'.format(username, state['game_type'], gid))
|
|
|
|
if not res2:
|
|
print('Error updating database, sorry.\nTry again later?\n...or contact the sysadmin.')
|
|
return False
|
|
print(play['message'])
|
|
|
|
|
|
|
|
# Binds to the LIST command
|
|
def list_games():
|
|
q = "SELECT rowid, p1, p2, game_type, game_status, turn FROM games WHERE winner is NULL and (p1 = ? or p2 = ?)"
|
|
v = (username, username)
|
|
res = db_do(q, v)
|
|
if res == False:
|
|
print('Database error. Contact the sysadmin or try another time.')
|
|
return False
|
|
print_header()
|
|
print('{:^8} {:^10} {:^10} {:^22} {:^12}'.format('ID', 'Player 1', 'Player 2', 'Game Status', 'Game Type'))
|
|
print('-------- ---------- ---------- ---------------------- ------------')
|
|
for x in res:
|
|
pturn = x[2] if x[5] % 2 == 0 else x[1]
|
|
print("{:^8} {:^10} {:^10} {:^22} {:^12}".format(x[0], x[1], x[2], pturn +"'s turn", x[3]))
|
|
if not len(res):
|
|
print('You are not currently playing any games...')
|
|
print('')
|
|
return True
|
|
|
|
|
|
# Binds to the HELP command
|
|
def display_help():
|
|
print_header()
|
|
print('syntax: oberon [command [option]]\n')
|
|
print('{:20} {}'.format('help','display this message'))
|
|
print('{:20} {}'.format('list', 'display your currently open games'))
|
|
print('{:20} {}'.format('history', 'display your win/loss record'))
|
|
print('{:20} {}'.format('create [user]', 'create a game against the user provided'))
|
|
print('{:20} {}'.format('play [game_id]', 'make a move or view the board for the game id provided'))
|
|
print('')
|
|
print('if you do not want to play games simply add a file named ".killoberon" to your home directory and users will not be able to create games with you')
|
|
print('')
|
|
|
|
|
|
# Binds to the CREATE command
|
|
def create_game(enemy):
|
|
if enemy == username:
|
|
print('This system is designed for multiuser games. You cannot create a game against yourself.')
|
|
return False
|
|
enemies = get_user_list()
|
|
|
|
if not enemy in enemies:
|
|
print('There is no avialable player with the name {}.\nGame not created.'.format(enemy))
|
|
return False
|
|
|
|
if fileexists('/home/{}/.killoberon'.format(enemy)):
|
|
print('{} does not wish to play games in oberon. Please try creating a game with a different user.'.format(enemy))
|
|
return False
|
|
|
|
print_header()
|
|
print('Which game would you like to play?\n\n(1) Gomoku/Five In a Row\n(2) Isola\n(3) Brandubh\n\nor (Q)uit')
|
|
while True:
|
|
game = input('> ')
|
|
if game.lower() == 'q':
|
|
sys.exit(0)
|
|
elif not game in ['1', '2', '3']:
|
|
continue
|
|
|
|
break
|
|
|
|
if game == '1':
|
|
game_board = [[' · ' for y in range(15)] for x in range(15)]
|
|
game_choice = 'gomoku'
|
|
elif game == '2':
|
|
board = [[ ' - ' for y in range(7)] for x in range(7)]
|
|
board[0][3] = ' X '
|
|
board[6][3] = ' O '
|
|
game_board = {'board': board, 'p1': [0,3], 'p2': [6,3]}
|
|
game_choice = 'isola'
|
|
elif game == '3':
|
|
game_board = [[ ' - ' for y in range(7)] for x in range(7)]
|
|
game_board[0][0] = ' * '
|
|
game_board[0][6] = ' * '
|
|
game_board[6][0] = ' * '
|
|
game_board[6][6] = ' * '
|
|
for x in [0,1,5,6]:
|
|
game_board[3][x] = ' a '
|
|
game_board[x][3] = ' a '
|
|
for x in [2,4]:
|
|
game_board[3][x] = ' d '
|
|
game_board[x][3] = ' d '
|
|
game_board[3][3] = ' D '
|
|
game_choice = 'brandubh'
|
|
|
|
player = '0'
|
|
print('')
|
|
while not player in ['1','2']:
|
|
print('Play as player 1 or player 2 (enter the number)')
|
|
player = input('> ')
|
|
|
|
if player == '2':
|
|
temp = enemy
|
|
enemy = username
|
|
player = temp
|
|
else:
|
|
player = username
|
|
|
|
q = "INSERT INTO games VALUES (?, ?, ?, ?, ?, ?, ?)"
|
|
v = (player, enemy, game_choice, 'playing', json.dumps(game_board), 1, None)
|
|
res = db_do(q, v, True)
|
|
if res:
|
|
print('Game between {} and {} created!'.format(username, enemy))
|
|
return True
|
|
return False
|
|
|
|
|
|
# Binds to the AVAILABLE command
|
|
def available_moves():
|
|
q = "SELECT rowid, p1, p2, turn FROM games WHERE winner is null and (p1 = ? or p2 = ?)"
|
|
v = (username, username)
|
|
res = db_do(q, v)
|
|
p1 = [x[0] for x in res if x[1] == username and x[3] % 2 != 0]
|
|
p2 = [x[0] for x in res if x[2] == username and x[3] % 2 == 0]
|
|
print('- - - -')
|
|
if len(p1) + len(p2) > 0:
|
|
p1 = ', '.join(str(x) for x in p1)
|
|
p2 = ', '.join(str(x) for x in p2)
|
|
spacer = ', ' if p1 and p2 else ''
|
|
print('\033[1;32mIt is your turn in the following game(s): {}{}{}\033[0m'.format(p1, spacer, p2))
|
|
else:
|
|
print('It is not your turn in any games...\n')
|
|
|
|
|
|
##############################
|
|
######------------------#######
|
|
###### Sys Utilities ########
|
|
######------------------#######
|
|
##############################
|
|
def parse_args():
|
|
args = sys.argv
|
|
arg_len = len(args)
|
|
if arg_len == 1:
|
|
display_help()
|
|
available_moves()
|
|
sys.exit(0)
|
|
if arg_len == 2:
|
|
if args[1] == 'list':
|
|
l = list_games()
|
|
if l:
|
|
sys.exit(0)
|
|
sys.exit(1)
|
|
elif args[1] == 'history':
|
|
h = show_game_record()
|
|
if h:
|
|
sys.exit(0)
|
|
sys.exit(1)
|
|
elif args[1] == 'available':
|
|
available_moves()
|
|
sys.exit(0)
|
|
elif args[1] == 'create':
|
|
print('Must include an opponent name.\nFind one in chat or cspc!')
|
|
sys.exit(1)
|
|
elif args[1] == 'play':
|
|
print('Must include a game id.\nRun "oberon list" to view your open games.')
|
|
sys.exit(1)
|
|
elif args[1] == 'help':
|
|
display_help()
|
|
sys.exit(0)
|
|
else:
|
|
print('Invalid arguments. Run "oberon help".')
|
|
sys.exit(1)
|
|
elif arg_len == 3:
|
|
if args[1] == 'create':
|
|
if create_game(args[2]):
|
|
sys.exit(0)
|
|
sys.exit(1)
|
|
elif args[1] == 'play':
|
|
try:
|
|
game_id = int(args[2])
|
|
except:
|
|
print('Invalid game id.')
|
|
sys.exit(1)
|
|
play_game(game_id)
|
|
sys.exit(0)
|
|
else:
|
|
print('Invalid arguments. Run "oberon help".')
|
|
sys.exit(1)
|
|
else:
|
|
print('Unknown paramaters: {}'.format(' '.join(args[3:])))
|
|
sys.exit(1)
|
|
|
|
|
|
|
|
def get_user_list():
|
|
res = subprocess.run(['awk', '-F', ':', '{if ($7 == "/usr/local/bin/colorsh") print $1}', '/etc/passwd'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
res_list = res.stdout.decode('utf-8').split('\n')
|
|
if res_list[0] == '':
|
|
return []
|
|
return res_list
|
|
|
|
|
|
##############################
|
|
######------------------#######
|
|
###### DB Utilities ########
|
|
######------------------#######
|
|
##############################
|
|
def check_and_build_db(db_path):
|
|
if not os.path.isfile(db_path):
|
|
conn = sql.connect(db_path)
|
|
c = conn.cursor()
|
|
|
|
c.execute("CREATE TABLE games (p1 text NOT NULL, p2 text DEFAULT NULL, game_type text NOT NULL, game_status text DEFAULT NULL, game_board text, turn INTEGER, winner text DEFAULT NULL)")
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def db_do(query, var=False, noresval=False):
|
|
global messages
|
|
db_path = database
|
|
if os.path.isfile(db_path):
|
|
conn = sql.connect(db_path)
|
|
c = conn.cursor()
|
|
if var:
|
|
c.execute(query, var)
|
|
else:
|
|
c.execute(query)
|
|
if noresval:
|
|
out = c.rowcount
|
|
else:
|
|
out = []
|
|
for row in c:
|
|
out.append(row)
|
|
conn.commit()
|
|
conn.close()
|
|
return out
|
|
else:
|
|
return False
|
|
|
|
|
|
if __name__ == '__main__':
|
|
check_and_build_db(database)
|
|
parse_args()
|