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
|
# -*- coding: utf-8 -*-
#
# File: portato/eix.py
# This file is part of the Portato-Project, a graphical portage-frontend.
#
# Copyright (C) 2006-2009 René 'Necoro' Neumann
# This is free software. You may redistribute copies of it under the terms of
# the GNU General Public License version 2.
# There is NO WARRANTY, to the extent permitted by law.
#
# Written by René 'Necoro' Neumann <necoro@necoro.net>
from __future__ import absolute_import, with_statement
import os
import struct
from functools import wraps
from .helper import debug
class EixError (Exception):
message = "Unknown error."
def __str__ (self):
return self.message
class EndOfFileException (EixError):
def __init__ (self, filename):
self.message = "End of file reached though it was not expected: '%s'" % filename
class EixReaderClosed (EixError):
message = "EixReader is already closed."
class UnsupportedVersionError (EixError):
def __init__ (self, version):
self.message = "Version '%s' is not supported." % version
class EixReader (object):
supported_versions = (28, )
def __init__ (self, filename):
self.filename = filename
self.file = open(filename, "r")
self.closed = 0
try:
self.version = self.number()
if self.version not in self.supported_versions:
raise UnsupportedVersionError(self.version)
debug("Started EixReader for version %s.", self.version)
except:
self.close()
raise
def check_closed (f):
@wraps(f)
def wrapper (self, *args, **kwargs):
if self.closed:
raise EixReaderClosed
return f(self, *args, **kwargs)
return wrapper
@check_closed
def number (self):
n = self._get_bytes(1)
if n < 0xFF:
value = n
else:
count = 0
while (n == 0xFF):
count += 1
n = self._get_bytes(1)
if n == 0:
n = 0xFF # 0xFF is encoded as 0xFF 0x00
count -= 1
value = n << (count*8)
if count > 0:
rest = self._get_bytes(count, expect_list = True)
for i, r in enumerate(rest):
value += r << ((count - i - 1)*8)
return value
@check_closed
def vector (self, get_type, skip = False):
nelems = self.number()
if skip:
for i in range(nelems):
get_type(skip = True)
else:
return (get_type() for i in range(nelems))
@check_closed
def string (self, skip = False):
nelems = self.number()
if skip:
self.file.seek(nelems, os.SEEK_CUR)
else:
s = self.file.read(nelems)
if len(s) != nelems:
raise EndOfFileException, self.filename
return s
@check_closed
def overlay (self, skip = False):
if skip:
self.file.seek(self.number(), os.SEEK_CUR) # path
self.file.seek(self.number(), os.SEEK_CUR) # label
else:
return (self.string(), self.string())
def _get_bytes (self, length, expect_list = False):
s = self.file.read(length)
if len(s) != length:
raise EndOfFileException, self.filename
if length == 1 and not expect_list:
return ord(s) # is faster than unpack and we have a scalar
else:
return struct.unpack("%sB" % length, s)
@check_closed
def close (self):
if self.closed:
raise EixReaderClosed
self.file.close()
self.closed = 1
debug("EixReader closed.")
|
