test_client.py

import numpy as np
import sys
import socket
import selectors
import traceback
import json
import io
import struct

import queue

class TMessage:
	def __init__(self, selector, sock, addr, request, actor):
		self.selector = selector
		self.sock = sock
		self.addr = addr
		self.request = request
		self.actor = actor
		self._recv_buffer = b""
		self._send_buffer = b""
		self._request_queued = False
		self._jsonheader_len = None
		self.jsonheader = None
		self.response = None

	def _set_selector_events_mask(self, mode):
		"""Set selector to listen for events: mode is 'r', 'w', or 'rw'."""
		if mode == "r":
			events = selectors.EVENT_READ
		elif mode == "w":
			events = selectors.EVENT_WRITE
		elif mode == "rw":
			events = selectors.EVENT_READ | selectors.EVENT_WRITE
		else:
			raise ValueError(f"Invalid events mask mode {mode!r}.")
		self.selector.modify(self.sock, events, data=self)

	def _read(self):
		try:
			# Should be ready to read
			data = self.sock.recv(4096)
		except BlockingIOError:
			# Resource temporarily unavailable (errno EWOULDBLOCK)
			pass
		else:
			if data:
				self._recv_buffer += data
			else:
				raise RuntimeError("Peer closed.")

	def _write(self):
		if self._send_buffer:
			print(f"Sending {self._send_buffer!r} to {self.addr}")
			try:
				# Should be ready to write
				sent = self.sock.send(self._send_buffer)
			except BlockingIOError:
				# Resource temporarily unavailable (errno EWOULDBLOCK)
				pass
			else:
				self._send_buffer = self._send_buffer[sent:]

	def _json_encode(self, obj, encoding):
		return json.dumps(obj, ensure_ascii=False).encode(encoding)

	def _json_decode(self, json_bytes, encoding):
		tiow = io.TextIOWrapper(
			io.BytesIO(json_bytes), encoding=encoding, newline=""
		)
		obj = json.load(tiow)
		tiow.close()
		return obj

	def _create_message(
		self, *, content_bytes, content_type, content_encoding
	):
		jsonheader = {
			"byteorder": sys.byteorder,
			"content-type": content_type,
			"content-encoding": content_encoding,
			"content-length": len(content_bytes),
		}
		jsonheader_bytes = self._json_encode(jsonheader, "utf-8")
		message_hdr = struct.pack(">H", len(jsonheader_bytes))
		message = message_hdr + jsonheader_bytes + content_bytes
		return message

	def _process_response_json_content(self):
		content = self.response
		result = content.get("result")
		print(f"Got result: {result}")

	def _process_response_binary_content(self):
		content = self.response
		print(f"Got response: {content!r}")

	def process_events(self, mask):
		if mask & selectors.EVENT_READ:
			self.read()
		if mask & selectors.EVENT_WRITE:
			self.write()

	def read(self):
		self._read()

		if self._jsonheader_len is None:
			self.process_protoheader()

		if self._jsonheader_len is not None:
			if self.jsonheader is None:
				self.process_jsonheader()

		if self.jsonheader:
			if self.response is None:
				self.process_response()

	def write(self):
		print("TMessage->write()")
		if not self._request_queued:
			self.queue_request()

		self._write()

		if self._request_queued:
			if not self._send_buffer:
				# Set selector to listen for read events, we're done writing.
				self._set_selector_events_mask("r")

	def close(self):
		print(f"Closing connection to {self.addr}")
		try:
			self.selector.unregister(self.sock)
		except Exception as e:
			print(
				f"Error: selector.unregister() exception for "
				f"{self.addr}: {e!r}"
			)

		try:
			self.sock.close()
		except OSError as e:
			print(f"Error: socket.close() exception for {self.addr}: {e!r}")
		finally:
			# Delete reference to socket object for garbage collection
			self.sock = None

	def queue_request(self):
		content = self.request["content"]
		content_type = self.request["type"]
		content_encoding = self.request["encoding"]
		if content_type == "text/json":
			req = {
				"content_bytes": self._json_encode(content, content_encoding),
				"content_type": content_type,
				"content_encoding": content_encoding,
			}
		else:
			req = {
				"content_bytes": content,
				"content_type": content_type,
				"content_encoding": content_encoding,
			}
		message = self._create_message(**req)
		self._send_buffer += message
		self._request_queued = True

	def process_protoheader(self):
		print("TMessage->process_protoheader()")
		hdrlen = 2
		if len(self._recv_buffer) >= hdrlen:
			self._jsonheader_len = struct.unpack(
				">H", self._recv_buffer[:hdrlen]
			)[0]
			self._recv_buffer = self._recv_buffer[hdrlen:]

	def process_jsonheader(self):
		print("TMessage->process_jsonheader()")
		hdrlen = self._jsonheader_len
		if len(self._recv_buffer) >= hdrlen:
			self.jsonheader = self._json_decode(
				self._recv_buffer[:hdrlen], "utf-8"
			)
			self._recv_buffer = self._recv_buffer[hdrlen:]
			for reqhdr in (
				"byteorder",
				"content-length",
				"content-type",
				"content-encoding",
			):
				if reqhdr not in self.jsonheader:
					raise ValueError(f"Missing required header '{reqhdr}'.")

	def process_response(self):
		print("TMessage->process_response()")
		content_len = self.jsonheader["content-length"]
		if not len(self._recv_buffer) >= content_len:
			return
		data = self._recv_buffer[:content_len]
		self._recv_buffer = self._recv_buffer[content_len:]
		if self.jsonheader["content-type"] == "text/json":
			encoding = self.jsonheader["content-encoding"]
			self.response = self._json_decode(data, encoding)
			action = self.response.get('action')
			print(f"action={action}")
			if action == "client":
				client_action = self.response.get(action)
				if client_action == "new":
					actor["cid"] = self.response.get(client_action)
					actor["properties"]["x"] = self.response.get("x")
					actor["properties"]["y"] = self.response.get("y")

			if action == "fov":
				result = self.response.get(action)
				fov = np.reshape(result,(11,11))
				print(f"{action}={fov!r}")
#				self._process_response_json_content()
		# Close when response has been processed
		self.close()

"""
MAIN CLIENT PROGRAM
"""
sel = selectors.DefaultSelector()

def create_request(action):
	return dict(
		type="text/json",
		encoding="utf-8",
		content=action,
	)

def start_connection(host, port, request, actor):
	addr = (host, port)
	print(f"Starting connection to {addr}")
	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	sock.setblocking(False)
	sock.connect_ex(addr)
	events = selectors.EVENT_READ | selectors.EVENT_WRITE
	message = TMessage(sel, sock, addr, request, actor)
	sel.register(sock, events, data=message)

host = "192.168.1.106"
port = 54321

actions = queue.SimpleQueue()

actions.put(
	create_request(
		{
			"cid": "", # Client ID provided by the server
			"action": "client", # Start a client connection with server
			"client": "new", # Ask the server for a new Client ID (other options are "open", "save", "quit")
			"actor": "@", # Symbolized as!
		}
	)
)

actor = {
	"active": True,
	"cid": None, # Client ID
	"character": {
		"name": "SethNai"

	},
	"properties": {
		"x": 1,
		"y": 1,
		"r": 9, # Field of Vision radius,
	},
}

xlow = actor["properties"]["x"] - actor["properties"]["r"] - 1
xhigh = actor["properties"]["x"] + actor["properties"]["r"] + 1
ylow = actor["properties"]["y"] - actor["properties"]["r"] - 1
yhigh = actor["properties"]["y"] + actor["properties"]["r"] + 1
actions.put(
	create_request(
		{
			"cid": actor["cid"],
			"action": "fov",
			"x": actor["properties"]["x"],
			"y": actor["properties"]["y"],
			"r": actor["properties"]["r"],
		}
	)
)

try:
	while actor["active"]:
		try:
			request = actions.get(False)
			print(f"request: {request!r}")
			start_connection(host, port, request, actor)
		except:
			actor["active"] = False

		events = sel.select(timeout=5)
		for key, mask in events:
			message = key.data
			try:
				message.process_events(mask)
			except Exception:
				print(
					f"Main: Error: Exception for {message.addr}:\n"
					f"{traceback.format_exc()}"
				)
				message.close()
		
		# Check for actor being active
		#print(f"actor: {actor!r}")

		# Check for a socket being monitored to continue.
		if not sel.get_map():
			break
except KeyboardInterrupt:
	print("Caught keyboard interrupt, exiting")
finally:
	sel.close()