From f37f2775b8456a8c3b70597d0aacf34a6860f3f9 Mon Sep 17 00:00:00 2001 From: Ben Hutcheson Date: Sat, 6 Jan 2024 21:08:10 +0100 Subject: [PATCH] fix(plc4py): Add abstract decorators to read buffer --- .../plc4py/spi/generation/ReadBuffer.py | 65 +++++++++---------- .../plc4py/spi/generation/WriteBuffer.py | 4 +- sandbox/plc4py/plc4py/utils/GenericTypes.py | 6 +- 3 files changed, 36 insertions(+), 39 deletions(-) diff --git a/sandbox/plc4py/plc4py/spi/generation/ReadBuffer.py b/sandbox/plc4py/plc4py/spi/generation/ReadBuffer.py index c781d755ca8..c981d5e7efd 100644 --- a/sandbox/plc4py/plc4py/spi/generation/ReadBuffer.py +++ b/sandbox/plc4py/plc4py/spi/generation/ReadBuffer.py @@ -16,20 +16,7 @@ # under the License. import struct import types -from ctypes import ( - c_byte, - c_ubyte, - c_uint16, - c_uint32, - c_uint64, - c_int16, - c_int32, - c_int64, - c_float, - c_double, - c_int8, - c_uint8, -) +from abc import abstractmethod, ABC from dataclasses import dataclass from typing import List, Union, Any @@ -43,95 +30,113 @@ class PositionAware: + @abstractmethod def get_pos(self) -> int: raise NotImplementedError @dataclass -class ReadBuffer(ByteOrderAware, PositionAware): +class ReadBuffer(ByteOrderAware, PositionAware, ABC): byte_order: ByteOrder + @abstractmethod def get_pos(self) -> int: raise NotImplementedError + @abstractmethod def push_context(self, logical_name: str, **kwargs) -> None: raise NotImplementedError + @abstractmethod def pop_context(self, logical_name: str, **kwargs) -> None: raise NotImplementedError + @abstractmethod def read_bit(self, logical_name: str = "", **kwargs) -> bool: raise NotImplementedError def read_byte(self, logical_name: str = "", **kwargs) -> int: self.read_signed_byte(8, logical_name, **kwargs) + @abstractmethod def read_byte_array( self, number_of_bytes: int, logical_name: str = "", **kwargs ) -> List[int]: raise NotImplementedError + @abstractmethod def read_unsigned_byte( self, bit_length: int = 8, logical_name: str = "", **kwargs ) -> int: raise NotImplementedError + @abstractmethod def read_unsigned_short( self, bit_length: int = 16, logical_name: str = "", **kwargs ) -> int: raise NotImplementedError + @abstractmethod def read_unsigned_int( self, bit_length: int = 32, logical_name: str = "", **kwargs ) -> int: raise NotImplementedError + @abstractmethod def read_unsigned_long( self, bit_length: int = 64, logical_name: str = "", **kwargs ) -> int: raise NotImplementedError + @abstractmethod def read_signed_byte( self, bit_length: int = 8, logical_name: str = "", **kwargs ) -> int: raise NotImplementedError + @abstractmethod def read_short(self, bit_length: int = 16, logical_name: str = "", **kwargs) -> int: raise NotImplementedError + @abstractmethod def read_int(self, bit_length: int = 32, logical_name: str = "", **kwargs) -> int: raise NotImplementedError + @abstractmethod def read_long(self, bit_length: int = 64, logical_name: str = "", **kwargs) -> int: raise NotImplementedError + @abstractmethod def read_float( self, bit_length: int = 32, logical_name: str = "", **kwargs ) -> float: raise NotImplementedError + @abstractmethod def read_double( self, bit_length: int = 64, logical_name: str = "", **kwargs ) -> float: raise NotImplementedError + @abstractmethod def read_str(self, bit_length: int = -1, logical_name: str = "", **kwargs) -> str: raise NotImplementedError - def read_virtual(self, logical_name: str = "", **kwargs) -> str: - raise NotImplementedError - + @abstractmethod def read_complex_array(self, logical_name: str = "", **kwargs) -> List[Any]: raise NotImplementedError + @abstractmethod def read_complex(self, logical_name: str = "", read_function=None, **kwargs) -> Any: raise NotImplementedError + @abstractmethod def read_enum( self, bit_length: int = -1, logical_name: str = "", read_function=None, **kwargs ) -> Any: raise NotImplementedError + @abstractmethod def read_array_field( self, logical_name: str = "", @@ -145,20 +150,6 @@ def read_array_field( class ReadBufferByteBased(ReadBuffer): - NUMERIC_UNION = Union[ - c_ubyte, - c_byte, - c_uint8, - c_uint16, - c_uint32, - c_uint64, - c_int8, - c_int16, - c_int32, - c_int64, - c_float, - c_double, - ] def __init__(self, bb: bytearray, byte_order: ByteOrder): if byte_order == ByteOrder.LITTLE_ENDIAN: @@ -181,6 +172,7 @@ def push_context(self, logical_name: str, **kwargs) -> None: pass def pop_context(self, logical_name: str, **kwargs) -> None: + # Byte buffer doesn't need context handling pass def read_bit(self, logical_name: str = "", **kwargs) -> bool: @@ -350,7 +342,6 @@ def read_double( def read_complex(self, logical_name: str = "", read_function=None, **kwargs) -> Any: if isinstance(read_function, types.FunctionType): return read_function(self, **kwargs) - pass def read_enum( self, bit_length: int = -1, logical_name: str = "", read_function=None, **kwargs @@ -361,6 +352,12 @@ def read_enum( else: raise RuntimeError("read_enum called but read_function wasn't an enum") + def read_complex_array(self, logical_name: str = "", **kwargs) -> List[Any]: + raise NotImplementedError + + def read_str(self, bit_length: int = -1, logical_name: str = "", **kwargs) -> str: + raise NotImplementedError + def read_array_field( self, logical_name: str = "", @@ -372,7 +369,7 @@ def read_array_field( ) -> List[Any]: if count is not None: parsed_array = [] - for i in range(count): + for _ in range(count): parsed_array.append(read_function(self, **kwargs)) return parsed_array else: diff --git a/sandbox/plc4py/plc4py/spi/generation/WriteBuffer.py b/sandbox/plc4py/plc4py/spi/generation/WriteBuffer.py index 0222fd9cda0..9d1c1c78ed2 100644 --- a/sandbox/plc4py/plc4py/spi/generation/WriteBuffer.py +++ b/sandbox/plc4py/plc4py/spi/generation/WriteBuffer.py @@ -130,9 +130,6 @@ def write_str( ) -> None: raise NotImplementedError - def write_virtual(self, value: str, logical_name: str = "", **kwargs) -> None: - raise NotImplementedError - def write_complex_array( self, value: List[PlcMessage], logical_name: str = "", **kwargs ) -> None: @@ -181,6 +178,7 @@ def push_context(self, logical_name: str, **kwargs) -> None: pass def pop_context(self, logical_name: str, **kwargs) -> None: + # Byte Based Buffer doesn't need a context. pass def write_bit(self, value: bool, logical_name: str = "", **kwargs) -> None: diff --git a/sandbox/plc4py/plc4py/utils/GenericTypes.py b/sandbox/plc4py/plc4py/utils/GenericTypes.py index e4f1aa19c20..582ead53c84 100644 --- a/sandbox/plc4py/plc4py/utils/GenericTypes.py +++ b/sandbox/plc4py/plc4py/utils/GenericTypes.py @@ -22,18 +22,20 @@ from typing import Generator -# TODO: Figure out what the parameters are and if we need this class GenericGenerator(Generator): def __enter__(self): return self - def send(self, _value, blah): + def send(self, _value): + # TODO I need to figure out why this was added pass def throw(self): + # TODO I need to figure out why this was added pass def __exit__(self, *args): + # TODO I need to figure out why this was added pass