Address code review

EosLib/format/__init__.py

@@ -1,4 +1,4 @@
-from EosLib.format.formats import telemetry_data, position
+from EosLib.format.formats import telemetry_data, position, empty_format
 
 
 from EosLib.format.definitions import Type as Type

EosLib/format/decode_factory.py

@@ -1,12 +1,27 @@
+from EosLib.format import csv_format
+
+
 class DecodeFactory:
     def __init__(self):
         self._decoders = {}
+        self._csv_decoders = {}
 
     def register_decoder(self, format_class):
         self._decoders[format_class.get_format_type()] = format_class.decode
+        if issubclass(format_class, csv_format.CsvFormat):
+            self._csv_decoders[format_class.get_format_type()] = format_class.decode_from_csv
 
     def decode(self, data_format, data: bytes):
-        return self._decoders[data_format](data)
+        decoder = self._decoders.get(data_format)
+        if decoder is None:
+            raise TypeError("No decoder found for this type")
+        return decoder(data)
+
+    def decode_from_csv(self, data_format, data: str):
+        decoder = self._csv_decoders.get(data_format)
+        if decoder is None:
+            raise TypeError("No decoder found for this type")
+        return decoder(data)
 
 
 decode_factory = DecodeFactory()

EosLib/format/definitions.py

@@ -11,4 +11,5 @@ class Type(IntEnum):
     COMMAND = 5
     RESPONSE = 6
     TELEMETRY_DATA = 7
+    EMPTY = 8
     ERROR = 255

EosLib/format/formats/empty_format.py

@@ -0,0 +1,30 @@
+import struct
+
+from typing_extensions import Self
+
+from EosLib.format.base_format import BaseFormat
+
+from EosLib.format.definitions import Type
+
+
+class EmptyFormat(BaseFormat):
+
+    def __init__(self, num_bytes=0):
+        self.num_bytes = num_bytes
+
+    def __eq__(self, other):
+        return self.num_bytes == other.num_bytes
+
+    @staticmethod
+    def get_format_type() -> Type:
+        return Type.EMPTY
+
+    def encode(self) -> bytes:
+        return struct.pack(self.get_format_string())
+
+    @classmethod
+    def decode(cls, data: bytes) -> Self:
+        return EmptyFormat(len(data))
+
+    def get_format_string(self) -> str:
+        return self.num_bytes * "x"

EosLib/format/formats/position.py

@@ -1,4 +1,6 @@
+import csv
 import datetime
+import io
 import struct
 from enum import IntEnum
 from typing_extensions import Self
@@ -86,17 +88,22 @@ def get_csv_headers(self):
         return ["GPS Timestamp", "Latitude", "Longitude", "Speed", "Altitude", "Number of Satellites", "Flight State"]
 
     def encode_to_csv(self) -> str:
-        return ",".join([self.gps_time.isoformat(),
+        output = io.StringIO()
+        writer = csv.writer(output)
+        writer.writerow([self.gps_time.isoformat(),
                          str(self.latitude),
                          str(self.longitude),
                          str(self.speed),
                          str(self.altitude),
                          str(self.number_of_satellites),
                          str(self.flight_state.value)])
 
+        return output.getvalue()
+
     @classmethod
-    def decode_from_csv(cls, csv: str) -> Self:
-        csv_list = csv.split(",")
+    def decode_from_csv(cls, csv_string: str) -> Self:
+        reader = csv.reader([csv_string])
+        csv_list = list(reader)[0]
 
         return Position(datetime.datetime.fromisoformat(csv_list[0]),
                         float(csv_list[1]),

EosLib/format/formats/telemetry_data.py

@@ -1,3 +1,5 @@
+import csv
+import io
 import struct
 from typing_extensions import Self
 
@@ -88,16 +90,21 @@ def get_csv_headers(self):
         return ["temperature", "pressure", "humidity", "x_rotation", "y_rotation", "z_rotation"]
 
     def encode_to_csv(self) -> str:
-        return ",".join([str(self.temperature),
+        output = io.StringIO()
+        writer = csv.writer(output)
+        writer.writerow([str(self.temperature),
                          str(self.pressure),
                          str(self.humidity),
                          str(self.x_rotation),
                          str(self.y_rotation),
                          str(self.z_rotation)])
 
+        return output.getvalue()
+
     @classmethod
-    def decode_from_csv(cls, csv: str) -> Self:
-        csv_list = csv.split(",")
+    def decode_from_csv(cls, csv_string: str) -> Self:
+        reader = csv.reader([csv_string])
+        csv_list = list(reader)[0]
 
         return TelemetryData(float(csv_list[0]),
                              float(csv_list[1]),

EosLib/packet/data_header.py

@@ -3,8 +3,8 @@
 
 from datetime import datetime
 
-import EosLib.format.definitions
-from EosLib.packet import definitions
+from EosLib.format.definitions import Type
+from EosLib.packet.definitions import Priority
 from EosLib.packet.definitions import HeaderPreamble, old_data_headers
 from EosLib.packet.exceptions import PacketFormatError, DataHeaderFormatError
 from EosLib.device import Device
@@ -22,8 +22,8 @@ class DataHeader:
 
     def __init__(self,
                  sender: Device,
-                 data_type: EosLib.format.definitions.Type = EosLib.format.definitions.Type.NO_TYPE,
-                 priority: definitions.Priority = definitions.Priority.NO_TRANSMIT,
+                 data_type: Type = Type.NO_TYPE,
+                 priority: Priority = Priority.NO_TRANSMIT,
                  destination: Device = Device.NO_DEVICE,
                  generate_time: datetime = None
                  ):

EosLib/packet/packet.py

@@ -1,15 +1,12 @@
-import datetime
 import struct
 
 import EosLib.format.decode_factory
 from EosLib.format.base_format import BaseFormat
-from EosLib.format.definitions import Type
 
 from EosLib.packet.transmit_header import TransmitHeader
 from EosLib.packet.data_header import DataHeader
 from EosLib.packet.definitions import HeaderPreamble, Priority
 from EosLib.packet.exceptions import PacketFormatError
-from EosLib.device import Device
 
 
 class Packet:
@@ -115,22 +112,6 @@ def encode(self) -> bytes:
 
         return packet_bytes
 
-    def encode_to_string(self):
-        """ Takes a packet and encodes it into a comma separated string
-
-        :return: The comma separated string representation of the packet
-        """
-        self.validate_packet()
-
-        # It's easier if we make all the encoded packet string arrays the same length, so we add a fake transmit header
-        if self.transmit_header is None:
-            self.transmit_header = TransmitHeader(0, send_rssi=0)
-
-        return "{transmit_header}, {data_header}, {body}".format(
-            transmit_header=self.transmit_header.encode_to_string(),
-            data_header=self.data_header.encode_to_string(),
-            body=self.body.encode().hex())
-
     @staticmethod
     def decode(packet_bytes: bytes):
         """Takes a bytes object and decodes it into a Packet object.
@@ -162,34 +143,3 @@ def decode(packet_bytes: bytes):
                                 decoded_transmit_header)
 
         return decoded_packet
-
-    @staticmethod
-    def decode_from_string(packet_string: str):
-        """Takes a string and decodes it into a Packet object.
-
-        The format is this: sequence num, send time, rssi, data type, sender, priority, generate time, body
-
-        :param packet_string: The string to be decoded
-        :return: The decoded Packet object
-        """
-
-        packet_array = packet_string.split(', ')
-
-        send_seq_num = int(packet_array[0])
-        send_rssi = int(packet_array[1])
-        send_time = datetime.datetime.fromisoformat(packet_array[2])
-
-        sender = Device(int(packet_array[3]))
-        data_type = Type(int(packet_array[4]))
-        priority = Priority(int(packet_array[5]))
-        destination = Device(int(packet_array[6]))
-        generate_time = datetime.datetime.fromisoformat(packet_array[7])
-
-        decoded_transmit_header = TransmitHeader(send_seq_num, send_time, send_rssi)
-        decoded_data_header = DataHeader(sender, data_type, priority, destination, generate_time)
-        decoded_packet = Packet(EosLib.format.decode_factory.decode_factory.decode(data_type,
-                                                                                   bytes.fromhex(packet_array[8])),
-                                decoded_data_header,
-                                decoded_transmit_header)
-
-        return decoded_packet

example.py

@@ -6,6 +6,7 @@
 import EosLib.packet.packet
 import EosLib.packet.transmit_header
 from EosLib.packet.packet import DataHeader, TransmitHeader, Packet
+from EosLib.format.formats.empty_format import EmptyFormat
 sequence_number = 0
 
 
@@ -18,17 +19,17 @@ def collect_data() -> int:
 def log_data(data):
     data_header = DataHeader(
         EosLib.device.Device.PRESSURE,
-        EosLib.format.Type.DATA,
+        EosLib.format.Type.EMPTY,
         EosLib.Priority.DATA
     )
 
-    body = str(data)
-    body = body.encode()
+    # Please don't store your own data as number of empty bytes, make your own format.
+    body = EmptyFormat(data)
 
     created_packet = Packet(body, data_header)
 
     with open("TestData.dat", 'w') as f:
-        f.write(created_packet.encode_to_string())
+        f.write(created_packet.encode().hex())
 
     return created_packet
 

tests/format/formats/csv_format_test.py

@@ -0,0 +1,14 @@
+from abc import ABC
+
+from format.formats.format_test import CheckFormat
+
+from EosLib.format.decode_factory import decode_factory
+
+
+class CheckCsvFormat(CheckFormat, ABC):
+    def test_encode_decode_csv(self):
+        base_format = self.get_good_format()
+        base_format_csv = base_format.encode_to_csv()
+        new_position = decode_factory.decode_from_csv(self.get_good_format().get_format_type(), base_format_csv)
+
+        assert base_format == new_position

tests/format/formats/format_test.py

@@ -0,0 +1,51 @@
+import abc
+import copy
+import datetime
+
+import EosLib.format.csv_format
+from EosLib.format.decode_factory import decode_factory
+
+
+# Name can't contain the word Test, hence Check
+class CheckFormat(abc.ABC):
+    @abc.abstractmethod
+    def get_format_from_list(self, format_list: []):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def get_good_format_list(self):
+        raise NotImplementedError
+
+    def get_good_format(self):
+        return self.get_format_from_list(self.get_good_format_list())
+
+    def test_is_eq(self):
+        data_1 = self.get_good_format()
+        data_2 = self.get_good_format()
+
+        assert data_1 == data_2
+
+    def test_not_eq(self):
+        test_passed = True
+
+        data_1 = self.get_good_format()
+
+        for i in range(len(self.get_good_format_list())):
+            new_data_list = copy.deepcopy(self.get_good_format_list())
+            if isinstance(new_data_list[i], (int, float)):
+                new_data_list[i] += 1
+            elif isinstance(new_data_list[i], datetime.datetime):
+                new_data_list[i] += datetime.timedelta(1)
+            data_2 = self.get_format_from_list(new_data_list)
+
+            if data_1 == data_2:
+                test_passed = False
+
+        assert test_passed
+
+    def test_encode_decode_bytes(self):
+        base_format = self.get_good_format()
+        base_position_bytes = base_format.encode()
+        new_format = decode_factory.decode(self.get_good_format().get_format_type(), base_position_bytes)
+        assert base_format == new_format
+

tests/format/formats/test_empty_format.py

@@ -0,0 +1,11 @@
+from EosLib.format.formats.empty_format import EmptyFormat
+
+from tests.format.formats.format_test import CheckFormat
+
+
+class TestEmptyFormat(CheckFormat):
+    def get_format_from_list(self, format_list: []):
+        return EmptyFormat()
+
+    def get_good_format_list(self):
+        return []

tests/format/formats/test_position.py

@@ -0,0 +1,47 @@
+import datetime
+
+from EosLib.format.formats.position import Position
+from EosLib.format.formats.position import FlightState
+
+from format.formats.csv_format_test import CheckCsvFormat
+
+good_data_list = [datetime.datetime.now(),
+                  33.7756,
+                  84.3963,
+                  974.5,
+                  70.2,
+                  5,
+                  FlightState.DESCENT]
+
+
+def get_position_from_list(position_list: [float]):
+    return Position(position_list[0],
+                    position_list[1],
+                    position_list[2],
+                    position_list[3],
+                    position_list[4],
+                    position_list[5],
+                    position_list[6])
+
+
+def get_good_position():
+    return get_position_from_list(good_data_list)
+
+
+def test_get_validity_valid():
+    good_position = get_good_position()
+    assert good_position.get_validity()
+
+
+def test_get_validity_bad_satellites():
+    bad_satellites_position = get_good_position()
+    bad_satellites_position.number_of_satellites = 3
+    assert not bad_satellites_position.get_validity()
+
+
+class TestPosition(CheckCsvFormat):
+    def get_format_from_list(self, format_list: []):
+        return get_position_from_list(format_list)
+
+    def get_good_format_list(self):
+        return good_data_list