FrameDetector: Implement load_script

Author: oscar-rc1

zbnt/Encoding.py

@@ -16,6 +16,8 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 
+import struct
+
 # encode_x : number to bytes
 
 def encode_bool(value):
@@ -45,6 +47,12 @@ def encode_s32(value):
 def encode_s64(value):
 	return value.to_bytes(8, byteorder="little", signed=True)
 
+def encode_float(value):
+	return struct.pack("<f", value)
+
+def encode_double(value):
+	return struct.pack("<d", value)
+
 def encode_str(value):
 	return value.encode("UTF-8")
 
@@ -83,6 +91,12 @@ def decode_s32(value):
 def decode_s64(value):
 	return int.from_bytes(value[:8], byteorder="little", signed=True)
 
+def decode_float(value):
+	return struct.unpack("<f", value)[0]
+
+def decode_double(value):
+	return struct.unpack("<d", value)[0]
+
 def decode_str(value):
 	return value.encode("UTF-8")
 

zbnt/FrameDetector.py

@@ -16,6 +16,8 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 
+import re
+import math
 from enum import IntFlag
 
 from .Enums import *
@@ -44,6 +46,24 @@ class FrameDetector(AxiDevice):
 		])
 	}
 
+	_regex_comp_instr = re.compile("^(?:nop|(s?[lg]tq?|eq|or|and)(8|16|24|32|40|48|56|64|[fd])(l?)|eof)$")
+	_regex_edit_instr = re.compile("^(?:nop|setr|(set|(?:xn)?or|and|add|s?mul)(8|16|32|64|[fd])(l?)|drop|corrupt)$")
+
+	_edit_opcode_dict = {
+		"nop":     0b00_00_000_0,
+		"set":     0b00_00_001_0,
+		"setr":    0b00_01_001_0,
+		"and":     0b00_00_010_0,
+		"or":      0b00_01_010_0,
+		"xor":     0b00_10_010_0,
+		"xnor":    0b00_11_010_0,
+		"add":     0b00_00_011_0,
+		"mul":     0b00_00_100_0,
+		"smul":    0b00_10_100_0,
+		"drop":    0b00_10_000_0,
+		"corrupt": 0b00_01_000_0
+	}
+
 	class FeatureBits(IntFlag):
 		HAS_CMP_UNIT  = 1
 		HAS_EDIT_UNIT = 2
@@ -96,3 +116,230 @@ def receive_measurement(self, data):
 			ext_data = ext_data[:-log_width] + ext_data[-log_width + ext_count % log_width:]
 
 		self.measurement_handler(self.id, (time, match_dir, match_mask, ext_data))
+
+	def load_script(self, path):
+		comparator_instr = [(0, 0)] * self.max_script_size
+		editor_instr = [(0, 0)] * self.max_script_size
+		extr_instr = [0] * self.max_script_size
+
+		i = 0
+		offset = 0
+		in_section = 0
+
+		with open(path) as file_handle:
+			for line in file_handle:
+				try:
+					line = line[:line.index("#")]
+				except ValueError:
+					pass
+
+				line = line.strip().lower().split()
+				i += 1
+
+				if len(line) > 2:
+					raise ValueError("line {0}: Invalid syntax".format(i))
+
+				if len(line) == 0:
+					continue
+
+				instr = line[0]
+				param = line[1] if len(line) == 2 else ""
+
+				if instr[0] == ".":
+					try:
+						in_section = ["", ".comp", ".extr", ".edit"].index(instr)
+					except ValueError:
+						raise ValueError("line {0}: Invalid section type: '{1}'".format(i, instr))
+
+					if len(line) == 2:
+						try:
+							offset = int(param)
+						except ValueError:
+							raise ValueError("line {0}: Invalid section offset: '{1}'".format(i, param))
+					else:
+						offset = 0
+
+					continue
+
+				if in_section == 0:
+					raise ValueError("line {0}: Invalid instruction, not inside a section".format(i))
+
+				if in_section == 1:
+					for index, opcode, param in self.__parse_comparator_instr(i, instr, param, offset):
+						if isinstance(index, tuple):
+							for j in range(index[0], index[1]):
+								offset = j + 1
+								comparator_instr[j] = (opcode, param)
+						else:
+							offset = index + 1
+							comparator_instr[index] = (opcode, param)
+				elif in_section == 2:
+					begin, end, value = self.__parse_extractor_instr(i, instr, param, offset)
+
+					offset = end
+
+					for j in range(begin, end):
+						extr_instr[j] = value
+				else:
+					for index, opcode, param in self.__parse_editor_instr(i, instr, param, offset):
+						if isinstance(index, tuple):
+							for j in range(index[0], index[1]):
+								offset = j + 1
+								editor_instr[j] = (opcode, param)
+						else:
+							offset = index + 1
+							editor_instr[index] = (opcode, param)
+
+		res = b""
+
+		for i in range(self.max_script_size):
+			res += encode_u8(comparator_instr[i][0])
+			res += encode_u8(((editor_instr[i][0] >> 1) << 1) | (extr_instr[i] & 0b1))
+			res += encode_u8(comparator_instr[i][1])
+			res += encode_u8(editor_instr[i][1])
+
+		return res
+
+	def __parse_comparator_instr(self, i, instr, param, offset):
+		match = FrameDetector._regex_comp_instr.match(instr)
+
+		if match == None:
+			raise ValueError("line {0}: Unknown comparator instruction: '{1}'".format(i, instr))
+
+		if instr == "nop":
+			return [ ((offset, offset + self.__parse_instr_len(i, offset, param)), 0, 0) ]
+		elif instr == "eof":
+			return [ (offset, 0b11110010, 0) ]
+		else:
+			if len(param) == 0:
+				raise ValueError("line {0}: Instruction requires a parameter: '{1}'".format(i, instr))
+
+			base = match[1]
+			op_size = match[2]
+			endianness = match[3]
+
+			instr_param = b""
+			instr_size = 0
+			opcode = 0
+
+			if len(endianness) == 0:
+				# Big-endian
+				opcode |= 0b100
+
+			if base[0] == "s":
+				# Signed
+				opcode |= 0b1000
+				base = base[1:]
+
+			try:
+				if op_size == "f":
+					instr_size = 4
+					instr_param = encode_float(float(param))
+				elif op_size == "d":
+					instr_size = 8
+					instr_param = encode_double(float(param))
+				else:
+					instr_size = int(op_size) // 8
+					instr_param = int(param, 0).to_bytes(instr_size, byteorder="little", signed=bool(opcode & 0b1000))
+			except ValueError:
+				raise ValueError("line {0}: Invalid parameter for instruction: '{1}'".format(i, param))
+
+			opcode |= ["eq", "gt", "lt", "gtq", "ltq", "or", "and"].index(base) << 4
+
+			if offset + instr_size > self.max_script_size:
+				raise ValueError("line {0}: Instruction ends beyond the script size limit: {1} + {2} > {3}".format(i, offset, instr_size, self.max_script_size))
+
+			if opcode & 0b100:
+				instr_param = instr_param[::-1]
+
+			res = []
+
+			for b in instr_param[:-1]:
+				res.append( (offset, opcode | 0b01, b) )
+				offset += 1
+
+			res.append( (offset, opcode | 0b11, instr_param[-1]) )
+			return res
+
+	def __parse_extractor_instr(self, i, instr, param, offset):
+		if instr not in ["nop", "ext"]:
+			raise ValueError("line {0}: Unknown extractor instruction: '{1}'".format(i, instr))
+
+		return (offset, offset + self.__parse_instr_len(i, offset, param), int(instr == "ext"))
+
+	def __parse_editor_instr(self, i, instr, param, offset):
+		match = FrameDetector._regex_edit_instr.match(instr)
+
+		if match == None:
+			raise ValueError("line {0}: Unknown editor instruction: '{1}'".format(i, instr))
+
+		if instr == "nop" or instr == "setr":
+			return [ ((offset, offset + self.__parse_instr_len(i, offset, param)), FrameDetector._edit_opcode_dict[instr], 0) ]
+		elif instr == "drop" or instr == "corrupt":
+			return [ (offset, FrameDetector._edit_opcode_dict[instr], 0) ]
+		else:
+			if len(param) == 0:
+				raise ValueError("line {0}: Instruction requires a parameter: '{1}'".format(i, instr))
+
+			base = match[1]
+			op_size = match[2]
+			endianness = match[3]
+
+			instr_param = b""
+			instr_size = 0
+			opcode = FrameDetector._edit_opcode_dict[base]
+
+			if len(endianness) == 0 and base != "set":
+				# Big-endian
+				opcode |= 0b10000
+
+			if base == "add" and param[0] == "-":
+				# Set as signed, allows using negative parameters
+				opcode |= 0b100000
+
+			try:
+				if op_size == "f":
+					instr_size = 4
+					instr_param = encode_float(float(param))
+				elif op_size == "d":
+					instr_size = 8
+					instr_param = encode_double(float(param))
+				else:
+					instr_size = int(op_size) // 8
+					instr_param = int(param, 0).to_bytes(instr_size, byteorder="little", signed=bool(opcode & 0b100000))
+			except ValueError:
+				raise ValueError("line {0}: Invalid parameter for instruction: '{1}'".format(i, param))
+
+			if offset + instr_size > self.max_script_size:
+				raise ValueError("line {0}: Instruction ends beyond the script size limit: {1} + {2} > {3}".format(i, offset, instr_size, self.max_script_size))
+
+			opcode |= int(math.log2(instr_size)) << 6
+
+			if opcode & 0b10000:
+				instr_param = instr_param[::-1]
+
+			res = []
+
+			for b in instr_param[:-1]:
+				res.append( (offset, 0, b) )
+				offset += 1
+
+			res.append( (offset, opcode, instr_param[-1]) )
+			return res
+
+	def __parse_instr_len(self, i, offset, param):
+		if len(param):
+			try:
+				param = int(param, 0)
+			except ValueError:
+				raise ValueError("line {0}: Invalid parameter, must be an integer: '{1}'".format(i, param))
+		else:
+			param = 1
+
+		if param < 0:
+			raise ValueError("line {0}: Invalid parameter, must be a positive integer: {1}".format(i, param))
+
+		if offset + param > self.max_script_size:
+			raise ValueError("line {0}: Instruction ends beyond the script size limit: {1} + {2} > {3}".format(i, offset, param, self.max_script_size))
+
+		return param