Huffman编码

文章发布时间:

最后更新时间:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
import numpy as np

import copy



# 节点类

class Node:

    def __init__(self,name=None,value=None):

        self._name=name

        self._value=value

        self.l_child = []    # 子节点列表

        self.code=b""

    # 添加子节点

    def add_child(self,node):

        self.l_child.append(node)



# You can modify the code as what you want

class huffman_coding_in_byte:

    def __init__(self, text):

        self.text = text

        self.bits = 256

        # you can store your encode and decode dict here

        self.encode_dict = dict()

        self.decode_dict = dict()

        self.words=self.get_prob()[0]

        self.probs=self.get_prob()[1]

        self.L=len(self.words)

        self.k=0

        while self.k*(self.bits-1)+1 < self.L:

            self.k+=1

        self.first=self.bits-(self.k*(self.bits-1)+1 - self.L) #第一次合并的项数

        self.nodes=[Node(self.words[i],self.probs[i]) for i in range(self.L)]

    # you can use this to find all the unique source with sorted prob.

    def get_prob(self):

        unique = np.array(list(set(self.text)))

        prob = np.array([self.text.count(u)/len(self.text) for u in unique])

        sort_idx = np.argsort(prob)[::-1]

        return list(unique[sort_idx]), list(prob[sort_idx])

    def select_sort_nodes(self):

        if len(self.nodes) == 1:

            return

        else:

            # 选择概率最小的几个节点合并成新节点

            if self.first != 0:

                new_node=Node(

                    name=self.nodes[-self.first]._name+self.nodes[-1]._name,

                    value=sum([self.nodes[i]._value for i in range(-self.first,0)])

                )                

                for i in range(-self.first,0):

                    new_node.add_child(self.nodes[i])

                for i in range(0,self.first):

                    self.nodes.pop()        

                self.first = 0  

            else:

                new_node=Node(

                    name=self.nodes[-self.bits]._name+self.nodes[-1]._name,

                    value=sum([self.nodes[i]._value for i in range(-self.bits,0)])

                )

                for i in range(-self.bits,0):

                    new_node.add_child(self.nodes[i])

                for i in range(0,self.bits):

                    self.nodes.pop()



            self.nodes.append(new_node)

            # 节点排序

            values = []

            for i in range(len(self.nodes)):

                values.append(self.nodes[i]._value)

            nodes = []

            for i in range(len(self.nodes)):

                nodes.append(self.nodes[i]._name)

            idx = np.argsort(values)[::-1]

            n=[]

            for name in np.array(nodes)[idx]:

                for i in range(len(self.nodes)):

                    if self.nodes[i]._name ==  name:

                        n.append(self.nodes[i])

            self.nodes=copy.deepcopy(n) #这个没必要好像==================================

            self.select_sort_nodes()

            return



    def generate_encode_dict(self,current_node):

        for i in range(0,len(current_node.l_child)):

            if current_node.l_child[i].l_child == []: # 如果没有子节点就给一个编码            

                current_node.l_child[i].code = current_node.code+i.to_bytes(1,"big")

                self.encode_dict[current_node.l_child[i]._name]=current_node.l_child[i].code

                # with open("编码对照文件.txt","a+",encoding="utf-8") as f:

                #     f.write(current_node.l_child[i]._name+":")

                #     f.write(str(current_node.l_child[i]._value))

                #     f.write(str(current_node.l_child[i].code)+"\n")

            else:

                current_node.l_child[i].code = i.to_bytes(1,"big")

                self.generate_encode_dict(current_node.l_child[i])

        return



    def encode(self):

        #encode the text with huffman coding

        self.select_sort_nodes()

        self.generate_encode_dict(self.nodes[0]) #传入根节点

        print("编码字典做好了,编码的字符个数为:",len(self.encode_dict))

        encoded_text=b""

        # 编码

        for w in self.text:

            encoded_text += self.encode_dict[w]

        return encoded_text

    def decode(self, encoded_text):

        # decode the encoded text

        decoded_text = ""

        # 解码的字典

        self.decode_dict={v:k for k,v in self.encode_dict.items()}

        # 解码

        code = b""

        for w in encoded_text:

            code = code + w.to_bytes(1,"big")

            #print(code) #前缀码

            if code in self.decode_dict:

                decoded_text+= self.decode_dict[code]

                code = b""

        return decoded_text




## test 1

# decoded text should be the same as the original text

with open('孤星.txt', 'r', encoding="utf-8") as f:

    file_content = f.read()



my_huffman_coding = huffman_coding_in_byte(file_content)

encoded_text = my_huffman_coding.encode()

#print(encoded_text)

print(my_huffman_coding.encode_dict) # 打印编码字典

print(my_huffman_coding.decode(encoded_text))  # 打印解码结果



## Test 2

utf_encoded_text = file_content.encode("utf-8")

print("压缩前:",len(utf_encoded_text))

print("压缩后",len(encoded_text))
1
2
压缩前: 429992
压缩后 204812