Simple peer-to-peer game of Battleship based on the classic board game.

battleship.py

#!/usr/bin/env python3
#
# A peer-to-peer game of Battleship because who doesn't like a bit of P2P gaming? Enjoy :D
#
# First edit: November 18th 2015
# Last edit: March 5th 2019
# Author: Victor Azzam

import re
import signal
import socket
import readline
from time import sleep
from sys import argv, exit
from random import randint
from threading import Thread

board = """
• • • • • • • • • •
• • • • • • • • • •
• • • • • # # # • •
• • # • • • • • • •
• • # • # • • • • •
• • # • # • # • • •
• • # • # • # • • •
• • # • # • # • # #
• • • • • • • • • •
• • • • • • • • • •
""".strip()

display = """
                   *********************
                  ** BATTLESHIP GAME **
                 *********************

      A B C D E F G H I J   |   A B C D E F G H I J
    ┌─────────────────────┐ | ┌─────────────────────┐
  1 │ {} │ | │ {} │ 1
  2 │ {} │ | │ {} │ 2
  3 │ {} │ | │ {} │ 3
  4 │ {} │ | │ {} │ 4
  5 │ {} │ | │ {} │ 5
  6 │ {} │ | │ {} │ 6
  7 │ {} │ | │ {} │ 7
  8 │ {} │ | │ {} │ 8
  9 │ {} │ | │ {} │ 9
 10 │ {} │ | │ {} │ 10
    └─────────────────────┘ | └─────────────────────┘
       {name:^17}    |    {friend:^17}
"""

clear = lambda: print("\033[1J\033[H", end="")

class Timeout(Exception): pass

def handler(signum, frame):
	raise Timeout()

signal.signal(signal.SIGALRM, handler)

def ip_check(H):
	return re.search(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", H) and all(int(x) < 256 for x in H.split("."))

class Player:
	def __init__(self, name, board):
		self.name = name if name else f"Player-{randint(1000,10000)}"
		self.dice = "".join(chr(randint(32, 127)) for i in range(16))
		self.board = board
	def fire(self, C, R):
		self.board[R][C] = (" ", "-")[self.board[R][C] == "#"]
		return self.board[R][C] == "-"

class Network:
	def __init__(self, IP):
		self.data = ""
		self.sock, self.serv = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM) for i in "  "]
		self.thread = Thread(target=self.update, daemon=True)
		self.thread.start()
		self.connect()
	def update(self):
		self.serv.bind(("", 12321))
		while 1:
			data, addr = self.serv.recvfrom(128)
			self.data = data.decode()
	def send(self, data):
		self.sock.send(data.encode())
	def connect(self):
		try:
			print("Sleeping for 3 seconds...")
			sleep(3)
			print("Connecting...")
			self.sock.connect((IP, 12321))
			print("Connected, sending dice: " + repr(you.dice))
			self.send(you.dice)
			print("Dice sent, waiting for response")
			while not self.data:
				pass
			print(self.data)
		except:
			quit(f"Could not connect to {friend}")

def setup():
	clear()
	IP = name = input("Your name: ").strip()[:17]
	friend = input("Your friend: ").strip()[:17]
	while not ip_check(IP):
		IP = input("Friend's IP: ").strip()
	return name, friend, IP

def quit(message):
	print(message)
	conn.sock.close()
	conn.serv.close()
	exit()

try:
	alpha = "ABCDEFGHIJ"
	units = board.count("#")
	name, friend, IP = setup() if not argv[1:4] else argv[1:4]
	you = Player(name, board=[row.split() for row in board.split("\n")])
	enemy = Player(friend, board=[list("•" * 10) for j in range(10)])
	conn = Network(IP)
	turn = conn.data < you.dice
	while 1:
		if "".join("".join(x) for x in enemy.board).count("-") == units:
			quit("\nCongratulations, you win!\n")
		elif "".join("".join(x) for x in you.board).count("-") == units:
			quit("\nYou lose :(\n")

		clear()
		conn.data = ""
		print(display.format(*[" ".join(i) for a, b in zip(you.board, enemy.board) for i in (a, b)], name=name, friend=friend))
		if turn:
			fire = input("It's your turn, fire: ").strip().upper()
			if fire[0:1] not in alpha or fire[1:2] not in "0123456789":
				input("Must be in the format CR where C = letter and R = number.")
				continue
			C, R = alpha.find(fire[0]), int(fire[1]) - 1
			if enemy.board[R][C] != "•":
				input("You've already tried that, try another one.")
				continue
			conn.send(fire[:2])
			if conn.data == "OFFLINE":
				quit(f"\nYour friend {friend} has disconnected :/")
			while not conn.data:
				pass
			enemy.board[R][C] = (" ", "-")[conn.data == "Y"]
		else:
			print("Awaiting missile...")
			while not conn.data:
				pass
			if conn.data == "OFFLINE":
				quit(f"\nYour friend {friend} has disconnected :/")
			C, R = alpha.find(conn.data[0]), int(conn.data[1]) - 1
			conn.send(("N", "Y")[you.fire(C, R)])
		turn = not turn
except Exception as e:
	exit(f"Error: {e.message}")
except (KeyboardInterrupt, EOFError):
	conn.send("OFFLINE")
	quit("\nExiting...")