Class definition

Task

Bases: PviObject

representing a task object

SNMP : can be used but is not necessary

Typical usage example:

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

Source code in pvi\Task.py

class Task(PviObject):
    '''representing a task object

    SNMP : can be used but is not necessary

    Typical usage example:
    ```
    cpu = Cpu( device, 'myArsim', CD='/IP=127.0.0.1' )
    task1 = Task( cpu, 'mainlogic')
    temperature = Variable( task1, 'gHeating.status.actTemp' )
    ```       
    '''
    def __init__( self, parent : PviObject, name : str, **objectDescriptor: Union[str,int, float]):
        '''
        Args:
            parent: CPU object  
            name : name of the task in PVI hierarchy. Will be used for name of task in plc if possible.
            objectDescriptor : 
                ANSL & INA2000 : see PVI documentation for 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_TASK, name, **objectDescriptor)


    @property
    def status(self) -> dict:
        '''
        read the task status        
        '''
        return super().status


    @status.setter
    def status(self, status : str ):
        '''
        sets the task status

        Args:
            status: 'Start', 'Stop', 'Resume', 'Cycle(<x<)'
        '''        
        s = create_string_buffer(b"ST=" + status.encode())
        self._result = PviXWrite( self._hPvi, self._linkID, POBJ_ACC_STATUS, byref(s), sizeof(s), None, 0 )
        if self._result != 0:
            raise PviError(self._result, self)


    def start(self)->None:
        '''
        start a task

        Returns:
            True if status was changed
        '''
        try:
            self.status = "Start"
        except PviError as e:
            if e.number != 11166:
                raise e


    def cycle(self, numberOfCycles : int = 1)->None:
        '''
        defines cycles for resume

        Args:
            numberOfCycles: number of single steps
        '''
        self.status = f"Cycle({numberOfCycles})"


    def resume(self)->None:
        '''
        resume a stopped task
        '''
        try:
            self.status = "Resume"
        except PviError as e:
            if e.number != 11166:
                raise e


    def stop(self)->None:
        '''
        stop task
        '''
        try:
            self.status = "Stop" 
        except PviError as e:
            if e.number != 11166:
                raise e


    @property
    def variables(self)-> List[str]:
        '''
        list of local variables
        '''
        s = create_string_buffer(b'\000' * 65536)   
        self._result = PviXRead( self._hPvi, self._linkID, POBJ_ACC_LIST_PVAR, None, 0, byref(s), sizeof(s) )
        if( self._result == 0 ):
            s = str(s, 'ascii').rstrip('\x00')
            variables = [v.split(' ')[0] for v in s.split('\t')]
            return variables
        else:
            raise PviError(self._result, self)


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

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

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

read the task status

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

variables: List[str] property

list of local variables

__init__(parent, name, **objectDescriptor)

Parameters:

Name	Type	Description	Default
parent	PviObject	CPU object	required
name		name of the task in PVI hierarchy. Will be used for name of task in plc if possible.	required
objectDescriptor		ANSL & INA2000 : see PVI documentation for details	{}

Source code in pvi\Task.py

def __init__( self, parent : PviObject, name : str, **objectDescriptor: Union[str,int, float]):
    '''
    Args:
        parent: CPU object  
        name : name of the task in PVI hierarchy. Will be used for name of task in plc if possible.
        objectDescriptor : 
            ANSL & INA2000 : see PVI documentation for 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_TASK, name, **objectDescriptor)

cycle(numberOfCycles=1)

defines cycles for resume

Parameters:

Name	Type	Description	Default
numberOfCycles	int	number of single steps	1

Source code in pvi\Task.py

def cycle(self, numberOfCycles : int = 1)->None:
    '''
    defines cycles for resume

    Args:
        numberOfCycles: number of single steps
    '''
    self.status = f"Cycle({numberOfCycles})"

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)

resume()

resume a stopped task

Source code in pvi\Task.py

def resume(self)->None:
    '''
    resume a stopped task
    '''
    try:
        self.status = "Resume"
    except PviError as e:
        if e.number != 11166:
            raise e

start()

start a task

Returns:

Type	Description
None	True if status was changed

Source code in pvi\Task.py

def start(self)->None:
    '''
    start a task

    Returns:
        True if status was changed
    '''
    try:
        self.status = "Start"
    except PviError as e:
        if e.number != 11166:
            raise e

stop()

stop task

Source code in pvi\Task.py

def stop(self)->None:
    '''
    stop task
    '''
    try:
        self.status = "Stop" 
    except PviError as e:
        if e.number != 11166:
            raise e