# Licensed under the MIT license
# http://opensource.org/licenses/mit-license.php
# Copyright (C) 2006 Fluendo, S.A. (www.fluendo.com).
# Copyright 2006,2007,2008,2009 Frank Scholz <coherence@beebits.net>
from twisted.python.util import OrderedDict
from coherence import log
[docs]class Argument:
def __init__(self, name, direction, state_variable):
self.name = name
self.direction = direction
self.state_variable = state_variable
[docs] def get_name(self):
return self.name
[docs] def get_direction(self):
return self.direction
[docs] def get_state_variable(self):
return self.state_variable
def __repr__(self):
return \
f'Argument: {self.get_name()}, {self.get_direction()}, ' \
f'{self.get_state_variable()}'
[docs] def as_tuples(self):
r = []
r.append(('Name', self.name))
r.append(('Direction', self.direction))
r.append(('Related State Variable', self.state_variable))
return r
[docs] def as_dict(self):
return {'name': self.name, 'direction': self.direction,
'related_state_variable': self.state_variable}
[docs]class Action(log.LogAble):
logCategory = 'action'
def __init__(self, service, name, implementation, arguments_list):
log.LogAble.__init__(self)
self.service = service
self.name = name
self.implementation = implementation
self.arguments_list = arguments_list
[docs] def _get_client(self):
client = self.service._get_client(self.name)
return client
[docs] def get_name(self):
return self.name
[docs] def get_implementation(self):
return self.implementation
[docs] def get_arguments_list(self):
return self.arguments_list
[docs] def get_in_arguments(self):
return [arg for arg in self.arguments_list if
arg.get_direction() == 'in']
[docs] def get_out_arguments(self):
return [arg for arg in self.arguments_list if
arg.get_direction() == 'out']
[docs] def get_service(self):
return self.service
[docs] def set_callback(self, callback):
self.callback = callback
[docs] def get_callback(self):
try:
return self.callback
except AttributeError:
return None
[docs] def call(self, *args, **kwargs):
self.info(f'calling {self.name}')
in_arguments = self.get_in_arguments()
args_names = [a.get_name() for a in in_arguments]
self.info(f'in arguments {args_names}')
instance_id = 0
for arg_name, arg in kwargs.items():
al = [a for a in in_arguments if arg_name == a.get_name()]
if len(al) > 0:
in_arguments.remove(al[0])
else:
self.error(
f'argument {arg_name} not valid for action {self.name}')
return
if arg_name == 'InstanceID':
instance_id = int(arg)
if len(in_arguments) > 0:
args_names = [a.get_name() for a in in_arguments]
self.error(f'argument {[args_names]} '
f'missing for action {self.name}')
return
action_name = self.name
if (hasattr(self.service.device.client, 'overlay_actions') and
self.name in self.service.device.client.overlay_actions):
self.info(
f'we have an overlay method '
f'{self.service.device.client.overlay_actions[self.name]} '
f'for action {self.name}')
action_name, kwargs = self.service.device.client.overlay_actions[
self.name](**kwargs)
self.info(f'changing action to {action_name} {kwargs}')
def got_error(failure):
self.warning(f'error on {self.name} request with '
f'{self.service.service_type} '
f'{self.service.control_url}')
self.info(failure)
return failure
if hasattr(self.service.device.client, 'overlay_headers'):
self.info(f'action call has headers {"headers"}' in kwargs)
if 'headers' in kwargs:
kwargs['headers'].update(
self.service.device.client.overlay_headers)
else:
kwargs['headers'] = self.service.device.client.overlay_headers
self.info(f'action call with new/updated headers '
f'{kwargs["headers"]}')
client = self._get_client()
ordered_arguments = OrderedDict()
for argument in self.get_in_arguments():
ordered_arguments[argument.name] = kwargs[argument.name]
if 'headers' in kwargs:
ordered_arguments['headers'] = kwargs['headers']
d = client.callRemote(action_name, ordered_arguments)
d.addCallback(self.got_results, instance_id=instance_id,
name=action_name)
d.addErrback(got_error)
return d
[docs] def got_results(self, results, instance_id, name):
instance_id = int(instance_id)
out_arguments = self.get_out_arguments()
self.info(f'call {name} (instance {instance_id:d}) '
f'returns {len(out_arguments):d} arguments')
# XXX A_ARG_TYPE_ arguments probably don't need a variable update
# if len(out_arguments) == 1:
# self.service.get_state_variable(
# out_arguments[0].get_state_variable(),
# instance_id).update(results)
# elif len(out_arguments) > 1:
if len(out_arguments) > 0:
for arg_name, value in list(results.items()):
state_variable_name = \
[a.get_state_variable() for a in out_arguments if
a.get_name() == arg_name]
self.service.get_state_variable(
state_variable_name[0], instance_id).update(value)
return results
def __repr__(self):
return \
f'Action: {self.get_name()} [{self.get_implementation()}], ' \
f'({len(self.get_arguments_list())} args)'
[docs] def as_tuples(self):
return [('Name', self.get_name()),
('Number of \'in\' arguments', len(self.get_in_arguments())),
('Number of \'out\' arguments', len(self.get_out_arguments()))]
[docs] def as_dict(self):
return {'name': self.get_name(),
'arguments': [a.as_dict() for a in self.arguments_list]}