/usr/lib/python3/dist-packages/panoramisk/actions.py is in python3-panoramisk 1.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | # -*- coding: utf-8 -*-
from .utils import asyncio
from . import utils
class Action(utils.CaseInsensitiveDict):
"""Dict like object to handle actions. Generate action ids for you:
..
>>> utils.IdGenerator.reset('myuuid')
.. code-block:: python
>>> action = Action({'Action': 'Status'})
>>> print(action) # doctest: +NORMALIZE_WHITESPACE
Action: Status
ActionID: action/myuuid/1/1
>>> action = Action({'Action': 'SIPnotify',
... 'Variable': ['1', '2']})
>>> print(action) # doctest: +NORMALIZE_WHITESPACE
Action: SIPnotify
ActionID: action/myuuid/1/2
Variable: 1
Variable: 2
"""
action_id_generator = utils.IdGenerator('action')
def __init__(self, *args, **kwargs):
self.as_list = kwargs.pop('as_list', False)
super(Action, self).__init__(*args, **kwargs)
if 'actionid' not in self:
self['ActionID'] = self.action_id_generator()
self.responses = []
self.future = asyncio.Future()
@property
def id(self):
return self.actionid
action_id = id
def __str__(self):
action = []
for k, v in sorted(self.items()):
if isinstance(v, (list, tuple)):
action.extend(['%s: %s' % (k, i) for i in v])
else:
action.append('%s: %s' % (k, v))
action.append(utils.EOL)
return utils.EOL.join(action)
@property
def multi(self):
resp = self.responses[0]
msg = resp.message.lower()
if resp.subevent == 'Start':
return True
elif 'will follow' in msg:
return True
elif msg.startswith('added') and msg.endswith('to queue'):
return True
elif msg.endswith('successfully queued') and self.async != 'false':
return True
elif self.as_list:
return True
return False
@property
def completed(self):
resp = self.responses[-1]
if resp.event.endswith('Complete'):
return True
elif resp.subevent in ('End', 'Exec'):
return True
elif resp.response in ('Success', 'Error', 'Fail'):
return True
elif not self.multi:
return True
return False
def add_message(self, message):
self.responses.append(message)
multi = self.multi
if self.completed:
if multi and len(self.responses) > 1:
self.future.set_result(self.responses)
elif not multi:
self.future.set_result(self.responses[0])
else:
return False
return True
class Command(Action):
"""Dict like object to handle Commands.
Generate action/command ids for you:
..
>>> utils.IdGenerator.reset('myuuid')
.. code-block:: python
>>> command = Command({'Command' : 'Do something'})
>>> print(command) # doctest: +NORMALIZE_WHITESPACE
Action: Command
ActionID: action/myuuid/1/1
Command: Do something
CommandID: command/myuuid/1/1
"""
command_id_generator = utils.IdGenerator('command')
def __init__(self, *args, **kwargs):
super(Command, self).__init__(*args, **kwargs)
if 'action' not in self:
self['Action'] = 'Command'
if 'commandid' not in self:
self['CommandID'] = self.command_id_generator()
@property
def id(self):
return self.commandid
@property
def action_id(self):
return self.actionid or None
|