SkinPopulation.py

"""
   Copyright (C) 2021-2022 Torsten Fietzek; Helge Ülo Dinkelbach
 
   SkinPopulation.py is part of the ANNarchy iCub interface
 
   This program 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.
 
   The ANNarchy iCub interface 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 this headers. If not, see [http://www.gnu.org/licenses/].
 """
import ANN_iCub_Interface
from ANNarchy.core.Neuron import Neuron
try:
    from ANNarchy.intern.SpecificPopulation import SpecificPopulation
    from ANNarchy.intern.Messages import _error
 
except:
    from ANNarchy.core.SpecificPopulation import SpecificPopulation
    from ANNarchy.core.Global import _error
 
 
class SkinPopulation(SpecificPopulation):
    """
        ANNarchy population class to connect with the iCub skin.
        Readout the skin sensor data and set it as population activation.
    """
 
    def __init__(self, geometry=None, skin_section="", ip_address="0.0.0.0", port=50015, copied=False, name=None):
        """Init the SkinPopulation.
 
        Args:
            geometry (tuple, optional): ANNarchy population geometry. Defaults to None.
            skin_section (str, optional): Specify the respective skin section of the used arm. Defaults to "".
            ip_address (str, optional): ip-address of the gRPC connection. Need to fit with the respective kinematic reader module. Defaults to "0.0.0.0".
            port (int, optional): port of the gRPC connection. Need to fit with the respective kinematic reader module. Defaults to 50015.
            copied (bool, optional): ANNarchy specific parameter. Defaults to False.
            name (str, optional): individiual name for the population. Defaults to None.
        """
        SpecificPopulation.__init__(self, geometry=geometry, neuron=Neuron(equations="r = 0.0"), copied=copied, name=name)
 
        self._skin_section = skin_section
        self._ip_address = ip_address
        self._port = port
 
    def _init_attributes(self):
        SpecificPopulation._init_attributes(self)
 
        self.cyInstance.set_ip_address(self._ip_address)
        self.cyInstance.set_port(self._port)
 
    def _copy(self):
        return SkinPopulation(geometry=self.geometry, ip_address=self._ip_address, port=self._port, copied=True, name=self.name)
 
    @property
    def ip_address(self):
        if self.initialized:
            return self.cyInstance.get_ip_address()
        else:
            _error("Read-out IP address is only valid after compile().")
 
    @ip_address.setter
    def ip_address(self, value):
        if self.initialized:
            self.cyInstance.set_ip_address(value)
        else:
            _error("Changing IP address is only valid after compile() or constructor.")
 
    @property
    def port(self):
        if self.initialized:
            return self.cyInstance.get_port()
        else:
            _error("Read-out port is only valid after compile().")
 
    @port.setter
    def port(self, value):
        if self.initialized:
            self.cyInstance.set_port(value)
        else:
            _error("Changing port is only valid after compile() or constructor.")
 
    def _generate(self):
        """
            read out code for iCub through gRPC
        """
        include_paths = ANN_iCub_Interface.__path__[0]+"/grpc/"
 
        self._specific_template['include_additional'] = """// grpc stuff
#include "ANN_iCub_Interface/grpc/ProvideInputClient.h"
"""
        self._specific_template['declare_additional'] = """
    std::string ip_address;
    unsigned int port;
    ClientInstance* skin_source=nullptr;
        """
        self._specific_template['access_additional'] = """
    // Skin Source ip address
    void set_ip_address(std::string value) {
        ip_address = value;
    }
    std::string get_ip_address() {
        return ip_address;
    }
 
    // Skin Source port
    void set_port(unsigned int value) {
        port = value;
    }
    unsigned int get_port() {
        return port;
    }
 
    void connect() {
        if (skin_source == nullptr) {
            skin_source = new ClientInstance(ip_address, port);
        } else {
            std::cerr << "Population %(name)s already connected" << std::endl;
        }
    }
    """% {'name': self.name}
        self._specific_template['export_additional'] = """
        # IP Address and port
        void set_ip_address(string)
        string get_ip_address()
        void set_port(unsigned int)
        unsigned int get_port()
        void connect()
"""
        self._specific_template['wrapper_access_additional'] = """
    # IP Address and port
    def set_ip_address(self, value):
        pop%(id)s.set_ip_address(value.encode('utf-8'))
    def get_ip_address(self):
        return pop%(id)s.get_ip_address()
    def set_port(self, value):
        pop%(id)s.set_port(value)
    def get_port(self):
        return pop%(id)s.get_port()
    def connect(self):
        pop%(id)s.connect()
""" %{'id': self.id}
 
        if self._skin_section == "arm":
            self._specific_template['update_variables'] = """
        r  = skin_source->retrieve_skin_arm();
        """
        elif self._skin_section == "forearm":
            self._specific_template['update_variables'] = """
        r  = skin_source->retrieve_skin_forearm();
        """
        elif self._skin_section == "hand":
            self._specific_template['update_variables'] = """
        r  = skin_source->retrieve_skin_hand();
        """
 
    def _instantiate(self, cython_module):
        # initializes cython wrapper
        SpecificPopulation._instantiate(self, cython_module)
 
    def connect(self):
        """Connect the population to the gRPC socket. Need to be called once after compile."""
        # create socket and connect
        if self.initialized:
            self.cyInstance.connect()