PyAE

The main module that encodes and decodes a text message.

Author: ahmedfgad

example.py

@@ -0,0 +1,29 @@
+import pyae
+# from decimal import getcontext
+
+# Example for encoding a simple text message using the PyAE module.
+
+frequency_table = {"a": 2,
+                   "b": 7,
+                   "c": 1}
+
+AE = pyae.ArithmeticEncoding(frequency_table)
+
+probability_table = AE.probability_table
+
+# Default precision is 28. Change it to do arithmetic operations with larger/smaller numbers.
+# getcontext().prec = 28
+
+original_msg = "abc"
+print("Original Message: {msg}".format(msg=original_msg))
+
+encoder, encoded_msg = AE.ae_encoder(msg=original_msg, 
+                                     probability_table=probability_table)
+print("Encoded Message: {msg}".format(msg=encoded_msg))
+
+decoder, decoded_msg = AE.ae_decoder(encoded_msg=encoded_msg, 
+                                     msg_length=len(original_msg),
+                                     probability_table=probability_table)
+print("Decoded Message: {msg}".format(msg=decoded_msg))
+
+print("Message Decoded Successfully? {result}".format(result=original_msg == decoded_msg))

pyae.py

@@ -0,0 +1,109 @@
+from decimal import Decimal
+
+class ArithmeticEncoding:
+    """
+    ArithmeticEncoding is a class for building the arithmetic encoding.
+    """
+
+    def __init__(self, frequency_table):
+
+        frequency_table = frequency_table
+        self.probability_table = self.get_probability_table(frequency_table)
+
+        self.probability_table
+
+    def get_probability_table(self, frequency_table):
+        """
+        Calculates the probability table out of the frequency table.
+        """
+        total_frequency = sum(list(frequency_table.values()))
+
+        probability_table = {}
+        for key, value in frequency_table.items():
+            probability_table[key] = value/total_frequency
+
+        return probability_table
+
+    def get_encoded_value(self, encoder):
+        """
+        After encoding the entire message, this method returns the single value that represents the entire message.
+        """
+        last_stage = list(encoder[-1].values())
+        last_stage_values = []
+        for sublist in last_stage:
+            for element in sublist:
+                last_stage_values.append(element)
+
+        last_stage_min = min(last_stage_values)
+        last_stage_max = max(last_stage_values)
+
+        return (last_stage_min + last_stage_max)/2
+
+    def process_stage(self, probability_table, stage_min, stage_max):
+        """
+        Processing a stage in the encoding/decoding process.
+        """
+        stage_probs = {}
+        stage_domain = stage_max - stage_min
+        for term_idx in range(len(probability_table.items())):
+            term = list(probability_table.keys())[term_idx]
+            term_prob = Decimal(probability_table[term])
+            cum_prob = term_prob * stage_domain + stage_min
+            stage_probs[term] = [stage_min, cum_prob]
+            stage_min = cum_prob
+        return stage_probs
+
+    def ae_encoder(self, msg, probability_table):
+        """
+        Encodes a message.
+        """
+
+        encoder = []
+    
+        stage_min = Decimal(0.0)
+        stage_max = Decimal(1.0)
+
+        for msg_term_idx in range(len(msg)):
+            stage_probs = self.process_stage(probability_table, stage_min, stage_max)
+
+            msg_term = msg[msg_term_idx]
+            stage_min = stage_probs[msg_term][0]
+            stage_max = stage_probs[msg_term][1]
+
+            encoder.append(stage_probs)
+
+        stage_probs = self.process_stage(probability_table, stage_min, stage_max)
+        encoder.append(stage_probs)
+
+        encoded_msg = self.get_encoded_value(encoder)
+
+        return encoder, encoded_msg
+
+    def ae_decoder(self, encoded_msg, msg_length, probability_table):
+        """
+        Decodes a message.
+        """
+
+        decoder = []
+        decoded_msg = ""
+
+        stage_min = Decimal(0.0)
+        stage_max = Decimal(1.0)
+
+        for idx in range(msg_length):
+            stage_probs = self.process_stage(probability_table, stage_min, stage_max)
+
+            for msg_term, value in stage_probs.items():
+                if encoded_msg >= value[0] and encoded_msg <= value[1]:
+                    break
+
+            decoded_msg = decoded_msg + msg_term
+            stage_min = stage_probs[msg_term][0]
+            stage_max = stage_probs[msg_term][1]
+
+            decoder.append(stage_probs)
+
+        stage_probs = self.process_stage(probability_table, stage_min, stage_max)
+        decoder.append(stage_probs)
+
+        return decoder, decoded_msg