Huffman Trees are mainly used for data compression and communication. Characters with high weight (high frequency) sit close to the root, while low-weight characters sit far from it, giving variable-length codes.
# Priority queue priorityqueue.py
class HeapNode:
def __init__(self,value,level):
self.value = value
self.level = level
class PriorityQueue:
def __init__(self):
self.heapList = [HeapNode(None,None)]
self.size = 0
def buildQueue(self,dic):
print(f'Building priority queue for {dic}......')
i = len(dic) // 2
self.size = len(dic)
for k,v in dic.items():
node = HeapNode(k,v)
self.heapList.append(node)
while (i > 0):
self.adjustDown(i)
i = i - 1
def adjustDown(self,i):
while (i * 2) <= self.size:
mlc = self.maxLevelChild(i)
if self.heapList[i].level < self.heapList[mlc].level:
self.heapList[i],self.heapList[mlc] = self.heapList[mlc],self.heapList[i]
i = mlc
def maxLevelChild(self,i):
if i * 2 + 1 > self.size:
return i * 2
else:
if self.heapList[i*2].level > self.heapList[i*2+1].level:
return i * 2
else:
return i * 2 + 1
def enqueue(self,value,level):
node = HeapNode(value,level)
self.heapList.append(node)
self.size = self.size + 1
self.adjustUp(self.size)
print(f'{value} enqueued, priority: {level}')
def adjustUp(self,i):
while i // 2 > 0:
if self.heapList[i].level > self.heapList[i // 2].level:
self.heapList[i],self.heapList[i // 2] = self.heapList[i // 2],self.heapList[i]
i = i // 2
def dequeue(self):
pop = self.heapList[1]
self.heapList[1] = self.heapList[self.size]
self.size = self.size - 1
self.heapList.pop()
self.adjustDown(1)
print(f'{pop.value} dequeued, priority: {pop.level}')
if __name__ == '__main__':
pq = PriorityQueue()
pq.buildQueue({'data a':1,'data b':2,'data c':3,'data d':4,'data e':5})
pq.enqueue('data f', 6)
pq.dequeue()
pq.dequeue()
pq.dequeue()
pq.enqueue('data g', 7)
pq.dequeue()
pq.dequeue()
from priorityqueue import PriorityQueue
from collections import defaultdict
class Node:
def __init__(self, character, weight):
self.parent = None
self.leftChild = None
self.rightChild = None
self.character = character
self.weight = weight
self.huffmancode = None
self.isLeaf = False
class Huffman:
def __init__(self, message):
self.message = message
self.pq = PriorityQueue()
self.root = None
self.huffmanEncode = defaultdict(list)
self.huffmanDecode = defaultdict(list)
dic = defaultdict(lambda: 0)
for c in self.message:
dic[c] += 1
for k in dic.keys():
node = Node(k, dic[k])
self.pq.enqueue(node, dic[k])
def createHuffmanTree(self):
while self.pq.size > 0:
if self.pq.size == 1:
root = self.pq.dequeue()
break
nd1 , nd2 = self.pq.dequeue() , self.pq.dequeue()
nd1.huffmancode , nd2.huffmancode = '0' , '1'
if not nd1.leftChild:
nd1.isLeaf = True
if not nd2.rightChild:
nd2.isLeaf = True
root = Node(None, nd1.weight + nd2.weight)
root.leftChild , root.rightChild = nd1 , nd2
nd1.parent , nd2.parent = root , root
self.pq.enqueue(root, root.weight)
self.dfs(root)
def dfs(self, node):
if node:
self.dfs(node.leftChild)
self.dfs(node.rightChild)
if node.isLeaf:
code = ''
temp = node
while temp.parent:
code = ''.join([code, temp.huffmancode])
temp = temp.parent
self.huffmanEncode.setdefault(node.character,code[::-1])
self.huffmanDecode.setdefault(code[::-1],node.character)
def encode(self):
return ''.join([self.huffmanEncode[i] for i in self.message])
def decode(self, ecs):
decode = ''
temp = ''
for i in ecs:
temp = ''.join([temp, i])
if temp in self.huffmanDecode.keys():
decode = ''.join([decode, self.huffmanDecode[temp]])
temp = ''
return decode
def compress(self,ec):
result = ''
start = 0
i = 16
while i <= len(ec):
result = ' '.join([result, str(int(ec[i - 16:i], 2))])
start = i
i = i + 16
if len(ec) % 16 != 0:
result = ' '.join([result, '123456789', ec[start:]])
return result
def decompression(self,ectext):
result = ''
ecList = ectext.split()
for i in ecList:
ec = int(i)
if ec == 123456789:
result = ''.join([result, ecList[-1]])
break
else:
result = ''.join([result, bin(ec).replace('0b', '').rjust(16, '0')])
result = self.decode(result)
return result
if __name__ == '__main__':
print('Building Huffman Tree...')
with open('source.txt', 'r') as f:
hfm = Huffman(f.read())
hfm.createHuffmanTree()
print('Huffman Tree built!')
print('Compressing...')
ec = hfm.encode()
result = hfm.compress(ec)
with open('compressed', 'w') as f:
f.write(result.strip())
print('Compression done!')
print('Decompressing...')
with open('compressed', 'r') as f:
codetext = f.read()
result2 = hfm.decompression(codetext)
with open('decompressed.txt', 'w') as f:
f.write(result2)
print('Decompression done!')The effect is obvious when a file contains many repeated characters. Because Python is slow, I compress a file of around 1 MB as a demonstration; compressing a large file this way would be painfully slow. Building the Huffman tree is fast, though, and this is not a real problem — it can be rewritten in C with the same method.