-
Notifications
You must be signed in to change notification settings - Fork 1
/
optimize_upip.py
106 lines (84 loc) · 2.46 KB
/
optimize_upip.py
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
#
# This script optimizes a Python source distribution tarball as produced by
# "python3 setup.py sdist" command for MicroPython's native package manager,
# upip. Optimization includes:
# * Removing metadata files not used by upip (this includes setup.py)
# * Recompressing gzip archive with 4K dictionary size so it can be
# installed even on low-heap targets.
#
import sys
import os
import zlib
from subprocess import Popen, PIPE
import glob
import tarfile
import re
import io
def gzip_4k(inf, fname):
comp = zlib.compressobj(level=9, wbits=16 + 12)
with open(fname + ".out", "wb") as outf:
while 1:
data = inf.read(1024)
if not data:
break
outf.write(comp.compress(data))
outf.write(comp.flush())
os.rename(fname, fname + ".orig")
os.rename(fname + ".out", fname)
def recompress(fname):
with Popen(["gzip", "-d", "-c", fname], stdout=PIPE).stdout as inf:
gzip_4k(inf, fname)
def find_latest(dir):
res = []
for fname in glob.glob(dir + "/*.gz"):
st = os.stat(fname)
res.append((st.st_mtime, fname))
res.sort()
latest = res[-1][1]
return latest
def recompress_latest(dir):
latest = find_latest(dir)
print(latest)
recompress(latest)
EXCLUDE = [r".+/setup.py"]
INCLUDE = [r".+\.py", r".+\.egg-info/(PKG-INFO|requires\.txt)"]
outbuf = io.BytesIO()
def filter_tar(name):
fin = tarfile.open(name, "r:gz")
fout = tarfile.open(fileobj=outbuf, mode="w")
for info in fin:
# print(info)
include = None
for p in EXCLUDE:
if re.match(p, info.name):
include = False
break
if include is None:
for p in INCLUDE:
if re.match(p, info.name):
include = True
print("Including:", info.name)
if not include:
continue
farch = fin.extractfile(info)
fout.addfile(info, farch)
fout.close()
fin.close()
from setuptools import Command
class OptimizeUpip(Command):
user_options = []
def run(self):
latest = find_latest("dist")
filter_tar(latest)
outbuf.seek(0)
gzip_4k(outbuf, latest)
def initialize_options(self):
pass
def finalize_options(self):
pass
# For testing only
if __name__ == "__main__":
# recompress_latest(sys.argv[1])
filter_tar(sys.argv[1])
outbuf.seek(0)
gzip_4k(outbuf, sys.argv[1])