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
|
import threading
from django.contrib.gis.geos.base import GEOSBase
from django.contrib.gis.geos.libgeos import CONTEXT_PTR, error_h, lgeos, notice_h
class GEOSContextHandle(GEOSBase):
"""Represent a GEOS context handle."""
ptr_type = CONTEXT_PTR
destructor = lgeos.GEOS_finish_r
def __init__(self):
self.ptr = lgeos.GEOS_init_r()
lgeos.GEOSContext_setNoticeHandler_r(self.ptr, notice_h)
lgeos.GEOSContext_setErrorHandler_r(self.ptr, error_h)
# Defining a thread-local object and creating an instance
# to hold a reference to GEOSContextHandle for this thread.
class GEOSContext(threading.local):
handle = None
thread_context = GEOSContext()
class GEOSFunc:
"""
Serve as a wrapper for GEOS C Functions. Use thread-safe function
variants when available.
"""
def __init__(self, func_name):
# GEOS thread-safe function signatures end with '_r' and take an
# additional context handle parameter.
self.cfunc = getattr(lgeos, func_name + "_r")
# Create a reference to thread_context so it's not garbage-collected
# before an attempt to call this object.
self.thread_context = thread_context
def __call__(self, *args):
# Create a context handle if one doesn't exist for this thread.
self.thread_context.handle = self.thread_context.handle or GEOSContextHandle()
# Call the threaded GEOS routine with the pointer of the context handle
# as the first argument.
return self.cfunc(self.thread_context.handle.ptr, *args)
def __str__(self):
return self.cfunc.__name__
# argtypes property
def _get_argtypes(self):
return self.cfunc.argtypes
def _set_argtypes(self, argtypes):
self.cfunc.argtypes = [CONTEXT_PTR, *argtypes]
argtypes = property(_get_argtypes, _set_argtypes)
# restype property
def _get_restype(self):
return self.cfunc.restype
def _set_restype(self, restype):
self.cfunc.restype = restype
restype = property(_get_restype, _set_restype)
# errcheck property
def _get_errcheck(self):
return self.cfunc.errcheck
def _set_errcheck(self, errcheck):
self.cfunc.errcheck = errcheck
errcheck = property(_get_errcheck, _set_errcheck)
|
