utils.py

   1 import os
   2 import math
   3 from collections import Counter
   4 import heapq
   5 import numpy as np
   6 import struct
   7
   8
   9 class BitMapFile:
  10     def __init__(self, text):
  11         self._text = text.split()
  12         self.magic_number = self._text[0]
  13         self.image_width = self._text[1]
  14         self.image_height = self._text[2]
  15         self.image_array = self._text[3:]
  16         self.zeros = 0
  17         self.ones = 0
  18         self.size = int(self.image_width) * int(self.image_height)
  19
  20         for _ in range(len(self.image_array)):
  21             if self.image_array[_] == "0":
  22                 self.zeros += 1
  23             elif self.image_array[_] == "1":
  24                 self.ones += 1
  25             else:
  26                 print("ignore")
  27
  28
  29 """
  30 To read files
  31 """
  32 def get_file(file_path):
  33     with open(file_path, 'r', encoding='utf-8') as file:
  34         _file_content = file.read()
  35     return _file_content
  36
  37 """
  38 To evaluate comprimento médio do código - l(c)
  39     l(all_pixels) = ceil(log2P(all_pixels))+1
  40 """
  41 def arithmetic_cod_preview(_pbm):
  42     print("\n***  Arithmetic Preview  ***")
  43     pixels = [int(pixel) for pixel in _pbm.image_array]
  44     _P0 = _pbm.zeros/_pbm.size
  45     _P1 = _pbm.ones/_pbm.size
  46     _PT01_int = 1
  47
  48     for pixel in pixels:
  49         if pixel == 0:
  50             _PT01_int *= _P0
  51         elif pixel == 1:
  52             _PT01_int *= _P1
  53
  54     _arithmetic_encode_lenght = np.ceil(-1*np.log2(_PT01_int)) + 1
  55     print(f"L(image) --> {_arithmetic_encode_lenght} bits")
  56     _entropy = calcular_entropia(_pbm.image_array)
  57     print(f"entropy --> {_entropy} bits")
  58     _datastream = clean_image_data(_pbm.image_array)
  59     _entropia_conditional = calcular_entropia_condicional(_datastream)
  60     print(f"entropy condicional --> {_entropia_conditional} bits")
  61     return int(_arithmetic_encode_lenght)
  62
  63 """
  64 Arithmetic Encoding approach
  65
  66 it uses static probabilities provided on output file
  67
  68 c(x)=|interval_low_val +(interval_high_val).F(x)_low
  69     =|interval_low_val +(interval_high_val).F(x)_high
  70 _to_encode_val = avg(C(all)_low,C(all)_high)
  71 _output = bin(_to_encode_val))
  72 """
  73 def arithmetic_encode_image(_pbm):
  74     pixels = [int(pixel) for pixel in _pbm.image_array]
  75     _P0 =  np.float64(_pbm.zeros/_pbm.size)
  76     _P1 =  np.float64(_pbm.ones/_pbm.size)
  77     _interval = [0,1]
  78     # probabilidades acomuladas orderdanas por ordem crescente
  79     _Fx = {1: (0,_P1),
  80            0: (_P1,1)}
  81     _output_opt_1 = f"{_pbm.zeros} {_pbm.ones} {_pbm.image_width} {_pbm.image_height} "
  82
  83     for pixel in pixels:
  84         if pixel == 0:
  85             _low_val = np.float64(_interval[0] + (_interval[1]-_interval[0]) * _Fx.get(0)[0])
  86             _high_val = np.float64(_interval[0] + (_interval[1]-_interval[0])*_Fx.get(0)[1])
  87         if pixel == 1:
  88             _low_val = np.float64(_interval[0] + (_interval[1] - _interval[0]) * _Fx.get(1)[0])
  89             _high_val = np.float64(_interval[0] + (_interval[1] - _interval[0]) * _Fx.get(1)[1])
  90         _interval[0] = _low_val
  91         _interval[1] = _high_val
  92
  93     _to_encode_value = np.float64((_interval[0]+_interval[1])/2)
  94     print(f"to encode --> {_to_encode_value}")
  95     _output_opt_2 = _output_opt_1 + f"{_to_encode_value}"
  96
  97     _size = arithmetic_cod_preview(_pbm)
  98     _res = _to_encode_value
  99     for _ in range(_size):
 100         _res *=2
 101         if _res < 1:
 102             _output_opt_1 += "0"
 103         else:
 104             _output_opt_1 += "1"
 105             _res = _res-1
 106
 107
 108     return _output_opt_1,_output_opt_2
 109
 110 """
 111 To decode encoded pbm files with arithmetic_encode_image()
 112     amount_of_zeros amount_of_ones image_width image_height encoded_image
 113 """
 114 def arithmetic_decode_file(_txt):
 115     _encoded_pbm_file = get_file(_txt).split()
 116
 117     amount_of_zeros = int(_encoded_pbm_file[0])
 118     amount_of_ones = int(_encoded_pbm_file[1])
 119     image_width = int(_encoded_pbm_file[2])
 120     image_height = int(_encoded_pbm_file[3])
 121     encoded_image = np.float64(_encoded_pbm_file[4])
 122
 123     _output = f"P1\n{image_width} {image_height}\n"
 124     _size = int(image_width) * int(image_height)
 125     _P0 = np.float64(amount_of_zeros/_size)
 126     _P1 = np.float64(amount_of_ones/_size)
 127     _Fx = {1:(0,_P1),
 128            0:(_P1,1)}
 129
 130     _constructed_image = ""
 131     for h in range(image_height):
 132         for w in range(image_width):
 133             if  _P1 < encoded_image < 1:
 134                 _constructed_image += "0 "
 135                 encoded_image = (encoded_image-_P1)/_P0
 136             elif 0 < encoded_image < _P1:
 137                 _constructed_image += "1 "
 138                 encoded_image = (encoded_image - 0) / _P1
 139
 140             if w == image_width-1:
 141                 _constructed_image += "\n"
 142
 143     _output += _constructed_image
 144     return _output
 145
 146
 147 def calcular_entropia(bitstream):
 148     n = len(bitstream)
 149     if n == 0: return 0
 150
 151     p0 = bitstream.count('0') / n
 152     p1 = bitstream.count('1') / n
 153
 154     entropia = 0
 155     for p in [p0, p1]:
 156         if p > 0:
 157             entropia -= p * math.log2(p)
 158     return entropia
 159
 160
 161 def calcular_entropia_condicional(bitstream):
 162     transicoes = {'00': 0, '01': 0, '10': 0, '11': 0}
 163     contagem_base = {'0': 0, '1': 0}
 164
 165     for i in range(len(bitstream) - 1):
 166         par = bitstream[i:i + 2]
 167         transicoes[par] += 1
 168         contagem_base[bitstream[i]] += 1
 169
 170     h_condicional = 0
 171     for base in ['0', '1']:
 172         p_base = contagem_base[base] / (len(bitstream) - 1)
 173         if p_base > 0:
 174             h_local = 0
 175             for prox in ['0', '1']:
 176                 p_transicao = transicoes[base + prox] / contagem_base[base]
 177                 if p_transicao > 0:
 178                     h_local -= p_transicao * math.log2(p_transicao)
 179             h_condicional += p_base * h_local
 180
 181     return h_condicional
 182
 183
 184 def clean_image_data(image_data):
 185     clean_text = ""
 186
 187     for line in image_data:
 188         line = line.strip()
 189         line = line.replace(" ", "")
 190         if not line or line.startswith('#'):
 191             continue
 192         clean_text += line
 193
 194     return clean_text
 195
 196
 197 def calcular_taxa_compressao(caminho_original, caminho_comprimido, total_pixeis):
 198     tamanho_orig = os.path.getsize(caminho_original)
 199     tamanho_comp = os.path.getsize(caminho_comprimido)
 200
 201     poupanca = (1 - (tamanho_comp / tamanho_orig)) * 100
 202
 203     bpp = (tamanho_comp * 8) / total_pixeis
 204
 205     return poupanca, bpp, tamanho_orig, tamanho_comp
 206
 207
 208 def xor(bitstream, largura, altura):
 209     matriz = []
 210     for i in range(altura):
 211         linha = [int(b) for b in bitstream[i * largura: (i + 1) * largura]]
 212         matriz.append(linha)
 213
 214     nova_imagem = ""
 215     # A primeira linha mantém-se igual (não tem linha anterior)
 216     nova_imagem += "".join(map(str, matriz[0]))
 217
 218     for i in range(1, altura):
 219         nova_linha = []
 220         for j in range(largura):
 221             res = matriz[i][j] ^ matriz[i - 1][j]
 222             nova_linha.append(str(res))
 223         nova_imagem += "".join(nova_linha)
 224
 225     return nova_imagem
 226
 227
 228 def descodificar_xor(bitstream_transformado, largura, altura):
 229     matriz_temp = []
 230     for i in range(altura):
 231         linha = [int(b) for b in bitstream_transformado[i * largura: (i + 1) * largura]]
 232         matriz_temp.append(linha)
 233
 234     matriz_original = []
 235
 236     matriz_original.append(matriz_temp[0])
 237
 238     for i in range(1, altura):
 239         linha_recuperada = []
 240         for j in range(largura):
 241             pixel_original = matriz_temp[i][j] ^ matriz_original[i - 1][j]
 242             linha_recuperada.append(pixel_original)
 243         matriz_original.append(linha_recuperada)
 244
 245     bitstream_final = ""
 246     for linha in matriz_original:
 247         bitstream_final += "".join(map(str, linha))
 248
 249     return bitstream_final
 250
 251
 252 # -------------------------------------------------------------------------
 253 # HUFFMAN IMPLEMENTATION
 254 # -------------------------------------------------------------------------
 255
 256 class HuffmanNode:
 257     def __init__(self, char, freq):
 258         self.char = char  # 0 ou 1 que está no pixel
 259         self.freq = freq  # Quantas vezes aparece
 260         self.left = None  # Filho à esquerda
 261         self.right = None  # Filho à direita
 262
 263     # Nó com menor frequencia
 264     def __lt__(self, other):
 265         return self.freq < other.freq
 266
 267
 268 # Constroi a árvore
 269 def build_huffman_tree(pixels):
 270     frequency = Counter(pixels)
 271     heap = [HuffmanNode(char, freq) for char, freq in frequency.items()]
 272     heapq.heapify(heap)
 273
 274     # Cria a árvore de baixo para cima / frequencia menor para maior
 275     while len(heap) > 1:
 276         node1 = heapq.heappop(heap)
 277         node2 = heapq.heappop(heap)
 278         merged = HuffmanNode(None, node1.freq + node2.freq)
 279         merged.left = node1
 280         merged.right = node2
 281         heapq.heappush(heap, merged)
 282
 283     return heap[0] if heap else None
 284
 285
 286 # Cria os códigos (0 para a esquerda, 1 para a direira)
 287 def make_codes(node, current_code="", codes=None):
 288     if codes is None:
 289         codes = {}
 290     if node is None:
 291         return codes
 292     if node.char is not None:
 293         codes[node.char] = current_code
 294         return codes
 295
 296     make_codes(node.left, current_code + "0", codes)
 297     make_codes(node.right, current_code + "1", codes)
 298     return codes
 299
 300
 301 # Faz encoding do ficheiro
 302 def huffman_encode_image(_pbm):
 303     print("\n*** Huffman Encoding ***")
 304     pixels = _pbm.image_array
 305
 306     # Erro caso não encontre a imagem
 307     if not pixels:
 308         print("Empty image.")
 309         return None
 310
 311     # Constroi a árvore e os códigos
 312     root = build_huffman_tree(pixels)
 313     codes = make_codes(root)
 314
 315     print(f"Codes: {codes}")
 316
 317     # Faz encode dos pixeis
 318     encoded_str = "".join([codes[p] for p in pixels])
 319
 320     # Garante que o total é multiplo de 8
 321     extra_padding = 8 - len(encoded_str) % 8
 322     encoded_str = encoded_str + "0" * extra_padding
 323
 324     # Reduz o tamanho convertendo de byte para bit
 325     b = bytearray()
 326     for i in range(0, len(encoded_str), 8):
 327         byte = encoded_str[i:i + 8]
 328         b.append(int(byte, 2))
 329
 330     # Cabeçalho
 331     width = int(_pbm.image_width)
 332     height = int(_pbm.image_height)
 333     freq0 = _pbm.zeros
 334     freq1 = _pbm.ones
 335     header = struct.pack('>IIIIB', width, height, freq0, freq1, extra_padding)
 336
 337     return header + b
 338
 339
 340 # Faz decode do ficheiro
 341 def huffman_decode_file(filename):
 342     print(f"\n*** Huffman Decoding from {filename} ***")
 343
 344     # Lê o ficheiro
 345     with open(filename, 'rb') as f:
 346         file_content = f.read()
 347
 348     # Separa o cabeçalho
 349     header_size = struct.calcsize('>IIIIB')
 350     width, height, freq0, freq1, extra_padding = struct.unpack('>IIIIB', file_content[:header_size])
 351     encoded_data = file_content[header_size:]
 352
 353     print(f"Header Info -> W:{width}, H:{height}, Zeros:{freq0}, Ones:{freq1}")
 354
 355     # Constroi a árvore
 356     fake_pixels = ['0'] * freq0 + ['1'] * freq1
 357     root = build_huffman_tree(fake_pixels)
 358     codes = make_codes(root)
 359
 360     # Faz decode dos pixeis
 361     reverse_codes = {v: k for k, v in codes.items()}
 362
 363     # Converte os bytes em string
 364     bit_string = ""
 365     for byte in encoded_data:
 366         bit_string += f"{byte:08b}"
 367
 368     # Desencripta
 369     decoded_pixels = []
 370     current_code = ""
 371     for bit in bit_string:
 372         current_code += bit
 373         if current_code in reverse_codes:
 374             decoded_pixels.append(reverse_codes[current_code])
 375             current_code = ""
 376
 377     # Limita o tamanho width * height
 378     expected_pixels = width * height
 379     decoded_pixels = decoded_pixels[:expected_pixels]
 380
 381     pbm_content = f"P1\n{width} {height}\n"
 382
 383     # Formata as linhas para não ficar sequencial
 384     row_str = ""
 385     for i, p in enumerate(decoded_pixels):
 386         row_str += p + " "
 387         if (i + 1) % width == 0:
 388             pbm_content += row_str.strip() + "\n"
 389             row_str = ""
 390
 391     print("Decoding finished.")
 392     return pbm_content