Class definition

Variable

Bases: PviObject

representing a PVI Variable object

Typical usage example:

task1 = Task( cpu, 'mainlogic')  
temperature = Variable( task1, 'gHeating.status.actTemp' )  

Source code in pvi\Variable.py

class Variable(PviObject):
    '''representing a PVI Variable object

    Typical usage example:
    ```
    task1 = Task( cpu, 'mainlogic')  
    temperature = Variable( task1, 'gHeating.status.actTemp' )  
    ```        
    '''
    def __init__( self, parent : PviObject, name : str, **objectDescriptor: Union[str,int, float] ):
        '''
        Args:
            parent : the task object (or the CPU object in case of a global variable)
            name : name of the variable in PVI hierarchy. Will be used for name of variable in plc if possible.
            objectDescriptor : e.g.  - see PVI documentation for details
                ANSL : e.g. CD="/RO=TempField[7] /ROI=1"

                INA2000 : e.g. AT=rwe CD="/RO=View::TempField[7]"

                SNMP : name of SNMP variable e.g. CD='serialNumber' or CD='ipAddress'

        '''
        if parent.type != T_POBJ_TYPE.POBJ_CPU and  parent.type != T_POBJ_TYPE.POBJ_TASK and  parent.type != T_POBJ_TYPE.POBJ_STATION:
            raise PviError(12009, self )        
        self._value = None 
        self._valueChanged = lambda _ : _
        self._valueChangedArgCount = 0      

        if 'CD' not in objectDescriptor:
            objectDescriptor.update({'CD':name})
        if 'RF' not in objectDescriptor:
            objectDescriptor.update({'RF':0}) # do not cyclic refrehs variables by default       
        super().__init__( parent, T_POBJ_TYPE.POBJ_PVAR, name, **objectDescriptor)
        self._variableTypeDescription = VariableTypeDescription(self._hPvi)

    def __repr__(self):
        return f"Variable( name={self._name}, linkID={self._linkID}, VT={self._objectDescriptor.get('VT')} )"


    @property
    def writable(self) -> bool:
        '''
        signals if this variable can be written
        '''
        if self._variableTypeDescription.vt != PvType.UNKNOWN:
            access = str(self._objectDescriptor.get('AT', ''))
            return 'w' in access # data type already read and write access ?
        else:
            return False

    @property
    def readable(self) -> bool:
        '''
        signals if this variable can be read
        '''
        if self._variableTypeDescription.vt != PvType.UNKNOWN:
            access = str(self._objectDescriptor.get('AT', ''))
            return 'r' in access # data type already read and read access ?
        else:
            return False

    def __readRawData(self, wParam, responseInfo):
        '''
        reads data (byte) from PVI_ACC_DATA or PVI_EVENT_DATA
        > wParam: points to response data
        > responseInfo: responseInfo or 'None' in case of synchronous read request
        '''
        if self._variableTypeDescription.vt == PvType.UNKNOWN:
            try:
                self._objectDescriptor = self._variableTypeDescription.readFrom(self._linkID)
            except PviError as e:
                raise PviError( e.number, self._name )

        buffer = create_string_buffer(self._variableTypeDescription.vn * self._variableTypeDescription.vl)

        if responseInfo: # data is answer of a request
            self._result = PviXReadResponse( self._hPvi, wParam, byref(buffer), sizeof(buffer) )
        else: # data shall be immediately read
            self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_DATA, None, 0, byref(buffer), sizeof(buffer) )

        if self._result == 0:
            self._value = self._variableTypeDescription.readFromBuffer(bytes(buffer))
        else:
            raise PviError(self._result, self)

        if self._errorChanged: # fire callback in case of response
            self._errorChanged(0)


    def _eventData( self, wParam, responseInfo ):
        self.__readRawData( wParam, responseInfo )
        if self._valueChangedArgCount == 1:
            self._valueChanged(self._value)
        elif self._valueChangedArgCount == 2:
            self._valueChanged(self, self._value) # type: ignore


    def _eventDataType( self, wParam, responseInfo ):
        s = create_string_buffer(b'\000' * 64*1024)       
        self._result = PviXReadResponse( self._hPvi, wParam, s, sizeof(s) )
        if self._result == 0:
            pass # TODO


    @property
    def value(self) ->Any:
        '''
        read value
        '''
        self.__readRawData( 0, None )
        return self._value


    @value.setter
    def value(self,v : Any):
        '''
        set value
        '''
        if self._variableTypeDescription.vt == PvType.UNKNOWN:
            try:
                self._objectDescriptor = self._variableTypeDescription.readFrom(self._linkID)
            except PviError as e:
                raise PviError( e.number, self._name )                

        if self._variableTypeDescription.vt == PvType.STRUCT: 
            raise ValueError(f'Writing of struct not implemented\n{repr(self)}')
        else:
            buffer = self._variableTypeDescription.writeToBuffer(v)
            self._result = PviXWrite( self._hPvi, self._linkID, POBJ_ACC_DATA, byref(buffer), sizeof(buffer), None, 0 )  
        if self._result:
            raise PviError(self._result, self)


    @property
    def valueChanged(self):
        '''
        callback for 'value changed'

        accepts:
        fn( value)
        or
        fn( value, vo ) where 'vo' is the Variable object itself

        '''
        return self._valueChanged

    @valueChanged.setter
    def valueChanged(self, cb):
        if callable(cb):
            self._valueChanged = cb
            self._valueChangedArgCount = len(signature(cb).parameters)
        else:
            raise ValueError


    @property
    def dataType(self) -> str:
        '''
        variable's type as string. e.g 'i8', 'i16', 'f32', 'boolean', 'string'
        in case of an array the indices are given in a bracket, e.g. 'i32[0..2]'
        '''
        t = "?"
        if self._variableTypeDescription.vt == PvType.UNKNOWN:
            try:
                self._objectDescriptor = self._variableTypeDescription.readFrom(self._linkID)
            except PviError as e:
                raise PviError( e.number, self._name )                  

        if self._variableTypeDescription.vt == PvType.STRUCT:
            t = self._variableTypeDescription.sn
            if self.isArray: # is variable an array of structs ?
                t += f'[{self._variableTypeDescription.indices[0]}..{self._variableTypeDescription.indices[1]}]'
        else:
            t = self._variableTypeDescription.vt.value
            if self.isArray: # is variable an array ?
                t += f'[{self._variableTypeDescription.indices[0]}..{self._variableTypeDescription.indices[1]}]'            
        return t

    @property
    def isArray(self) -> bool:
        return( self._variableTypeDescription.vn > 1 or self._variableTypeDescription.sn.startswith('a'))

    @property
    def attributes(self) -> dict:
        '''
        object attributes

        Typical usage example:
        ```
        temperature = Variable( task1, name='gHeating.status.actTemp', AT = 'r' )        
        ```

        '''
        s = create_string_buffer(b'\000' * 256)   
        self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_TYPE , None, 0, byref(s), sizeof(s) )
        if self._result == 0:
            ret = dict()
            ret.update( dictFromParameterPairString(str(s, 'ascii').rstrip('\x00')))
            self._objectDescriptor.update(ret)
            return ret
        else:
            raise PviError(self._result, self)  

    @attributes.setter
    def attributes(self, a : str ):
        '''
        object attributes
        '''
        s = create_string_buffer(a.encode('ascii'))
        self._result = PviXWrite( self._hPvi, self._linkID, POBJ_ACC_TYPE, byref(s), sizeof(s), None, 0 ) 
        if self._result:
            raise PviError(self._result, self)  



    @property
    def refresh(self) -> int:
        '''
        refresh time [ms]

        Typical usage example:
        ```
        temperature = Variable( task1, name='gHeating.status.actTemp', RF = 2000 )      
        ```
        '''
        t = c_int32(0)
        self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_REFRESH , None, 0, byref(t), sizeof(t) )
        if self._result == 0:
            self._objectDescriptor.update({'RF' : t.value }) # type: ignore
            return t.value
        else:
            raise PviError(self._result, self)  

    @refresh.setter
    def refresh(self, time : int ):
        '''
        refresh time [ms]
        '''
        t = c_int32(time)
        self._result = PviXWrite( self._hPvi, self._linkID, POBJ_ACC_REFRESH, byref(t), sizeof(t), None, 0 ) 
        if self._result:
            raise PviError(self._result, self) 
        self._objectDescriptor.update({'RF' : time }) # type: ignore 


    @property
    def hysteresis(self) -> float:
        '''
        event hysteresis

        Typical usage example:
        ```
        temperature = Variable( task1, name='gHeating.status.actTemp', HY = 5.0, LinkDescriptor="LT=prc" )        
        ```
        '''
        s = create_string_buffer(b'\000' * 10)   
        self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_HYSTERESE , None, 0, byref(s), sizeof(s) )
        if self._result == 0:
            s = str(s, 'ascii').rstrip('\x00')
            self._objectDescriptor.update({'HY' : s }) # type: ignore  
            try:          
                return float(s)
            except ValueError:
                return 0.0
        else:
            raise PviError(self._result, self)  

    @hysteresis.setter
    def hysteresis(self, h : float ):
        '''
        event hysteresis
        '''
        s = create_string_buffer(str(h).encode('ascii'))
        self._result = PviXWrite( self._hPvi, self._linkID, POBJ_ACC_HYSTERESE, byref(s), sizeof(s), None, 0 ) 
        if self._result:
            raise PviError(self._result, self)  
        self._objectDescriptor.update({'HY' : h }) # type: ignore                    

attributes: dict property writable

object attributes

Typical usage example:

temperature = Variable( task1, name='gHeating.status.actTemp', AT = 'r' )        

dataType: str property

variable's type as string. e.g 'i8', 'i16', 'f32', 'boolean', 'string' in case of an array the indices are given in a bracket, e.g. 'i32[0..2]'

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

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'}]

hysteresis: float property writable

event hysteresis

Typical usage example:

temperature = Variable( task1, name='gHeating.status.actTemp', HY = 5.0, LinkDescriptor="LT=prc" )        

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

readable: bool property

signals if this variable can be read

refresh: int property writable

refresh time [ms]

Typical usage example:

temperature = Variable( task1, name='gHeating.status.actTemp', RF = 2000 )      

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

value: Any property writable

read value

valueChanged property writable

callback for 'value changed'

accepts: fn( value) or fn( value, vo ) where 'vo' is the Variable object itself

writable: bool property

signals if this variable can be written

__init__(parent, name, **objectDescriptor)

Parameters:

Name	Type	Description	Default
parent		the task object (or the CPU object in case of a global variable)	required
name		name of the variable in PVI hierarchy. Will be used for name of variable in plc if possible.	required
objectDescriptor		e.g. - see PVI documentation for details ANSL : e.g. CD="/RO=TempField[7] /ROI=1" INA2000 : e.g. AT=rwe CD="/RO=View::TempField[7]" SNMP : name of SNMP variable e.g. CD='serialNumber' or CD='ipAddress'	{}

Source code in pvi\Variable.py

def __init__( self, parent : PviObject, name : str, **objectDescriptor: Union[str,int, float] ):
    '''
    Args:
        parent : the task object (or the CPU object in case of a global variable)
        name : name of the variable in PVI hierarchy. Will be used for name of variable in plc if possible.
        objectDescriptor : e.g.  - see PVI documentation for details
            ANSL : e.g. CD="/RO=TempField[7] /ROI=1"

            INA2000 : e.g. AT=rwe CD="/RO=View::TempField[7]"

            SNMP : name of SNMP variable e.g. CD='serialNumber' or CD='ipAddress'

    '''
    if parent.type != T_POBJ_TYPE.POBJ_CPU and  parent.type != T_POBJ_TYPE.POBJ_TASK and  parent.type != T_POBJ_TYPE.POBJ_STATION:
        raise PviError(12009, self )        
    self._value = None 
    self._valueChanged = lambda _ : _
    self._valueChangedArgCount = 0      

    if 'CD' not in objectDescriptor:
        objectDescriptor.update({'CD':name})
    if 'RF' not in objectDescriptor:
        objectDescriptor.update({'RF':0}) # do not cyclic refrehs variables by default       
    super().__init__( parent, T_POBJ_TYPE.POBJ_PVAR, name, **objectDescriptor)
    self._variableTypeDescription = VariableTypeDescription(self._hPvi)

__readRawData(wParam, responseInfo)

reads data (byte) from PVI_ACC_DATA or PVI_EVENT_DATA

wParam: points to response data responseInfo: responseInfo or 'None' in case of synchronous read request

Source code in pvi\Variable.py

def __readRawData(self, wParam, responseInfo):
    '''
    reads data (byte) from PVI_ACC_DATA or PVI_EVENT_DATA
    > wParam: points to response data
    > responseInfo: responseInfo or 'None' in case of synchronous read request
    '''
    if self._variableTypeDescription.vt == PvType.UNKNOWN:
        try:
            self._objectDescriptor = self._variableTypeDescription.readFrom(self._linkID)
        except PviError as e:
            raise PviError( e.number, self._name )

    buffer = create_string_buffer(self._variableTypeDescription.vn * self._variableTypeDescription.vl)

    if responseInfo: # data is answer of a request
        self._result = PviXReadResponse( self._hPvi, wParam, byref(buffer), sizeof(buffer) )
    else: # data shall be immediately read
        self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_DATA, None, 0, byref(buffer), sizeof(buffer) )

    if self._result == 0:
        self._value = self._variableTypeDescription.readFromBuffer(bytes(buffer))
    else:
        raise PviError(self._result, self)

    if self._errorChanged: # fire callback in case of response
        self._errorChanged(0)

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

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)