Coverage for janitoo_roomba.roomba : 79%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
|
# -*- coding: utf-8 -*- It handle all communications to the Roomba vacuum
"""
This file is part of Janitoo.
Janitoo 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.
Janitoo 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 Janitoo. If not, see <http://www.gnu.org/licenses/>.
"""
# Set default logging handler to avoid "No handler found" warnings.
############################################################## #Check that we are in sync with the official command classes #Must be implemented for non-regression
##############################################################
else:
"clean":135, "dock":143, "full":132, "max":136, "play":141, "spot":143, "start":128, "spot":134, "LED_COMMAND":139, "SENSOR_COMMAND":142, "SENSOR_PACKET_0":0, "SENSOR_PACKET_1":1, "SENSOR_PACKET_2":2, "SENSOR_PACKET_3":3, "NUM_BYTES_PACKET_0":26, }
"Bumps Wheeldrops":0, "Wall":0, "Cliff Left":0, "Cliff Front Left":0, "Cliff Front Right":0, "Cliff Right":0, "Virtual Wall":0, "Motor Overcurrents":0, "Dirt Detector - Left":0, "Dirt Detector - Right":0, "Remote Opcode":0, "Buttons":0, "Distance":0, "Angle":0, "State":0, "Voltage":0, "Current":0, "Temperature":0, "Charge":0, "Capacity":0, "Battery-level":0, }
0:"Not charging", 1:"Charging Recovery", 2:"Charging", 3:"Trickle charging", 4:"Waiting", 5:"Charging Error", 6:"Cleaning", 7:"Docking", 8:"Maximun", }
"""The Roomba thread """
"""Build the bus """
"""This class abstracts a roowifi and gives attributes for telemetry data, as well as methods to command the robot """ """ """ product_name="Roomba Vacuum", product_type="Roowifi", product_manufacturer="IRobot", **kwargs)
node_uuid=self.uuid, help='The dock current', label='Current', units='mA', get_data_cb=self.get_current, )
node_uuid=self.uuid, help='The dock voltage', label='Voltage', units='V', get_data_cb=self.get_voltage, )
node_uuid=self.uuid, help='The current of the battery', label='Charge', units='mA', get_data_cb=self.get_battery_charge, )
node_uuid=self.uuid, help='The capacity of the battery', label='Capacity', units='mA', get_data_cb=self.get_battery_capacity, )
node_uuid=self.uuid, help='The charge of the battery', label='Percent', get_data_cb=self.get_battery_percent, )
node_uuid=self.uuid, help='Ping the vacuum', label='Ping', )
node_uuid=self.uuid, help='Username to connect the roomba', label='Username', )
node_uuid=self.uuid, help='Password to connect the roomba', label='Password', )
node_uuid=self.uuid, help='The temperature of the roomba', label='Temperature', get_data_cb=self.get_temperature, )
node_uuid=self.uuid, help='The state of the roomba dock', label='Dock', get_data_cb=self.get_state, )
node_uuid=self.uuid, help='The buttons on the roomba', label='Buttons', list_items=['clean', 'spot', 'dock', 'idle'], set_data_cb=self.set_button, is_writeonly = True, cmd_class = COMMAND_ROOMBA_VACUUM, )
"""Check that the component is 'available'
""" #~ print "it's me %s : %s" % (self.values['upsname'].data, self._ups_stats_last)
"""Roomba method which fetches telemetry data about the robot.
Other values that can be implemented:
- sensors['Bumps Wheeldrops'] = self.j['response']['r0']['value'] - sensors['Wall'] = self.j['response']['r1']['value'] - sensors['Cliff Left'] = self.j['response']['r2']['value'] - sensors['Cliff Front Left'] = self.j['response']['r3']['value'] - sensors['Cliff Front Right'] = self.j['response']['r4']['value'] - sensors['Cliff Right'] = self.j['response']['r5']['value'] - sensors['Virtual Wall'] = self.j['response']['r6']['value'] - sensors['Motor Overcurrents'] = self.j['response']['r7']['value'] - sensors['Dirt Detector - Left'] = self.j['response']['r8']['value'] - sensors['Dirt Detector - Right'] = self.j['response']['r9']['value'] - sensors['Remote Opcode'] = self.j['response']['r10']['value'] - sensors['Buttons'] = self.j['response']['r11']['value'] - sensors['Distance'] = self.j['response']['r12']['value'] - sensors['Angle'] = self.j['response']['r13']['value'] - sensors['State'] = State[int(self.j['response']['r14']['value'])] - sensors['Voltage'] = self.j['response']['r15']['value'] - sensors['Current'] = self.j['response']['r16']['value'] - sensors['Temperature'] = self.j['response']['r17']['value'] - sensors['Charge'] = self.j['response']['r18']['value'] - sensors['Capacity'] = self.j['response']['r19']['value'] - sensors['battery-level'] = int(self.j['response']['r18']['value'])*100 / int (self.j['response']['r19']['value']) """ except ZeroDivisionError: self._battery = -1.0 except: logger.exception("[%s] - Exception in get_telemetry", self.__class__.__name__) self._telemetry_last = False finally:
"""Return the battery """
"""Return the battery """
"""Return the battery charge """
"""Return the temperture """
"""Return the current """
"""Return the voltage """
"""Return the dock state """ return states[self._state]
"""Return the dock state """ auth = (self.values['username'].data, self.values['password'].data) params = {} if data == "clean": elif data == "spot": params['exec'] = 5 elif data == "dock": params['exec'] = 6 elif data == "idle": params['exec'] = 1 r = requests.get('http://' + self.values['ip_ping_config'].data + '/rwr.cgi', params = params, auth=auth)
"""Other way to acces the roowifi using a simple tcp socket. A simple copy paste ... does not work """ self._log.info("Start processing clean Command on %s " % (device)) try: self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) time.sleep(1) self.s.connect((ip , port)) time.sleep(1) self.s.send(chr(132)) time.sleep(1) #print("Le code de la fonction est %s " % table[command]) self.s.send(chr(table[command])) time.sleep(1) self._log.info("%s command Success on %s" % (command,device)) if str(command) == "clean": sensors['State'] = State[int(6)] if str(command) == "dock" : sensors['State'] = State[int(7)] if str(command) == "max" : sensors['State'] = State[int(8)] return True except: self._log.error(" %s Command Failed on %s" % (command,device)) return False |