<commit_before>#
# Copyright (c) 2015, Open-RnD Sp. z o.o.  All rights reserved.
#
"""Servo driver proxy"""
from __future__ import absolute_import

from sparts.tasks.dbus import DBusTask
from sparts.sparts import option
from ros3ddevcontroller.param.store import ParametersStore
import glib
import logging
import dbus

_log = logging.getLogger(__name__)


class ParamApplyRequest(object):
    """Wrapper for keeping a parameter apply request in check

    :ivar param str: parameter name
    :ivar value: parameter value"""
    def __init__(self, param, value):
        self.param = param
        self.value = value


class ParamApplyError(Exception):
    """Error wrapper class"""
    pass


class ServoTask(DBusTask):
    """Servo driver proxy. The proxy will automatically find a DBus servo
    service and connect to it.

    Since the task is running in a Glib MainLoop possibly on a
    different thread than the caller of `change_param()` method, a
    trampoline is used. The request is wrapped into
    `ParamApplyRequest` and `change_param()` returns a `Future()`
    object.
    """
    OPT_PREFIX = 'servo-dbus'

    SERVO_DBUS_SERVICE = 'pl.ros3d.servo'
    SERVO_DBUS_PATH = '/pl/ros3d/servo'
    SERVO_DBUS_INTERFACE = "pl.ros3d.servo"
    SERVO_CALL_TIMEOUT_S = 120 # seconds

    session_bus = option(action='store_true', type=bool,
                         default=False,
                         help='Use session bus to access servo service')

    def __init__(self, *args, **kwargs):
        super(ServoTask, self).__init__(*args, **kwargs)

        self.bus = None
        self.servo = None

    def start(self):
        self.asyncRun(self._start_servo)
        super(ServoTask, self).start()

    def _start_servo(self):
        """Start task. Perform necessary setup:
        - bus connection
        - setup servo service name monitoring
        - obtain proxy to servo if possible
        """
        _log.debug('starting servo task for service: %s', self.service.name)
        # get bus
        if self.session_bus:
            self.bus = dbus.SessionBus(private=True)
        else:
            self.bus = dbus.SystemBus(private=True)

        _log.info('using %s bus to access servo',
                  'session' if self.session_bus else 'system')
        # proxy to servo service
        self.servo = None

        if self.bus.name_has_owner(ServoTask.SERVO_DBUS_SERVICE):
            _log.debug('servo module already present, grab proxy')
            self._setup_servo_proxy()

        # we're monitoring servo name changes
        self.bus.watch_name_owner(ServoTask.SERVO_DBUS_SERVICE,
                                  self._servo_name_changed)


    def stop(self):
        """Stop task"""
        # get rid of reference to bus
        self.bus = None

        super(ServoTask, self).stop()

    def _servo_name_changed(self, name):
        """Callback for when a name owner of servo service has changed

        :param str name: service name, either actual servo name or empty
        """

        _log.debug('servo name changed: %s', name)

        if not name:
            _log.info('servo driver lost')
            self.servo = None
        else:
            self._setup_servo_proxy()

    def _setup_servo_proxy(self):
        """Setup proxy to servo service, call only whe name is resolvable"""
        if self.servo:
            return

        _log.debug('obtain proxy to servo')
        try:
            servo_obj = self.bus.get_object(ServoTask.SERVO_DBUS_SERVICE,
                                            ServoTask.SERVO_DBUS_PATH)
            self.servo = dbus.Interface(servo_obj,
                                        ServoTask.SERVO_DBUS_INTERFACE)
        except dbus.DBusException:
            _log.exception('failed to obtain proxy to servo')
        else:
            _log.debug('got proxy')
            # connect to value change signal
            self.servo.connect_to_signal('valueChanged', self._servo_value_changed)

    def _servo_value_changed(self, parameter, motor, limit, in_progress, value):
        """pl.ros3d.servo.valueChanged signal handler"""
        _log.debug('got signal for parameter %s, value: %d', parameter, value)
        ParametersStore.set(parameter, value)
        _log.debug('parameter value updated')

    def change_param(self, param, value):
        """Attempt to set a parameter is servo.<commit_msg>servo: update parameter status based on servo's availability<commit_after>#
# Copyright (c) 2015, Open-RnD Sp. z o.o.  All rights reserved.
#
"""Servo driver proxy"""
from __future__ import absolute_import

from sparts.tasks.dbus import DBusTask
from sparts.sparts import option
from ros3ddevcontroller.param.store import ParametersStore, SERVO_PARAMETERS
from ros3ddevcontroller.param.parameter import ParameterStatus
import glib
import logging
import dbus

_log = logging.getLogger(__name__)


class ParamApplyRequest(object):
    """Wrapper for keeping a parameter apply request in check

    :ivar param str: parameter name
    :ivar value: parameter value"""
    def __init__(self, param, value):
        self.param = param
        self.value = value


class ParamApplyError(Exception):
    """Error wrapper class"""
    pass


class ServoTask(DBusTask):
    """Servo driver proxy. The proxy will automatically find a DBus servo
    service and connect to it.

    Since the task is running in a Glib MainLoop possibly on a
    different thread than the caller of `change_param()` method, a
    trampoline is used. The request is wrapped into
    `ParamApplyRequest` and `change_param()` returns a `Future()`
    object.
    """
    OPT_PREFIX = 'servo-dbus'

    SERVO_DBUS_SERVICE = 'pl.ros3d.servo'
    SERVO_DBUS_PATH = '/pl/ros3d/servo'
    SERVO_DBUS_INTERFACE = "pl.ros3d.servo"
    SERVO_CALL_TIMEOUT_S = 120 # seconds

    session_bus = option(action='store_true', type=bool,
                         default=False,
                         help='Use session bus to access servo service')

    def __init__(self, *args, **kwargs):
        super(ServoTask, self).__init__(*args, **kwargs)

        self.bus = None
        self.servo = None

    def start(self):
        self.asyncRun(self._start_servo)
        super(ServoTask, self).start()

    def _start_servo(self):
        """Start task. Perform necessary setup:
        - bus connection
        - setup servo service name monitoring
        - obtain proxy to servo if possible
        """
        _log.debug('starting servo task for service: %s', self.service.name)
        # get bus
        if self.session_bus:
            self.bus = dbus.SessionBus(private=True)
        else:
            self.bus = dbus.SystemBus(private=True)

        _log.info('using %s bus to access servo',
                  'session' if self.session_bus else 'system')
        # proxy to servo service
        self.servo = None

        if self.bus.name_has_owner(ServoTask.SERVO_DBUS_SERVICE):
            _log.debug('servo module already present, grab proxy')
            self._setup_servo_proxy()

        # we're monitoring servo name changes
        self.bus.watch_name_owner(ServoTask.SERVO_DBUS_SERVICE,
                                  self._servo_name_changed)


    def stop(self):
        """Stop task"""
        # get rid of reference to bus
        self.bus = None

        super(ServoTask, self).stop()

    def _servo_name_changed(self, name):
        """Callback for when a name owner of servo service has changed

        :param str name: service name, either actual servo name or empty
        """

        _log.debug('servo name changed: %s', name)

        if not name:
            _log.info('servo driver lost')
            self.servo = None
            pstatus = ParameterStatus.SOFTWARE
        else:
            self._setup_servo_proxy()
            pstatus = ParameterStatus.HARDWARE

        self._set_servo_params_status(pstatus)

    @classmethod
    def _set_servo_params_status(cls, pstatus):
        for pname in SERVO_PARAMETERS:
            pdesc = ParametersStore.get(pname)
            pdesc_status = pdesc.status
            pdesc_status.set_status(pstatus)
            ParametersStore.set_status(pname, pdesc_status)

    def _setup_servo_proxy(self):
        """Setup proxy to servo service, call only whe name is resolvable"""
        if self.servo:
            return

        _log.debug('obtain proxy to servo')
        try:
            servo_obj = self.bus.get_object(ServoTask.SERVO_DBUS_SERVICE,
                                            ServoTask.SERVO_DBUS_PATH)
            self.servo = dbus.Interface(servo_obj,
                                        ServoTask.SERVO_DBUS_INTERFACE)
        except dbus.DBusException:
            _log.exception('failed to obtain proxy to servo')
        else:
            _log.debug('got proxy')
            # connect to value change signal
            self.servo.connect_to_signal('valueChanged', self._servo_value_changed)

    def _servo_value_changed(self, parameter, motor, limit, in_progress, value):
        """pl.ros3d.servo.valueChanged signal handler"""
        _log.debug('got signal for parameter %s, value: %d', parameter, value)
        ParametersStore.set(parameter, value)
        _log.debug('parameter value updated')

    def change_param(self, param, value):
        """Attempt to set a parameter is servo.
{'fraction_non_alphanumeric': 0.08674724569472671, 'fraction_numerical': 0.002674082789603166, 'mean_word_length': 2.3536585365853657, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '795402', 'n_tokens_mistral': 2734, 'n_tokens_neox': 2629, 'n_words': 765}