File: object.pyx

package info (click to toggle)
obitools 3.0.1~b26%2Bdfsg-4
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 26,756 kB
  • sloc: ansic: 24,299; python: 657; sh: 27; makefile: 21
file content (106 lines) | stat: -rwxr-xr-x 2,762 bytes parent folder | download
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
#cython: language_level=3

import functools


__c_cython_mapping__ = {}


cdef class OBIObject:

    def __init__(self, __internalCall__) :
                
        if __internalCall__ != 987654 or \
           type(self) == OBIObject or \
           type(self) == OBIWrapper or \
           not isinstance(self, OBIObject) :
            
            raise RuntimeError('OBIObject constructor can not be called directly')
        
        self._dependent_objects = {}


    cdef register(self, OBIObject object):
        self._dependent_objects[id(object)] = object

        
    cdef unregister(self, OBIObject object):
        del self._dependent_objects[id(object)]
        

    def close(self):
    
        cdef OBIObject object
        cdef list      to_close = list((self._dependent_objects).values())
        
        for object in to_close:
            object.close()
        
        assert len(self._dependent_objects.values()) == 0


cdef class OBIWrapper(OBIObject) :
    '''
    The OBIWrapper class enables to wrap a C object representing a DMS or an element from a DMS.
    '''
    
    @staticmethod
    def checkIsActive(instance):
        '''
        Decorator function to check that an instance is still active (associated pointer not NULL)
        '''
        @functools.wraps(instance)
        def check(self,*args,**kargs):
            if self.dead:
                raise OBIDeactivatedInstanceError()            
            return instance(self,*args,**kargs)
        return check
    
    cdef inline size_t cid(self) :
        return <size_t>(self._pointer)
    

    cdef inline bint active(self) :
        return self._pointer != NULL


    def close(self):
        if (self._pointer != NULL):
            OBIObject.close(self)
            del __c_cython_mapping__[<size_t>self._pointer]
            self._pointer = NULL
            
        assert len(self._dependent_objects.values()) == 0


    def __dealloc__(self):
        '''
        Destructor of any OBI instance.
        
        The destructor automatically calls the `close` method and
        therefore closes and frees all associated objects and memory.
        '''
        self.close()

    @property
    def dead(self):
        return self._pointer==NULL
    
    @staticmethod
    cdef object new_wrapper(type constructor, void* pointer) :
    
        cdef OBIWrapper o
        
        if (<size_t>pointer in __c_cython_mapping__):
            #print("Pointer already in cython dict")
            return __c_cython_mapping__[<size_t>pointer]
        else:
            o = constructor(987654)
            o._pointer = pointer
            __c_cython_mapping__[<size_t>pointer] = o
            return o


cdef class OBIDeactivatedInstanceError(Exception):
    pass