Source code for automate.extensions.rpc.rpc

# -*- coding: utf-8 -*-
# (c) 2015 Tuomas Airaksinen
#
# This file is part of automate-rpc.
#
# automate-rpc is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# automate-rpc is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with automate-rpc.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import unicode_literals
from builtins import object
from . import wsgi_xmlrpc
from traits.api import CSet, Str, Any

from automate.extensions.wsgi.abstractwsgi import TornadoService


[docs]class ExternalApi(object): def __init__(self, system, tag): self.system = system self.tag = tag
[docs] def set_status(self, name, status): """ Set sensor ``name`` status to ``status``. """ getattr(self.system, name).status = status return True
[docs] def get_status(self, name): """ Get status of object with name ``name``. """ return getattr(self.system, name).status
[docs] def set_object_status(self, statusdict): """ Set statuses from a dictionary of format ``{name: status}`` """ for name, value in statusdict.items(): getattr(self.system, name).status = value return True
[docs] def toggle_object_status(self, objname): """ Toggle boolean-valued sensor status between ``True`` and ``False``. """ o = getattr(self.system, objname) o.status = not o.status self.system.flush() return o.status
[docs] def get_sensors(self): """ Get sensors as a dictionary of format ``{name: status}`` """ return {i.name: i.status for i in self.system.sensors}
[docs] def get_websensors(self): """ Get sensors with defined tag as a dictionary of format ``{name: status}`` """ return {i.name: i.status for i in self.system.sensors if self.tag & i.tags}
[docs] def get_actuators(self): """ Get actuators as a dictionary of format ``{name: status}`` """ return {i.name: i.status for i in self.system.actuators}
[docs] def flush(self): """ Flush the system queue. If you have just set a status and then read a value, it might be necessary to flush queue first, such that related changes have been applied. """ self.system.flush()
[docs] def is_alive(self): """ Simple RPC command that returns always True. """ return True
[docs] def log(self): """ Return recent log entries as a string. """ logserv = self.system.request_service('LogStoreService') return logserv.lastlog(html=False)
[docs]class RpcService(TornadoService): #: Tags that are displayed via get_websensors RPC function view_tags = CSet(trait=Str, value={'rpc'}) #: If you want to define custom api (similar to, or derived from :class:`.ExternalApi`, it can be given here. api = Any def get_wsgi_application(self): instance = self.api or ExternalApi(self.system, tag=self.view_tags) wsgiapp = wsgi_xmlrpc.WSGIXMLRPCApplication(instance=instance) return wsgiapp