-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathglomo.py
194 lines (168 loc) · 11.9 KB
/
glomo.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
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
import logging
from collections import defaultdict, namedtuple
from dataclasses import dataclass
from urllib.error import URLError
from pathlib import Path
from typing import List, Iterable, Dict, Optional, Tuple, DefaultDict
from utils.files import read_from_web_or_disk, Pathlike
from utils.highwinds_cdn import CdnServer
@dataclass
class ComponentBlock:
component_name: str
cdn_subdomain: CdnServer
package_path: str
manifest_versions: List[int]
require_auth: bool
sim_version_low: int
sim_version_high: int
def __str__(self):
return f'COMPONENT com.laminarresearch.xplane_10.{self.component_name}\n' \
f'{self.cdn_subdomain.for_component_list()}\n' \
f'{self.package_path}\n' \
f'MANIFEST_VERSIONS {self.manifest_versions[0]} {self.manifest_versions[1]}\n' \
f'REQUIRE_AUTH {int(self.require_auth)}\n' \
f'SIM_VERSIONS {self.sim_version_low}-{self.sim_version_high}\n\n'
@staticmethod
def from_str(file_text: str):
lines = file_text.strip().splitlines(keepends=False)
assert len(lines) == 6, 'Expected exactly: SERVER, CDN subdomain, package path, MANIFEST_VERSION, REQUIRE_AUTH, SIM_VERSIONS lines'
sim_versions_str = lines[5].split('SIM_VERSIONS ', maxsplit=1)[1].split()
package_path = lines[2]
assert package_path.startswith('/'), 'Your package path is not absolute on the server---this will go poorly for you!'
return ComponentBlock(component_name=lines[0].split('com.laminarresearch.xplane_10.', maxsplit=1)[1],
cdn_subdomain=CdnServer.from_component_list_id(lines[1]),
package_path=package_path,
manifest_versions=[int(v) for v in lines[3].split('MANIFEST_VERSIONS ', maxsplit=1)[1].strip().split()],
require_auth=bool(lines[4].split('REQUIRE_AUTH', maxsplit=1)[1].strip()),
sim_version_low=int(sim_versions_str[0]),
sim_version_high=int(sim_versions_str[1]))
def parse_component_list(path_to_component_list_txt: Pathlike='https://lookup.x-plane.com/_lookup_mobile_/component_list.txt') -> List[ComponentBlock]:
try:
component_list = read_from_web_or_disk(path_to_component_list_txt)
except URLError as e:
iphone_repo_root = Path(__file__).parent.parent.parent
fallback_path = iphone_repo_root / 'resources/common_ios/config/component_list.txt'
logging.warning(f'Failed to read component list from the web; falling back to local copy at {fallback_path}')
component_list = read_from_web_or_disk(fallback_path)
component_token = 'COMPONENT '
skipped_header = component_token + component_list.split(component_token, maxsplit=1)[1]
nuked_end = skipped_header.split('ENDOFLIST')[0]
return [ComponentBlock.from_str(component_token + block) for block in nuked_end.split(component_token) if block]
def component_list_version(path_to_component_list_txt: Pathlike='https://lookup.x-plane.com/_lookup_mobile_/component_list.txt') -> int:
next_line_is_version = False
for line in read_from_web_or_disk(path_to_component_list_txt).splitlines():
if line.rstrip() == 'COMPONENTS':
assert not next_line_is_version
next_line_is_version = True
elif next_line_is_version:
return int(line)
raise RuntimeError('Failed to parse component_list.txt version')
def component_versions(components: Iterable[ComponentBlock]) -> Dict[str, List[int]]:
return {component.component_name: component.manifest_versions
for component in components}
ManifestHistory = namedtuple('ManifestHistory', ['version']) # the most *recent* manifest version which touched this file for a modification or delete
@dataclass
class ManifestEntry:
hash: str
in_zip: Optional[Path]=None # None if this is a RAWFILE, or the path of the ZIP this is contained in
@dataclass(frozen=True)
class ComponentManifest:
"""Represents the directory.txt manifest for a component, with both the hashes of current files and the file history"""
version: int
install_path_prefix: Path
entries: DefaultDict[Path, List[ManifestEntry]] # A given path on disk may be in many places in the manifest (included in an arbitrary number of ZIPs)
history: Dict[Path, ManifestHistory]
zips: Dict[Path, str] # associates ZIP paths with their hash
def all_paths_all_entries(self) -> List[Tuple[Path, ManifestEntry]]:
"""Flattens the entries dict to give you an iterable of all paths on disk and all their corresponding manifest entries"""
return [(Path(path), entry)
for path, locations in self.entries.items()
for entry in locations]
# I'm sorry to whomever needs to maintain this...including you future-Chris...but for some reason I found it easier
# to use REGEX to parse this manifest file than normal string utilities and so....here we are. I will say, 99% of the
# matching is REGEX 101 with two exceptions:
# 1) Matching floating point values = ([+-]?(?:[0-9]*[.])?[0-9]+)
# 2) Matching space-escaped strings like our paths which may have Earth\ Nav\ Data for example. We don't want to see those spaces as whitespace and so we have = ((?:[^\\\s]|\\.)+)
# Feel free to insult and scathe me for it but I'm at least making SOME attempt to document this so...I'm not a complete asshole.
@classmethod
def from_file(cls, manifest_file_path_or_url: Optional[Pathlike]) -> Optional['ComponentManifest']:
import re
prev_manifest_version: Optional[int] = None
install_path_prefix: Optional[Path] = None
entries = defaultdict(list)
history = dict()
zips = dict()
if manifest_file_path_or_url:
most_recent_zip: Optional[Path] = None
for line in read_from_web_or_disk(manifest_file_path_or_url).splitlines():
# Look at each line for the version number. Until we find it, we don't care about anything else!
if prev_manifest_version is None:
match_obj = re.match(r'^MANIFEST_VERSION\s+([0-9]+)\s*$', str(line))
if match_obj:
prev_manifest_version = int(match_obj.group(1))
continue
else:
line.strip()
# A NOTE ABOUT 'install_path_prefix':
#
# Chris says: We don't want the install path prepended before any of the externally visible paths! It is only relevant on the CLIENT device when it's installed. The server paths have no
# relation to the install_path despite the fact that it may SEEM as though they are in lock-step (because they often are). As I write this, we have a new and an old 737 on the server in different
# paths. One will only work for an old install of the sim and the other will only work for a new install, but they both have the same install_path_prefix, component name etc. This is
# normal operation and it's why the component list allows for different paths between components with the same name. It is the COMPONENT's path that needs to be prepended...not the install_path_prefix...
# but that happens outside of this class when components do stuff with this manifest data.
#
# Now look at each line and see if it's an install path prefix
if not install_path_prefix and line.startswith("INSTALL_PATH_PREFIX"):
match_obj = re.match(r'^INSTALL_PATH_PREFIX\s+(.*)', str(line))
if match_obj:
install_path_prefix = cls.unescape_spaces(str(match_obj.group(1)))
continue
elif line.startswith("ZIP") or line.startswith("ZIPFILE") or line.startswith("RAWFILE"):
# Check for RAWFILE line
match_obj = re.match(r'^RAWFILE\s+([-+]?\d+)\s+([-+]?\d+)\s+([-+]?\d+)\s+([-+]?\d+)\s+(\S+)\s+([+-]?(?:[0-9]*[.])?[0-9]+)\s+(\S+)\s+((?:[^\\\s]|\\.)+)\s+((?:[^\\\s]|\\.)+)', str(line))
if match_obj and len(match_obj.groups()) == 9:
path = cls.unescape_spaces(match_obj.group(8))
#path = cls.unescape_spaces(install_path_prefix / match_obj.group(8))
assert all(entry.in_zip for entry in entries[path]), f'Duplicated raw file {path}'
entries[path].append(ManifestEntry(hash=match_obj.group(7)))
continue
# Check for ZIP line
match_obj = re.match(r'^ZIP\s+([+-]?(?:[0-9]*[.])?[0-9]+)\s+(\S+)\s+((?:[^\\\s]|\\.)+)\s+((?:[^\\\s]|\\.)+)', str(line))
if match_obj and len(match_obj.groups()) == 4:
most_recent_zip = cls.unescape_spaces(match_obj.group(3))
#most_recent_zip = Path(install_path_prefix) / cls.unescape_spaces(match_obj.group(3))
zips[most_recent_zip] = match_obj.group(2)
continue
# Check for ZIPFILE line
match_obj = re.match(r'^ZIPFILE\s+([-+]?\d+)\s+([-+]?\d+)\s+([-+]?\d+)\s+([-+]?\d+)\s+(\S+)\s+([+-]?(?:[0-9]*[.])?[0-9]+)\s+(\S+)\s+(.+)', str(line))
if match_obj and len(match_obj.groups()) == 8:
assert most_recent_zip, 'This ZIPFILE does not seem to be contained in a ZIP...?'
path = cls.unescape_spaces(match_obj.group(8))
#path = cls.unescape_spaces(install_path_prefix / match_obj.group(8))
assert not any(mfst_entry.in_zip == most_recent_zip for mfst_entry in entries[path]), f'File {path} must be unique within the ZIP {most_recent_zip}'
entries[path].append(ManifestEntry(hash=match_obj.group(7), in_zip=most_recent_zip))
continue
raise RuntimeError("We either found a mal-formed manifest line, or this parser has a bug. The line was:\n" + line)
elif line.startswith("FILE_HISTORY"):
match_obj = re.match(r'^FILE_HISTORY\s+([0-9]+)\s+(.+)', str(line))
if match_obj and len(match_obj.groups()) == 2:
path = Path(match_obj.group(2))
assert path not in history, f'Founnd duplicate history entry for {path}\nHistory entry should only be the most *recent* manifest version which touched this file for a modification or delete.'
history[path] = ManifestHistory(int(match_obj.group(1)))
continue
assert prev_manifest_version is not None, 'Manifest was missing a version'
assert install_path_prefix is not None, 'Manifest was missing an install path prefix'
assert entries, 'No entries for manifest... this will not be a very useful component!'
for disk_path, manifest_entries in entries.items():
assert len(set(entry.hash for entry in manifest_entries)) <= 1, f'The same file ({disk_path}) wound up with different hashes... Huh?'
return cls(prev_manifest_version, Path(install_path_prefix), entries, history, zips)
else: # no file given
return None
@classmethod
def from_file_with_next_version(cls, manifest_file_path_or_url: Optional[Pathlike]) -> Tuple[Optional['ComponentManifest'], int]:
out_manifest = cls.from_file(manifest_file_path_or_url)
next_version = out_manifest.version + 1 if out_manifest else 1
return out_manifest, next_version
@staticmethod
def unescape_spaces(pathlike: Pathlike) -> Path:
return Path(str(pathlike).replace("\\ ", " "))