-
Notifications
You must be signed in to change notification settings - Fork 16
/
geodifflib.py
225 lines (182 loc) · 7.37 KB
/
geodifflib.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
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
225
# -*- coding: utf-8 -*-
'''
pygeodiff.geodifflib
--------------------
This module provides wrapper of geodiff C library
:copyright: (c) 2019 Peter Petrik
:license: MIT, see LICENSE for more details.
'''
import ctypes
import os
import platform
from ctypes.util import find_library
from .__about__ import __version__
class GeoDiffLibError(Exception):
pass
class GeoDiffLibConflictError(GeoDiffLibError):
pass
class GeoDiffLibUnsupportedChangeError(GeoDiffLibError):
pass
class GeoDiffLibVersionError(GeoDiffLibError):
pass
# keep in sync with c-library
SUCCESS=0
ERROR=1
CONFLICT=2
UNSUPPORTED_CHANGE=3
def _parse_return_code(rc, msg):
if rc == SUCCESS:
return
elif rc == ERROR:
raise GeoDiffLibError(msg)
elif rc == CONFLICT:
raise GeoDiffLibConflictError(msg)
elif rc == UNSUPPORTED_CHANGE:
raise GeoDiffLibUnsupportedChangeError(msg)
else:
raise GeoDiffLibVersionError("Internal error (enum " + str(rc) + " not handled)")
class GeoDiffLib:
def __init__(self, name):
if name is None:
self.libname = self.package_libname()
if not os.path.exists(self.libname):
# not found, try system library
self.libname = find_library("geodiff")
else:
self.libname = name
if self.libname is None:
raise GeoDiffLibVersionError("Unable to locate GeoDiff library, tried " + self.package_libname() + " and geodiff on system.")
self.lib = ctypes.CDLL(self.libname, use_errno=True)
self.init()
self.check_version()
def package_libname(self):
# assume that the package is installed through PIP
if platform.system() == 'Windows':
prefix = ""
arch = platform.architecture()[0] # 64bit or 32bit
if "32" in arch:
suffix = "-win32.dll"
else:
suffix = ".dll"
elif platform.system() == 'Darwin':
prefix = "lib"
suffix = ".dylib"
else:
prefix = "lib"
suffix = ".so"
whl_lib = prefix + 'pygeodiff-' + __version__ + '-python' + suffix
dir_path = os.path.dirname(os.path.realpath(__file__))
return os.path.join(dir_path, whl_lib)
def init(self):
func = self.lib.GEODIFF_init
func()
def set_logger_callback(self, callback):
func = self.lib.GEODIFF_setLoggerCallback
CMPFUNC = ctypes.CFUNCTYPE(None, ctypes.c_int, ctypes.c_char_p)
func.argtypes = [CMPFUNC]
# do not remove self, callback needs to be member
self.callbackLogger = CMPFUNC(callback)
func(self.callbackLogger)
def set_maximum_logger_level(self, maxLevel):
func = self.lib.GEODIFF_setMaximumLoggerLevel
func.argtypes = [ctypes.c_int]
func(maxLevel)
def version(self):
func = self.lib.GEODIFF_version
func.restype = ctypes.c_char_p
ver = func()
return ver.decode('utf-8')
def check_version(self):
cversion = self.version()
pyversion = __version__
if cversion != pyversion:
raise GeoDiffLibVersionError("version mismatch (%s C vs %s PY)".format(cversion, pyversion))
def create_changeset(self, base, modified, changeset):
func = self.lib.GEODIFF_createChangeset
func.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = base.encode('utf-8')
b_string2 = modified.encode('utf-8')
b_string3 = changeset.encode('utf-8')
res = func(b_string1, b_string2, b_string3)
_parse_return_code(res, "createChangeset")
def invert_changeset(self, changeset, changeset_inv):
func = self.lib.GEODIFF_invertChangeset
func.argtypes = [ctypes.c_char_p, ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = changeset.encode('utf-8')
b_string2 = changeset_inv.encode('utf-8')
res = func(b_string1, b_string2)
_parse_return_code(res, "invert_changeset")
def create_rebased_changeset(self, base, modified, changeset_their, changeset, conflict):
func = self.lib.GEODIFF_createRebasedChangeset
func.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = base.encode('utf-8')
b_string2 = modified.encode('utf-8')
b_string3 = changeset_their.encode('utf-8')
b_string4 = changeset.encode('utf-8')
b_string5 = conflict.encode('utf-8')
res = func(b_string1, b_string2, b_string3, b_string4, b_string5)
_parse_return_code(res, "createRebasedChangeset")
def rebase(self, base, modified_their, modified, conflict):
func = self.lib.GEODIFF_rebase
func.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = base.encode('utf-8')
b_string2 = modified_their.encode('utf-8')
b_string3 = modified.encode('utf-8')
b_string4 = conflict.encode('utf-8')
res = func(b_string1, b_string2, b_string3, b_string4)
_parse_return_code(res, "rebase")
def apply_changeset(self, base, changeset):
func = self.lib.GEODIFF_applyChangeset
func.argtypes = [ctypes.c_char_p, ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = base.encode('utf-8')
b_string2 = changeset.encode('utf-8')
res = func(b_string1, b_string2)
_parse_return_code(res, "apply_changeset")
def list_changes(self, changeset, result):
func = self.lib.GEODIFF_listChanges
func.argtypes = [ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = changeset.encode('utf-8')
b_string2 = result.encode('utf-8')
res = func(b_string1, b_string2)
_parse_return_code(res, "list_changes")
def list_changes_summary(self, changeset, result):
func = self.lib.GEODIFF_listChangesSummary
func.argtypes = [ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = changeset.encode('utf-8')
b_string2 = result.encode('utf-8')
res = func(b_string1, b_string2)
_parse_return_code(res, "list_changes_summary")
def has_changes(self, changeset):
func = self.lib.GEODIFF_hasChanges
func.argtypes = [ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = changeset.encode('utf-8')
nchanges = func(b_string1)
if nchanges < 0:
raise GeoDiffLibError("has_changes")
return nchanges == 1
def changes_count(self, changeset):
func = self.lib.GEODIFF_changesCount
func.argtypes = [ctypes.c_char_p]
func.restype = ctypes.c_int
# create byte objects from the strings
b_string1 = changeset.encode('utf-8')
nchanges = func(b_string1)
if nchanges < 0:
raise GeoDiffLibError("changes_count")
return nchanges