Skip to content

Class definition

Module

Bases: PviObject

class representing modules

SNMP : can be used but is not necessary

Typical usage example:

cpu = Cpu( device, 'myArsim', CD='/IP=127.0.0.1' )
module = Module( cpu, 'bigmod' )
Source code in pvi\Module.py 
 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
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class Module(PviObject):
    '''class representing modules

    SNMP : can be used but is not necessary

    Typical usage example:
    ```
    cpu = Cpu( device, 'myArsim', CD='/IP=127.0.0.1' )
    module = Module( cpu, 'bigmod' )
    ```
    '''
    def __init__( self, parent : PviObject, name : str, **objectDescriptor: Union[str,int, float]):
        '''
        Args:
            parent : CPU object
            name : name of module
            objectDescriptor : see PVI documentation for more details
        '''
        if parent.type != T_POBJ_TYPE.POBJ_CPU:
            raise PviError(12009, self)
        if 'CD' not in objectDescriptor:
            objectDescriptor.update({'CD':name})                    
        super().__init__( parent, T_POBJ_TYPE.POBJ_MODULE, name, **objectDescriptor)
        self._uploaded = None
        self._progress = None


    def __repr__(self):
        return f"Module( name={self._name}, linkID={self._linkID} )"


    def _eventUploadStream( self, wParam, responseInfo, dataLen : int ):
        '''
        (internal) upload a data module as stream
        '''
        s = create_string_buffer(dataLen)       
        self._result = PviXReadResponse( self._hPvi, wParam, s, sizeof(s) )
        if self._result == 0:
            if self._uploaded:
                sig = inspect.signature(self._uploaded)
                if len(sig.parameters) == 2:
                    self._uploaded(self, s.raw)
                elif len(sig.parameters) == 1:
                    self._uploaded(s.raw)                      
        else:
            raise PviError(self._result, self)


    def _eventUploadLogData( self, wParam, responseInfo, dataLen : int ):
        '''
        (internal) upload XML logger data (ANSL only)
        '''
        s = create_string_buffer(dataLen+256)       
        self._result = PviXReadResponse( self._hPvi, wParam, s, sizeof(s) )
        s = s.raw.replace(b'\x00',b'')
        if self._result == 0:
            logger = ET.fromstring(s) 
            entries = list()
            for entry in logger:
                cols = {'Version','RecordId', 'OriginRecordId', 'EventId', 'AddDataSize', 'AddDataFormat', 'Severity', 'Info'}
                for c in cols:
                    try: # try to convert columns into integer values
                        value = int(entry.attrib[c])
                        entry.attrib.update({c : str(value)} )
                    except:
                        pass
                try: # try to convert timestamp into Python datatype
                    value = datetime.datetime.fromtimestamp( float(entry.attrib['TimestampUtc']) )
                    entry.attrib.update({ 'TimestampUtc' : str(value)} )
                except:
                    pass
                entries.append(entry.attrib)
            if self._uploaded:
                sig = inspect.signature(self._uploaded)
                if len(sig.parameters) == 2:
                    self._uploaded(self, entries)
                elif len(sig.parameters) == 1:
                    self._uploaded(entries)      
        else:
            raise PviError(self._result, self)


    def _eventUploadModData( self, wParam, responseInfo, dataLen : int ):
        '''
        (internal) upload logger data (INA2000)
        see GUID 75bf0748-45f2-4610-a68d-53760ab5fa98
        '''
        patternParameterPairs = re.compile(r"\s*([A-Z]{1,4}=\w*)\s*")        
        s = create_string_buffer(dataLen)       
        self._result = PviXReadResponse( self._hPvi, wParam, s, sizeof(s) )
        if self._result == 0:
            entries = list()            
            data = s.raw.split(b'\00')                    
            n = 0
            noOfEntries = int(data[0][3:])
            while n <  noOfEntries*3:
                n += 1 # point to next entry
                entry = dict()
                # read info string
                matches = patternParameterPairs.findall(str(data[n]))            
                for m in matches:
                    if str(m).startswith('TIME'):
                        dt = datetime.datetime.fromtimestamp( int(m[5:]))
                        entry.update({ 'date' : dt})
                    elif str(m).startswith('ID'):
                        id = int(m[3:])
                        entry.update({ 'id' : id})
                    elif str(m).startswith('E'):
                        error = int(m[2:])
                        entry.update({ 'error' : error })                        
                    elif str(m).startswith('INFO'):
                        info = int(m[5:])                        
                        entry.update({ 'info' : info })
                    elif str(m).startswith('LEV'):
                        level = int(m[4:])                        
                        entry.update({ 'level' : level })
                    elif str(m).startswith('TASK'):
                        entry.update({ 'task' : str(data[5:]) })
                # read ascii data
                n += 1
                entry.update({'ascii': data[n] })
                # read binary data
                n += 1
                entry.update({'bin': data[n] })   
                entries.append(entry) 
            if self._uploaded:
                sig = inspect.signature(self._uploaded)
                if len(sig.parameters) == 2:
                    self._uploaded(self, entries)
                elif len(sig.parameters) == 1:
                    self._uploaded(entries)                    
        else:
            raise PviError(self._result, self)


    def _eventProceeding( self, wParam, responseInfo : T_RESPONSE_INFO ):  
        '''
        (internal) return proceeding info
        '''  
        proceedingInfo = T_PROCEEDING_INFO()
        self._result = PviXReadResponse( self._hPvi, wParam, byref(proceedingInfo), sizeof(proceedingInfo) )
        if self._result == 0:
            if self._progress:
                sig = inspect.signature(self._progress)
                if len(sig.parameters) == 2:
                    self._progress(self, int(proceedingInfo.Percent))
                elif len(sig.parameters) == 1:                         
                    self._progress(int(proceedingInfo.Percent))
        else:
            raise PviError(self._result, self)               


    def upload(self, **kwargs : Union[str, Callable]):
        '''
        uploadLoggerData 
        loads logger data if module is a logger module else load binary data

        Args: 
            kwargs:    
                uploaded - callback - is fired when module was uploaded
                progress - callback(int) - returns percentage of progress
                MT - Moduletype e.g. 'BRT', '_LOGM'
        '''
        arguments = ''
        loggerModule = False
        for key, value in kwargs.items():
            if key == 'uploaded':
                if callable(value):
                    self._uploaded = value
                else:
                    raise TypeError("only type 'callable' for argument 'uploaded' allowed !")
            elif key == 'progress':
                if callable(value):
                    self._progress = value
                else:
                    raise TypeError("only type 'callable' for argument 'progress' allowed !")
            elif key == 'MT' and value == '_LOGM':
                loggerModule = True # interpret as logger module
            else:
                arguments += f"{key}={value}"

        # check if module is a logger module
        if loggerModule:
            # try ANSL logger module
            s = create_string_buffer(b'\000' * 4096)   
            self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_LN_XML_LOGM_INFO, None, 0, byref(s), sizeof(s) )
            if self._result == 0:
                s = str(s, 'ascii').rstrip('\x00')
                xmlTree = ET.fromstring(s)
                loggerVersion = xmlTree.attrib.get('Version', '1000').encode('ascii')
                s = create_string_buffer(b'DN=10000000 VI=' + loggerVersion )   
                self._result = PviXReadArgumentRequest( self._hPvi, self._linkID, POBJ_ACC_LN_XML_LOGM_DATA, byref(s), sizeof(s), PVI_HMSG_NIL, SET_PVIFUNCTION, 0 )            
                if self._result:
                    raise PviError(self._result)           
            elif self._result == 12058: # access not aupported ?
                s = create_string_buffer(b'DN=100000') # maximum possible is undocumented.
                self._result = PviXReadArgumentRequest( self._hPvi, self._linkID, POBJ_ACC_MOD_DATA, byref(s), sizeof(s), PVI_HMSG_NIL, SET_PVIFUNCTION, 0    ) 
                if self._result:
                    raise PviError(self._result)           
            else:
                raise PviError(self._result)
        else:
            s = create_string_buffer(bytes(arguments, 'ascii'))
            self._result = PviXReadArgumentRequest( self._hPvi, self._linkID, POBJ_ACC_UPLOAD_STM, byref(s), sizeof(s), PVI_HMSG_NIL, SET_PVIFUNCTION, 0    ) 
            if self._result:
                raise PviError(self._result)

descriptor: dict property

object descriptor example:

temperature = Variable( task1, 'gHeating.status.actTemp' )
...
print( "descriptor=", temperature.name)

results in:

descriptor= {'CD': 'gHeating.status.actTemp', 'RF': 0}

errorChanged: Callable property writable

callback for 'error changed'

It is advisable to always check the error status '0' before accessing an object.

Args:
    cb: callback( PviObject, int ) or callback( int )

typical example:

cpu = Cpu( device, 'myArsim', CD='/IP=127.0.0.1' )
...
def cpuErrorChanged( error : int ):
    if error != 0:
        raise PviError(error)

cpu.errorChanged = cpuErrorChanged

evmask: str property writable

event mask in link descriptor

"e": Change in error state "d": Data changes "f": Change to the data format "c": Change to the connection description "p": Progress information about active requests "s": Status changes "u": Change in the user tag string

externalObjects property

PviObject.externalObjects : list of dict get a list of external objects retrieved by POBJ_ACC_LIST_EXTERN

only available with ANSL, not with INA2000

example:

cpu = Cpu( device, 'myArsim', CD='/IP=127.0.0.1' )
...
print("external objects", cpu.externalObjects )

results in:

external objects [{'name': '$$sysconf', 'type': 'Module'}, {'name': '$arlogsys', 'type': 'Module'}
            ...... name': 'visvc', 'type': 'Module'}]

name: str property

hierarchical PVI object name

example:

name= @Pvi/LNANSL/TCP/myArsim/mainlogic/gHeating.status.actTemp

objectName: str property

object name

example:

temperature = Variable( task1, 'gHeating.status.actTemp' )
...
print( "oname=", temperature.name)

results in:

oname= gHeating.status.actTemp

status: dict property writable

PviObject.status read the object's status

example:

cpu = Cpu( device, 'myArsim', CD='/IP=127.0.0.1' )
task1 = Task( cpu, 'mainlogic')
temperature = Variable( task1, 'gHeating.status.actTemp' )
...
print("status=", cpu.status )

results in:

cpu.status= {'ST': 'WarmStart', 'RunState': 'RUN'}
task1.status {'ST': 'Running'}
temperature.status= {'ST': 'Var', 'SC': 'g'}

type: T_POBJ_TYPE property

object type

example:

temperature = Variable( task1, 'gHeating.status.actTemp' )
...
print( "type=", temperature.type)

results in:

type= T_POBJ_TYPE.POBJ_PVAR

userName: str property

user defined object name defaults to .objectName

userTag: str property writable

user tag

Typical usage example:

temperature = Variable( task1, name='gHeating.status.actTemp', UT="actual water temperature" )

version: str property

PviObject.version read the object's version

__init__(parent, name, **objectDescriptor)

Parameters:

Name Type Description Default
parent

CPU object

 required
name

name of module

 required
objectDescriptor

see PVI documentation for more details

 {}
Source code in pvi\Module.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__( self, parent : PviObject, name : str, **objectDescriptor: Union[str,int, float]):
    '''
    Args:
        parent : CPU object
        name : name of module
        objectDescriptor : see PVI documentation for more details
    '''
    if parent.type != T_POBJ_TYPE.POBJ_CPU:
        raise PviError(12009, self)
    if 'CD' not in objectDescriptor:
        objectDescriptor.update({'CD':name})                    
    super().__init__( parent, T_POBJ_TYPE.POBJ_MODULE, name, **objectDescriptor)
    self._uploaded = None
    self._progress = None

kill()

PviObject.kill: kills this object this should be called when object is not beeing used anymore to save PVI resources

Source code in pvi\Object.py 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
def kill(self):
    '''
    PviObject.kill: kills this object
    this should be called when object is not beeing used anymore
    to save PVI resources
    '''
    if self._linkID != 0 and self._connection != None:

        self._connection._linkIDs.pop(self._linkID) # remove from linkIDs
        self._connection._pviObjects.remove(self) # remove from PviObjects
        self._result = PviXUnlink(self._hPvi, self._linkID)
        self._linkID = 0
        if self._result != 0 and self._result != 12045:
            raise PviError(self._result, self)

upload(**kwargs)

uploadLoggerData loads logger data if module is a logger module else load binary data

Parameters:

Name Type Description Default
kwargs Union[str, Callable]

uploaded - callback - is fired when module was uploaded progress - callback(int) - returns percentage of progress MT - Moduletype e.g. 'BRT', '_LOGM'

 {}
Source code in pvi\Module.py 
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
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def upload(self, **kwargs : Union[str, Callable]):
    '''
    uploadLoggerData 
    loads logger data if module is a logger module else load binary data

    Args: 
        kwargs:    
            uploaded - callback - is fired when module was uploaded
            progress - callback(int) - returns percentage of progress
            MT - Moduletype e.g. 'BRT', '_LOGM'
    '''
    arguments = ''
    loggerModule = False
    for key, value in kwargs.items():
        if key == 'uploaded':
            if callable(value):
                self._uploaded = value
            else:
                raise TypeError("only type 'callable' for argument 'uploaded' allowed !")
        elif key == 'progress':
            if callable(value):
                self._progress = value
            else:
                raise TypeError("only type 'callable' for argument 'progress' allowed !")
        elif key == 'MT' and value == '_LOGM':
            loggerModule = True # interpret as logger module
        else:
            arguments += f"{key}={value}"

    # check if module is a logger module
    if loggerModule:
        # try ANSL logger module
        s = create_string_buffer(b'\000' * 4096)   
        self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_LN_XML_LOGM_INFO, None, 0, byref(s), sizeof(s) )
        if self._result == 0:
            s = str(s, 'ascii').rstrip('\x00')
            xmlTree = ET.fromstring(s)
            loggerVersion = xmlTree.attrib.get('Version', '1000').encode('ascii')
            s = create_string_buffer(b'DN=10000000 VI=' + loggerVersion )   
            self._result = PviXReadArgumentRequest( self._hPvi, self._linkID, POBJ_ACC_LN_XML_LOGM_DATA, byref(s), sizeof(s), PVI_HMSG_NIL, SET_PVIFUNCTION, 0 )            
            if self._result:
                raise PviError(self._result)           
        elif self._result == 12058: # access not aupported ?
            s = create_string_buffer(b'DN=100000') # maximum possible is undocumented.
            self._result = PviXReadArgumentRequest( self._hPvi, self._linkID, POBJ_ACC_MOD_DATA, byref(s), sizeof(s), PVI_HMSG_NIL, SET_PVIFUNCTION, 0    ) 
            if self._result:
                raise PviError(self._result)           
        else:
            raise PviError(self._result)
    else:
        s = create_string_buffer(bytes(arguments, 'ascii'))
        self._result = PviXReadArgumentRequest( self._hPvi, self._linkID, POBJ_ACC_UPLOAD_STM, byref(s), sizeof(s), PVI_HMSG_NIL, SET_PVIFUNCTION, 0    ) 
        if self._result:
            raise PviError(self._result)