-
Notifications
You must be signed in to change notification settings - Fork 618
/
check-protobuf-version-compatibility.py
134 lines (113 loc) 路 4.53 KB
/
check-protobuf-version-compatibility.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
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
import pathlib
import platform
import re
import subprocess
import sys
from typing import List, Tuple
from pkg_resources import parse_version # noqa: F401
def get_available_protobuf_versions() -> List[str]:
"""Get a list of available protobuf versions."""
try:
output = subprocess.check_output(
["pip", "index", "versions", "protobuf"],
).decode("utf-8")
versions = list({o for o in output.split() if o[0].isnumeric()})
versions = [v if not v.endswith(",") else v[:-1] for v in versions]
return sorted(versions)
except subprocess.CalledProcessError:
return []
def parse_protobuf_requirements() -> List[Tuple[str, str]]:
"""Parse protobuf requirements from a requirements.txt file."""
path_requirements = pathlib.Path(__file__).parent.parent / "requirements.txt"
with open(path_requirements) as f:
requirements = f.readlines()
system_python_version = f"{sys.version_info.major}.{sys.version_info.minor}"
system_platform = sys.platform
system_machine = platform.machine()
protobuf_reqs = []
for line in requirements:
if line.startswith("protobuf"):
version_reqs = line.strip().split(";")[0]
# first, check the system requirements
system_reqs = line.strip().split(";")[1]
# regex to find quoted python version in system_reqs
python_version = re.search(
r"python_version\s+([<>=!]+)\s+[',\"]([2,3]*[.][0-9]+)[',\"]",
system_reqs,
)
if python_version is not None:
version_check = (
f"parse_version('{system_python_version}') "
f"{python_version.group(1)} "
f"parse_version('{python_version.group(2)}')"
)
if not eval(version_check):
continue
# regex to find quoted platform in system_reqs
platform_reqs = re.search(
r"sys_platform\s+([<>=!]+)\s+[',\"]([a-z]+)[',\"]",
system_reqs,
)
if platform_reqs is not None:
if not eval(
f"'{system_platform}' {platform_reqs.group(1)} '{platform_reqs.group(2)}'"
):
continue
# regex to find platform machine in system_reqs
platform_machine = re.search(
r"platform[.]machine\s+([<>=!]+)\s+[',\"]([a-z]+)[',\"]",
system_reqs,
)
if platform_machine is not None:
if not eval(
f"'{system_machine}' {platform_machine.group(1)} '{platform_machine.group(2)}'"
):
continue
# finally, parse the protobuf version requirements
reqs = version_reqs.split("protobuf")[1].split(",")
print(reqs)
for req in reqs:
for i, char in enumerate(req):
if char.isnumeric():
protobuf_reqs.append(
(
req[:i].strip(),
req[i:].strip(),
)
)
break
print(protobuf_reqs)
return protobuf_reqs
def get_matching_versions(
available_protobuf_vs: List[str], protobuf_reqs: List[Tuple[str, str]]
) -> List[str]:
matching_vs = []
for v in available_protobuf_vs:
if all(
eval(f"parse_version('{v}') {rq[0]} parse_version('{rq[1]}')")
for rq in protobuf_reqs
):
matching_vs.append(v)
return sorted(list(set(matching_vs)))
def attempt_install_protobuf_version(version: str) -> bool:
try:
subprocess.check_call(["pip", "install", f"protobuf=={version}"])
subprocess.check_call(["python", "-c", "import wandb"])
return True
except subprocess.CalledProcessError:
return False
if __name__ == "__main__":
available_protobuf_versions = get_available_protobuf_versions()
protobuf_requirements = parse_protobuf_requirements()
matching_versions = get_matching_versions(
available_protobuf_versions,
protobuf_requirements,
)
version_compatibility = {
version: attempt_install_protobuf_version(version)
for version in matching_versions
}
for version, compatible in version_compatibility.items():
print(f"protobuf=={version}: {compatible}")
if not all(version_compatibility.values()):
sys.exit(1)