Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Raspoid - autodetect script (receiver side).
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socket
from threading import Thread
import sys
import time
import json
## logs levels
FORCED = 0 # Very few logs - critically needed
INFO = 1 # Few logs - all relevant
DEBUG = 2 # Many logs
class bcolors:
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
class CLI(Thread):
def __init__(self, name = "cli"):
Thread.__init__(self)
self.name = name
self.terminated = False
def run(self):
global last_received_message
available_network_commands = ["search"]
available_cli_commands = ["stop"]
log("Thread started properly", INFO, self.name)
while not self.terminated:
new_command = raw_input("Enter a new command:\n")
if new_command in available_network_commands:
last_received_message = new_command
elif new_command in available_cli_commands:
if new_command == "stop":
for thread in threads:
if thread.name != "cli":
thread.stop()
self.stop()
else:
log("Unrecognized command", INFO, self.name)
time.sleep(1.0)
log("Thread ended properly", INFO, self.name)
def stop(self):
self.terminated = True
class Network(Thread):
def __init__(self, name = "network"):
Thread.__init__(self)
self.name = name
self.terminated = False
def run(self):
global last_received_message
log("Thread started properly", INFO, self.name)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('', 4242))
#log("New BrickPi detected - Connection accepted ", INFO, self.name)# + str(infos_connection_with_master), INFO, self.name)
robots_ip = set()
while not self.terminated:
data, addr = sock.recvfrom(1024)
data = json.loads(data)
idd = str(data['id'])
if(idd == "robot_detected"):
ip_addresses = str(data['ip_addresses'])
if(ip_addresses not in robots_ip):
print "New robot detected"
robots_ip.add(ip_addresses)
hostname = str(data['hostname'])
print "-IP Address: " + ip_addresses
print "-Hostname: " + hostname
#log("New packet received: " + str(data) + " " + str(addr), INFO, self.name)
log("Thread ended properly", INFO, self.name)
def stop(self):
self.terminated = True
selected_log_level = INFO
def log(arg, log_level = INFO, thread_name = None):
if log_level <= selected_log_level:
if thread_name == None:
print arg
else:
color = None
if thread_name == "cli":
color = bcolors.OKBLUE
elif thread_name == "Raspoid":
color = bcolors.OKGREEN
elif thread_name == "network":
color = bcolors.WARNING
if color == None:
print "(%s) %s" % (thread_name, arg)#(print "(" + thread_name + ") " + arg
else:
print "%s(%s) %s %s" % (color, thread_name, arg, bcolors.ENDC)#color + "(" + thread_name + ") " + arg + bcolors.ENDC
threads = [Network(), CLI()]
for thread in threads:
thread.start()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.