Skip to content

Commit

Permalink
feat(plc4py): Finish unsigned integer writes for ByteBuffer
Browse files Browse the repository at this point in the history
  • Loading branch information
hutcheb committed May 12, 2023
1 parent 9c8431c commit 5ae793e
Show file tree
Hide file tree
Showing 2 changed files with 112 additions and 48 deletions.
68 changes: 44 additions & 24 deletions sandbox/plc4py/plc4py/spi/generation/WriteBuffer.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,7 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.

from ctypes import *
from ctypes import c_bool, c_byte, c_ubyte, c_uint16, c_uint32, c_uint64, c_int16, c_int32, c_int64, c_float, c_double
from dataclasses import dataclass
from typing import List

Expand Down Expand Up @@ -127,12 +126,12 @@ def write_byte(self, value: c_byte, logical_name: str = "", **kwargs) -> None:
self.write_signed_byte(value, logical_name, **kwargs)

def write_byte_array(self, value: List[c_byte], logical_name: str = "", **kwargs) -> None:
for aByte in value:
self.write_signed_byte(aByte, logical_name, **kwargs)
for a_byte in value:
self.write_signed_byte(a_byte, logical_name, **kwargs)

def write_unsigned_byte(self, value: c_byte, bit_length: int = 8, logical_name: str = "", **kwargs) -> None:
if bit_length <= 0:
raise SerializationException("unsigned byte can only contain max 8 bits")
raise SerializationException("unsigned byte must contain at least 1 bit")
elif bit_length > 8:
raise SerializationException("unsigned byte can only contain max 8 bits")
else:
Expand All @@ -143,25 +142,46 @@ def write_unsigned_byte(self, value: c_byte, bit_length: int = 8, logical_name:

def write_unsigned_short(self, value: c_uint16, bit_length: int = 16, logical_name: str = "", **kwargs) -> None:
if bit_length <= 0:
raise SerializationException("unsigned short can only contain max 8 bits")
raise SerializationException("unsigned short must contain at least 1 bit")
elif bit_length > 16:
raise SerializationException("unsigned short can only contain max 16 bits")
else:
value_encoding: str = kwargs.get("encoding", "default")
if value_encoding == "ASCII":
if bit_length % 8 != 0:
raise SerializationException("'ASCII' encoded fields must have a length that is a multiple of 8 bits long")
char_len: int = int(bit_length / 8)
max_value: int = int(10**char_len - 1)
if value.value > max_value:
raise SerializationException("Provided value of " + str(value) + " exceeds the max value of " + str(max_value))
string_value: str = "{}".format(value.value)
src = bitarray(endian=ByteOrder.get_short_name(self.byte_order))
src.frombytes(bytearray(string_value, value_encoding))
self.bb[self.position:bit_length] = src[:bit_length]
self.position += bit_length
elif value_encoding == "default":
src = bitarray(endian=ByteOrder.get_short_name(self.byte_order))
src.frombytes(value)
self.bb[self.position:bit_length] = src[:bit_length]
self.position += bit_length
self._handle_integer_encoding(c_uint16(value.value), bit_length, **kwargs)

def write_unsigned_int(self, value: c_uint32, bit_length: int = 32, logical_name: str = "", **kwargs) -> None:
if bit_length <= 0:
raise SerializationException("unsigned int must contain at least 1 bit")
elif bit_length > 32:
raise SerializationException("unsigned int can only contain max 32 bits")
else:
self._handle_integer_encoding(c_uint32(value.value), bit_length, **kwargs)

def write_unsigned_long(self, value: c_uint64, bit_length: int = 64, logical_name: str = "", **kwargs) -> None:
if bit_length <= 0:
raise SerializationException("unsigned long must contain at least 1 bit")
elif bit_length > 64:
raise SerializationException("unsigned long can only contain max 16 bits")
else:
self._handle_integer_encoding(c_uint64(value.value), bit_length, **kwargs)

def _handle_integer_encoding(self, value, bit_length: int, **kwargs):
value_encoding: str = kwargs.get("encoding", "default")
if value_encoding == "ASCII":
if bit_length % 8 != 0:
raise SerializationException(
"'ASCII' encoded fields must have a length that is a multiple of 8 bits long")
char_len: int = int(bit_length / 8)
max_value: int = int(10 ** char_len - 1)
if value.value > max_value:
raise SerializationException(
"Provided value of " + str(value) + " exceeds the max value of " + str(max_value))
string_value: str = "{}".format(value.value)
src = bitarray(endian=ByteOrder.get_short_name(self.byte_order))
src.frombytes(bytearray(string_value, value_encoding))
self.bb[self.position:bit_length] = src[:bit_length]
self.position += bit_length
elif value_encoding == "default":
src = bitarray(endian=ByteOrder.get_short_name(self.byte_order))
src.frombytes(value)
self.bb[self.position:bit_length] = src[:bit_length]
self.position += bit_length
92 changes: 68 additions & 24 deletions sandbox/plc4py/tests/unit/plc4py/spi/test_write_buffer.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
# specific language governing permissions and limitations
# under the License.
#
from ctypes import c_bool, c_byte, c_uint16
from ctypes import c_bool, c_byte, c_uint16, c_uint64

import pytest
from bitarray import bitarray
Expand All @@ -40,94 +40,94 @@ def test_write_buffer_get_pos(mocker) -> None:

def test_write_buffer_set_bit_little_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.LITTLE_ENDIAN)
wb.write_bit(c_bool(True), "Test String")
wb.write_bit(c_bool(True))
ba: memoryview = wb.get_bytes()
assert(b'\x01' == ba.tobytes())


def test_write_buffer_set_bit_big_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.BIG_ENDIAN)
wb.write_bit(c_bool(True), "Test String")
wb.write_bit(c_bool(True))
ba: memoryview = wb.get_bytes()
assert(b'\x80' == ba.tobytes())


def test_write_buffer_set_bit_get_pos(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.LITTLE_ENDIAN)
wb.write_bit(c_bool(True), "Test String")
wb.write_bit(c_bool(True))
pos: int = wb.get_pos()
assert(1 == pos)


def test_write_buffer_set_bit_x2_little_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.LITTLE_ENDIAN)
wb.write_bit(c_bool(True), "Test String 1")
wb.write_bit(c_bool(True), "Test String 2")
wb.write_bit(c_bool(True))
wb.write_bit(c_bool(True))
ba: memoryview = wb.get_bytes()
assert (b'\x03' == ba.tobytes())


def test_write_buffer_set_bit_x2_big_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.BIG_ENDIAN)
wb.write_bit(c_bool(True), "Test String 1")
wb.write_bit(c_bool(True), "Test String 2")
wb.write_bit(c_bool(True))
wb.write_bit(c_bool(True))
ba: memoryview = wb.get_bytes()
assert (b'\xc0' == ba.tobytes())


def test_write_buffer_set_bit_get_pos_dual(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.LITTLE_ENDIAN)
wb.write_bit(c_bool(True), "Test String 1")
wb.write_bit(c_bool(True), "Test String 2")
wb.write_bit(c_bool(True))
wb.write_bit(c_bool(True))
pos: int = wb.get_pos()
assert(pos == 2)


def test_write_buffer_set_unsigned_byte_get_pos(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.LITTLE_ENDIAN)
wb.write_unsigned_byte(c_byte(0xFF), 8, "Test String 1")
wb.write_unsigned_byte(c_byte(0xFF), 8)
pos: int = wb.get_pos()
assert(pos == 8)


def test_write_buffer_set_unsigned_byte_little_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.LITTLE_ENDIAN)
wb.write_unsigned_byte(c_byte(0x12), 8, "Test String 1")
wb.write_unsigned_byte(c_byte(0x12), 8)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("01001000"))


def test_write_buffer_set_unsigned_byte_big_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.BIG_ENDIAN)
wb.write_unsigned_byte(c_byte(0x12), 8, "Test String 1")
wb.write_unsigned_byte(c_byte(0x12), 8)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00010010"))


def test_write_buffer_set_unsigned_byte_little_endian_niblet(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.LITTLE_ENDIAN)
wb.write_unsigned_byte(c_byte(0x12), 4, "Test String 1")
wb.write_unsigned_byte(c_byte(0x12), 4)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("01000000"))


def test_write_buffer_set_unsigned_byte_big_endian_niblet(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(1, ByteOrder.BIG_ENDIAN)
wb.write_unsigned_byte(c_byte(0x12), 4, "Test String 1")
wb.write_unsigned_byte(c_byte(0x12), 4)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00010000"))


def test_write_buffer_write_unsigned_short_little_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.LITTLE_ENDIAN)
wb.write_unsigned_short(c_uint16(0x12), 16, "Test String 1")
wb.write_unsigned_short(c_uint16(0x12), 16)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("01001000 00000000", endian="little"))


def test_write_buffer_write_unsigned_short_big_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.BIG_ENDIAN)
wb.write_unsigned_short(c_uint16(0x12), 16, "Test String 1")
wb.write_unsigned_short(c_uint16(0x12), 16)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00010010 00000000", endian="big"))

Expand All @@ -142,24 +142,24 @@ def test_write_buffer_write_unsigned_short_little_endian_dual(mocker) -> None:

def test_write_buffer_write_unsigned_short_big_endian_dual(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.BIG_ENDIAN)
wb.write_unsigned_short(c_uint16(0x12), 16, "Test String 1")
wb.write_unsigned_short(c_uint16(0x34), 16, "Test String 1")
wb.write_unsigned_short(c_uint16(0x12), 16)
wb.write_unsigned_short(c_uint16(0x34), 16)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00010010 00000000 00110100 00000000", endian="big"))


def test_write_buffer_write_unsigned_short_big_endian_full(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.BIG_ENDIAN)
wb.write_unsigned_short(c_uint16(-1), 16, "Test String 1")
wb.write_unsigned_short(c_uint16(-1), 16)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("11111111 11111111", endian="bit"))
assert (ba.obj == bitarray("11111111 11111111", endian="big"))


def test_write_buffer_write_unsigned_short_bit_big_endian_full(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.BIG_ENDIAN)
wb.write_bit(c_bool(True), "Test String 1")
wb.write_bit(c_bool(False), "Test String 1")
wb.write_unsigned_short(c_uint16(-1), 16, "Test String 1")
wb.write_bit(c_bool(True))
wb.write_bit(c_bool(False))
wb.write_unsigned_short(c_uint16(-1), 16)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("10 11111111 11111111", endian="big"))

Expand All @@ -176,3 +176,47 @@ def test_write_buffer_write_unsigned_short_ascii_encoding_big_endian(mocker) ->
wb.write_unsigned_short(c_uint16(1), 16, "ASCII Value of 1 - 0x31", encoding="ASCII")
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00110001", endian="big"))


def test_write_buffer_write_unsigned_int_little_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.LITTLE_ENDIAN)
# Testing incompatible format being sent to it.
wb.write_unsigned_int(c_uint16(0x12), 32)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("01001000 00000000 00000000 00000000", endian="little"))


def test_write_buffer_write_unsigned_int_big_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.BIG_ENDIAN)
# Testing incompatible format being sent to it.
wb.write_unsigned_int(c_uint16(0x12), 32)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00010010 00000000 00000000 00000000", endian="big"))


def test_write_buffer_write_unsigned_long_little_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.LITTLE_ENDIAN)
wb.write_unsigned_long(c_uint64(0x12), 64)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("01001000 00000000 00000000 00000000 00000000 00000000 00000000 00000000", endian="little"))


def test_write_buffer_write_unsigned_long_big_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(2, ByteOrder.BIG_ENDIAN)
wb.write_unsigned_long(c_uint64(0x12), 64)
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00010010 00000000 00000000 00000000 00000000 00000000 00000000 00000000", endian="big"))


def test_write_buffer_write_unsigned_long_ascii_encoding_little_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(8, ByteOrder.LITTLE_ENDIAN)
wb.write_unsigned_long(c_uint64(11111111), 64, "ASCII Value of 1111 1111 - 0x3131313131313131", encoding="ASCII")
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("10001100 10001100 10001100 10001100 10001100 10001100 10001100 10001100", endian="little"))


def test_write_buffer_write_unsigned_short_ascii_encoding_big_endian(mocker) -> None:
wb: WriteBufferByteBased = WriteBufferByteBased(8, ByteOrder.BIG_ENDIAN)
wb.write_unsigned_long(c_uint64(11111111), 64, "ASCII Value of 1111 1111 - 0x3131313131313131", encoding="ASCII")
ba: memoryview = wb.get_bytes()
assert (ba.obj == bitarray("00110001 00110001 00110001 00110001 00110001 00110001 00110001 00110001", endian="big"))

0 comments on commit 5ae793e

Please sign in to comment.