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
|
import os
from collections import namedtuple
from struct import unpack
from typing import IO
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
import pytzdata
from pytzdata.exceptions import TimezoneNotFound
from pendulum.utils._compat import PY2
from .exceptions import InvalidTimezone
from .exceptions import InvalidZoneinfoFile
from .posix_timezone import PosixTimezone
from .posix_timezone import posix_spec
from .timezone import Timezone
from .transition import Transition
from .transition_type import TransitionType
_offset = namedtuple("offset", "utc_total_offset is_dst abbr_idx")
header = namedtuple(
"header",
"version " "utclocals " "stdwalls " "leaps " "transitions " "types " "abbr_size",
)
class Reader:
"""
Reads compiled zoneinfo TZif (\0, 2 or 3) files.
"""
def __init__(self, extend=True): # type: (bool) -> None
self._extend = extend
def read_for(self, timezone): # type: (str) -> Timezone
"""
Read the zoneinfo structure for a given timezone name.
:param timezone: The timezone.
"""
try:
file_path = pytzdata.tz_path(timezone)
except TimezoneNotFound:
raise InvalidTimezone(timezone)
return self.read(file_path)
def read(self, file_path): # type: (str) -> Timezone
"""
Read a zoneinfo structure from the given path.
:param file_path: The path of a zoneinfo file.
"""
if not os.path.exists(file_path):
raise InvalidZoneinfoFile("The tzinfo file does not exist")
with open(file_path, "rb") as fd:
return self._parse(fd)
def _check_read(self, fd, nbytes): # type: (...) -> bytes
"""
Reads the given number of bytes from the given file
and checks that the correct number of bytes could be read.
"""
result = fd.read(nbytes)
if (not result and nbytes > 0) or len(result) != nbytes:
raise InvalidZoneinfoFile(
"Expected {} bytes reading {}, "
"but got {}".format(nbytes, fd.name, len(result) if result else 0)
)
if PY2:
return bytearray(result)
return result
def _parse(self, fd): # type: (...) -> Timezone
"""
Parse a zoneinfo file.
"""
hdr = self._parse_header(fd)
if hdr.version in (2, 3):
# We're skipping the entire v1 file since
# at least the same data will be found in TZFile 2.
fd.seek(
hdr.transitions * 5
+ hdr.types * 6
+ hdr.abbr_size
+ hdr.leaps * 4
+ hdr.stdwalls
+ hdr.utclocals,
1,
)
# Parse the second header
hdr = self._parse_header(fd)
if hdr.version != 2 and hdr.version != 3:
raise InvalidZoneinfoFile(
"Header versions mismatch for file {}".format(fd.name)
)
# Parse the v2 data
trans = self._parse_trans_64(fd, hdr.transitions)
type_idx = self._parse_type_idx(fd, hdr.transitions)
types = self._parse_types(fd, hdr.types)
abbrs = self._parse_abbrs(fd, hdr.abbr_size, types)
fd.seek(hdr.leaps * 8 + hdr.stdwalls + hdr.utclocals, 1)
trule = self._parse_posix_tz(fd)
else:
# TZFile v1
trans = self._parse_trans_32(fd, hdr.transitions)
type_idx = self._parse_type_idx(fd, hdr.transitions)
types = self._parse_types(fd, hdr.types)
abbrs = self._parse_abbrs(fd, hdr.abbr_size, types)
trule = None
types = [
TransitionType(off, is_dst, abbrs[abbr]) for off, is_dst, abbr in types
]
transitions = []
previous = None
for trans, idx in zip(trans, type_idx):
transition = Transition(trans, types[idx], previous)
transitions.append(transition)
previous = transition
if not transitions:
transitions.append(Transition(0, types[0], None))
return Timezone(transitions, posix_rule=trule, extended=self._extend)
def _parse_header(self, fd): # type: (...) -> header
buff = self._check_read(fd, 44)
if buff[:4] != b"TZif":
raise InvalidZoneinfoFile(
'The file "{}" has an invalid header.'.format(fd.name)
)
version = {0x00: 1, 0x32: 2, 0x33: 3}.get(buff[4])
if version is None:
raise InvalidZoneinfoFile(
'The file "{}" has an invalid version.'.format(fd.name)
)
hdr = header(version, *unpack(">6l", buff[20:44]))
return hdr
def _parse_trans_64(self, fd, n): # type: (IO[Any], int) -> List[int]
trans = []
for _ in range(n):
buff = self._check_read(fd, 8)
trans.append(unpack(">q", buff)[0])
return trans
def _parse_trans_32(self, fd, n): # type: (IO[Any], int) -> List[int]
trans = []
for _ in range(n):
buff = self._check_read(fd, 4)
trans.append(unpack(">i", buff)[0])
return trans
def _parse_type_idx(self, fd, n): # type: (IO[Any], int) -> List[int]
buff = self._check_read(fd, n)
return list(unpack("{}B".format(n), buff))
def _parse_types(
self, fd, n
): # type: (IO[Any], int) -> List[Tuple[Any, bool, int]]
types = []
for _ in range(n):
buff = self._check_read(fd, 6)
offset = unpack(">l", buff[:4])[0]
is_dst = buff[4] == 1
types.append((offset, is_dst, buff[5]))
return types
def _parse_abbrs(
self, fd, n, types
): # type: (IO[Any], int, List[Tuple[Any, bool, int]]) -> Dict[int, str]
abbrs = {}
buff = self._check_read(fd, n)
for offset, is_dst, idx in types:
if idx not in abbrs:
abbr = buff[idx : buff.find(b"\0", idx)].decode("utf-8")
abbrs[idx] = abbr
return abbrs
def _parse_posix_tz(self, fd): # type: (...) -> Optional[PosixTimezone]
s = fd.read().decode("utf-8")
if not s.startswith("\n") or not s.endswith("\n"):
raise InvalidZoneinfoFile('Invalid posix rule in file "{}"'.format(fd.name))
s = s.strip()
if not s:
return
return posix_spec(s)
|
