gammasimulator

#!/usr/bin/python3
import re
import os
import sys
import socket
import datetime
import threading
import random
import struct

from FriendlyArgumentParser import FriendlyArgumentParser
from RXBuffer import RXBuffer
from ReaderThreads import SocketReaderThread
from TimeAlert import TimeAlert
from GSOnline import GSOnline
from GSProtocolHandler import GSProtocolHandler

parser = FriendlyArgumentParser(prog = sys.argv[0], description = "Tool to simulate a Gamma Scout device for development and debugging", add_help = False)
parser.add_argument("-p", "--protocol", metavar = "version", type = str, choices = [ "v1", "v2", "v3" ], default = "v2", help = "Specifies the device protocol the simulated Gamma Scout uses. Possible options are %(choices)s, default is %(default)s")
parser.add_argument("-m", "--model", metavar = "model", type = str, choices = [ "basic", "alert", "online" ], default = "alert", help = "Simulated Gamma Scout model. Possible options are %(choices)s, default is %(default)s")
parser.add_argument("socket", metavar = "socket", type = str, help = "UNIX socket that is used for Gamma Scout simulation")
args = parser.parse_args(sys.argv[1:])

class SimulationLogFile(object):
	def __init__(self, filename):
		self._filename = filename
		f = open(self._filename, "rb")
		data = f.read()

		idx = data.find(0)
		self._infotext = data[:idx].decode("utf-8")
		data = data[idx + 1:]

		(self._buffill, ) = struct.unpack("<L", data[:4])
		self._data = data[4:]
		assert((len(self._data) % 32) == 0)

	def getsimresponse(self):
		response = [ "", "GAMMA-SCOUT Protokoll" ]
		for index in range(0, len(self._data), 32):
			chunk = self._data[index : index + 32]
			checksum = sum(chunk) & 0xff
			chunk = chunk + bytes([ checksum ])

			hexstr = "".join([ "%02x" % (c) for c in chunk ])
			response.append(hexstr)
		return response

	@property
	def infotext(self):
		return self._infotext

	@property
	def buffill(self):
		return self._buffill

	@property
	def filename(self):
		return self._filename

class SimModel(object):
	def __init__(self, args):
		self._args = args
		if self._args.protocol in [ "v2", "v3" ]:
			self._mode = GSProtocolHandler.MODE_STANDARD
#			self._logfile = SimulationLogFile("simdata/gs_v2_alert_65083.bin")
			self._logfile = SimulationLogFile("simdata/gs_v2_alert_maxed.bin")
		else:
			self._mode = GSProtocolHandler.MODE_PC
		self._timeoffset = datetime.timedelta(0)
		self._onlineintvl = 0
		self._onlinetime = TimeAlert()

	def command(self, connhandlerthread):
		while True:
			result = self.wait_command(connhandlerthread)
			if result is not None:
				return result
			if (result is None) and (connhandlerthread.haveeof()):
				return

			if self._mode == GSProtocolHandler.MODE_ONLINE:
				if self._onlinetime.triggeredreset():
					interval = self._onlinetime.getinterval()
					activity = abs(random.gauss(0.6, 0.3))
					counts = round(interval * activity)
					return [ "I%04x%06x" % (interval, counts) ]


	def wait_command(self, connhandlerthread):
		cmd = connhandlerthread.waitforbytes(1, 1.0)
		if cmd is None:
			return

		print("<-", cmd)
		response = [ "" ]
		unknowncmd = False

		if self._mode == GSProtocolHandler.MODE_PC:
			if self._args.protocol in [ "v2", "v3" ]:
				if cmd == "v":
					datestr = (datetime.datetime.utcnow() + self._timeoffset).strftime("%d.%m.%y %H:%M:%S")
					if self._args.protocol == "v2":
						response.append("Version 6.05 032123 %04x %s" % (self._logfile.buffill, datestr))
					else:
						gammascout_datetime = datetime.datetime.now() + datetime.timedelta((random.random() - 0.5) * 20)
						response.append("Version 6.05 032123 %04x %s %s" % (self._logfile.buffill, datestr, gammascout_datetime.strftime("%d.%m.%y %H:%M:%S")))
				elif cmd == "z":
					response.append("Protokollspeicher wieder frei")
					self._logfile = SimulationLogFile("simdata/gs_v2_alert_00009.bin")
				elif cmd == "t":
					remainder = connhandlerthread.waitforbytes(12, 10.0)
					if remainder is None:
						print("gammascoututil did not send further bytes after 't'.")
						response = [ ]
					else:
						cmd += remainder
						time_re = re.compile("t(?P<day>[0-9]{2})(?P<month>[0-9]{2})(?P<year>[0-9]{2})(?P<hour>[0-9]{2})(?P<minute>[0-9]{2})(?P<second>[0-9]{2})")
						match = time_re.match(cmd)
						if not match:
							print("Unknown time format received")
							response = [ ]
						else:
							match = match.groupdict()
							match = { key: int(value) for (key, value) in match.items() }
							match = datetime.datetime(match["year"] + 2000, match["month"], match["day"], match["hour"], match["minute"], match["second"])
							self._timeoffset = match - datetime.datetime.utcnow()
							response.append("Datum und Zeit gestellt")
							self._logfile = SimulationLogFile("simdata/gs_v2_alert_00017.bin")
				elif cmd == "X":
					response.append("PC-Mode beendet")
					self._mode = GSProtocolHandler.MODE_STANDARD
				elif cmd == "i":
					# Device reset
					self._mode = GSProtocolHandler.MODE_STANDARD
					self._logfile = SimulationLogFile("simdata/gs_v2_alert_00009.bin")
					self._timeoffset = datetime.timedelta(0)
				elif cmd == "c":
					response.append("08 04 ab310b00")
					response.append("f3020000fb1de1f68601")
					response.append("cb2c00000600e6ff7602")
					response.append("ffffffff5b16acfc6702")
					response.append("00000000000000000000")
					response.append("00000000000000000000")
					response.append("00000000000000000000")
					response.append("00000000000000000000")
					response.append("00000000000000000000")
				elif cmd == "b":
					response = self._logfile.getsimresponse()
				else:
					unknowncmd = True
			else:
				if cmd == "v":
					response.append(" Version 5.43")
				elif (cmd == "d") or (cmd == "u"):
					cmd += connhandlerthread.waitforbytes(6, 1.0)
					time_re = re.compile("(?P<type>[du])(?P<par1>[0-9]{2})(?P<par2>[0-9]{2})(?P<par3>[0-9]{2})")
					match = time_re.match(cmd)
					if not match:
						print("Unknown date/time format")
						response = [ ]
					else:
						match = match.groupdict()
						if match["type"] == "d":
							response.append(" Datum gestellt ")
						else:
							response.append(" Zeit gestellt ")
				elif cmd == "b":
					response = [ line[:-1] for line in open("simdata/gs_basic_v1_full.txt", "r") ]
				elif cmd == "z":
					response.append(" Protokollspeicher wieder frei ")
				elif cmd == "t":
					# No idea what this command is or does
					response.append(" AUS ")
				else:
					unknowncmd = True

		elif self._mode == GSProtocolHandler.MODE_STANDARD:
			if cmd == "v":
				response.append("Standard")
			elif cmd == "P":
				response.append("PC-Mode gestartet")
				self._mode = GSProtocolHandler.MODE_PC
			elif (cmd == "O") and (self._args.model == "online"):
				response = [ "S00020000000000", "I0002000001" ]
				self._mode = GSProtocolHandler.MODE_ONLINE
			else:
				unknowncmd = True

		elif self._mode == GSProtocolHandler.MODE_ONLINE:
			if cmd == "v":
				response.append("ONLINE")
			elif cmd == "X":
				response.append("ONLINE-Mode beendet")
				self._mode = GSProtocolHandler.MODE_STANDARD
			elif cmd in "0123456789":
				self._onlineintvl = int(cmd)
				seconds = GSOnline.intervalcmd_to_time(self._onlineintvl)
				self._onlinetime.setinterval(seconds)
				response = [ ]
			else:
				unknowncmd = True

		if unknowncmd:
			print("Not understood command: '%s' in mode %s with model %s" % (cmd, self._mode, self._args.model))
			response = [ ]

		return response

class ConnectionHandlerThread(threading.Thread):
	def __init__(self, conn, model):
		threading.Thread.__init__(self)
		self._model = model
		self._conn = conn

		self._rxbuf = RXBuffer()
		self._readerthread = SocketReaderThread(self._conn, self._rxbuf.push, self._rxbuf.seteof)

	def haveeof(self):
		return self._rxbuf.haveeof()

	def waitforbytes(self, bytecnt, timeout = None):
		return self._rxbuf.waitforbytes(bytecnt, timeout)

	def run(self):
		self._readerthread.start()
		while True:
			rsp = self._model.command(self)
			if rsp is None:
				break
			for line in rsp:
				print("->", line)
				self._conn.send((line + "\r\n").encode("utf-8"))
		self._conn.close()
		print("Remote closed connection")

class Simulator():
	def __init__(self, args):
		self._args = args
		self._socket = None
		self._model = SimModel(args)

	def _bindsocket(self):
		self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
		try:
			os.unlink(self._args.socket)
		except OSError:
			pass
		self._socket.bind(self._args.socket)
		self._socket.listen(1)

	def _handle(self, conn):
		print("New connection %s" % (conn))
		handlerthread = ConnectionHandlerThread(conn, self._model)
		handlerthread.start()

	def simulate(self):
		self._bindsocket()
		print("Now simulating Gamma Scout model '%s' using protocol version %s on UNIX socket '%s'" % (self._args.model, self._args.protocol, self._args.socket))

		while True:
			(conn, addr) = self._socket.accept()
			self._handle(conn)

sim = Simulator(args)
try:
	sim.simulate()
except KeyboardInterrupt:
	try:
		os.unlink(args.socket)
	except OSError:
		pass