Skip to content

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\pvi_objects\Task.py
 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
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
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

errorChanged: Union[Callable[[PviObject, int], None], Callable[[int], None]] property writable

externalObjects: List[Dict[str, str]] property

name: str property

objectName: str property

status: dict property writable

read the task status

type: T_POBJ_TYPE property

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\pvi_objects\Task.py
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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\pvi_objects\Task.py
 94
 95
 96
 97
 98
 99
100
101
def cycle(self, numberOfCycles : int = 1)->None:
    '''
    defines cycles for resume

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

kill()

Source code in pvi\pvi_objects\Object.py
262
263
264
265
266
267
268
269
def kill(self) -> None:
    if self._linkID != 0 and self._connection is not None:  
        self._connection._linkIDs.pop(self._linkID)
        self._connection._pviObjects.remove(self)
        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\pvi_objects\Task.py
104
105
106
107
108
109
110
111
112
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\pvi_objects\Task.py
80
81
82
83
84
85
86
87
88
89
90
91
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\pvi_objects\Task.py
115
116
117
118
119
120
121
122
123
def stop(self)->None:
    '''
    stop task
    '''
    try:
        self.status = "Stop" 
    except PviError as e:
        if e.number != 11166:
            raise e