Source code for rsbhostmonitor.executable

# ============================================================
# Copyright (C) 2015 by Johannes Wienke <jwienke at techfak dot uni-bielefeld dot de>
# This file may be licensed under the terms of the
# GNU Lesser General Public License Version 3 (the ``LGPL''),
# or (at your option) any later version.
# Software distributed under the License is distributed
# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
# express or implied. See the LGPL for the specific language
# governing rights and limitations.
# You should have received a copy of the LGPL along with this
# program. If not, go to
# or write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
# The development of this software was supported by:
#   CoR-Lab, Research Institute for Cognition and Robotics
#     Bielefeld University
# ============================================================

Provides executable functions.

.. codeauthor:: jwienke

import argparse
import logging
import logging.config
import platform
import signal

import rsb
import rsb.converter

import rst  # pylint: disable=unused-import
import rstsandbox  # pylint: disable=unused-import
from rst.devices.generic.HostInformation_pb2 import HostInformation

from rsbhostmonitor import MonitoringThread

LOGGER_NAME = 'rsbhostmonitor.executable'

[docs]def configure_logging(file_or_flag): """ Configure the python :mod:`logging` system. If the provided argument is a `file` instance, try to use the pointed to file as a configuration for the logging system. Otherwise, if the given argument evaluates to :class:True:, use a default configuration with many logging messages. If everything fails, just log starting from the warning level. Args: file_or_flag (file or bool): either a configuration file pointed by a :ref:`file object <python:bltin-file-objects>` instance or something that evaluates to :class:`bool`. """ if isinstance(file_or_flag, file): logging.config.fileConfig( elif file_or_flag: try: import pkg_resources default_config = pkg_resources.resource_filename( __name__, 'logging.withoutrsb.cfg') print default_config logging.config.fileConfig(default_config) except: logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(LOGGER_NAME) logger.warning( 'Unable to configure logging with the default ' 'configuration file. Falling back to basicConfig.', exc_info=True) else: # at least configure warnings logging.basicConfig(level=logging.WARNING)
[docs]def generate_argument_parser(): """ Generates an :module:`argparse` parser for the adapter executable. Returns: argparse.ArgumentParser: configured parser instance with all arguments """ def rsb_scope(candidate): """ Check an :mod:`argparse` argument for being a :class:`rsb.Scope`. """ try: return rsb.Scope(candidate) except ValueError, error: raise argparse.ArgumentTypeError( '{0} is not a valid scope. ' 'Reason: {1}'.format(candidate, error)) def positive_int(candidate): """ Checks a command line argument value for being a positive integer. """ try: num = int(candidate) if num <= 0: raise argparse.ArgumentTypeError('Number must be >= 0.') return num except ValueError, error: raise argparse.ArgumentTypeError( '{0} is not a valid int. Reason: {1}'.format(candidate, error)) parser = argparse.ArgumentParser( description='Monitors a host and exposes results via RSB.', formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument('-o', '--scope', type=rsb_scope, metavar='SCOPE', default=rsb.Scope('/host/{hostname}'.format( hostname=platform.node())), help='Output scope for gathered data.') parser.add_argument('-c', '--cycle', type=positive_int, metavar='MILLIS', default=2000, help='The time between consecutive measurements.') parser.add_argument('-n', '--host', type=str, metavar='HOST', default=platform.node(), help='Host name to report in generated data.') parser.add_argument('-p', '--print', action='store_true', dest='print_results', help='Print gathered results on stdout.') parser.add_argument('-l', '--logging', type=argparse.FileType('r'), metavar='FILE', nargs='?', default=False, const=True, help='Configures the python logging system. If used ' 'without an argument, all logging is enabled to ' 'the console. If used with an argument, the ' 'configuration is read from the specified file.') return parser
[docs]def main(): """ Main method for the host monitor executable. Includes all the necessary work to run as a command line application. """ args = generate_argument_parser().parse_args() configure_logging(args.logging) logger = logging.getLogger(LOGGER_NAME) logger.debug('parsed arguments: %s', args) host_converter = rsb.converter.ProtocolBufferConverter( messageClass=HostInformation) rsb.converter.registerGlobalConverter(host_converter) thread = MonitoringThread(float(args.cycle) / 1000.0, rsb.createInformer(args.scope), print_results=args.print_results, thread.start() print('Started monitoring. Waiting...') try: signal.pause() except KeyboardInterrupt: # expected pass finally: print('Terminating') thread.interrupt() thread.join()
if __name__ == '__main__': main()