# robi_comm.py # # contains functions to: # communicate via web server to receive commands and send status information # parse received commands to update parameters and trigger actions # # # initial version 2-Jul-2022 by wolf2018 # 13-Sep22: removed code to test this module stand alone # 18-Jan23: removed unused code # 25-Feb23: add controler shutdown and exception management # # Remarks: # data is presented to functions using a dictionary # several print commands are commented out, un-comment for debugging as needed # # version date see version_date after imports # import picoweb import ujson import gc import cfg from walking_positions import calc_positions import sys version_date = "25-Feb-2023" # assign pointers to the global objects status_dict = cfg.status_dict w_parameter0 = cfg.w_parameter0 w_parameter1 = cfg.w_parameter1 # subclass picoweb's WebApp to add exception management # robot controller is shut down by raising an exception # class Web_App(picoweb.WebApp): def __init__(self, pkg, routes=None, serve_static=True): super().__init__(pkg, routes, serve_static) # function to handle exceptions in picoweb def handle_exc(self, req, resp, e): """exception handler function to replace picoweb exception handler""" if cfg.debug > 1: print("exception handler of subclass called") print("exception is: ", e) exception_string = str(e) if "shut down" in exception_string: print("closing last request .....") await resp.aclose() print(".......last request closed") # raise system exit exception to shut down the robot controller sys.exit() # instantiate the WebApp # app = picoweb.WebApp(__name__) app = Web_App(None) async def send_status(resp): """ send status back to caller""" gc.collect() robi_status = {} robi_status["w_status"] = cfg.status_dict["w_status"] robi_status["action"] = cfg.status_dict["action"] robi_status["battery"] = cfg.read_batt() robi_status["memory"] = gc.mem_free() robi_status["message"] = cfg.status_dict["message"] response = ujson.dumps(robi_status) if cfg.debug > 1: print("response: "+str(response)+" "+str(type(response))) await picoweb.start_response(resp, status="201") await resp.awrite(response) # define web routes @app.route("/") def index(req, resp): gc.collect() # un-comment print statements below as needed for debugging requests # determine action based on method and content method = req.method # check if request has content if cfg.debug > 1: print("headers is: "+str(req.headers)+"\n") if b"Content-Length" in req.headers: content_lenght = int(req.headers[b"Content-Length"]) else: content_lenght = 0 # print("received "+method+" request\n") # print("req has: "+str(dir(req))+"\n") # print("path is: "+req.path) # print("qs is: "+req.qs) # print("type of header is: "+str(type(req.headers))+"\n") # print("content lenghth is: "+str(l)+" type l is: "+str(type(l))+"\n") # determine action based on method of request if method == "GET": if content_lenght > 0: # GET request with content await req.read_form_data() if cfg.debug > 0: print("received data from "+method+" is: "+str(req.form)) # Get is used for life tick: toggle onboard LED cfg.pico_led.toggle() # serve GET request: send robot controller's status await send_status(resp) elif method == "POST": # walking parameters received if content_lenght == 0: # POST request without content # send "received POST request without data" html error await picoweb.http_error(resp, "400") # process request else: await req.read_form_data() # extract the request data from request form req_data = ujson.loads(list(req.form.keys())[0]) if cfg.debug > 0: print("Received data: "+str(req_data)+" req_data type is: " + str(type(req_data))) # store parameters data = ujson.loads(req_data) cfg.parameters["CoM_height"] = float(data["CoM"]) cfg.parameters["m_step"] = float(data["m_step"]) cfg.parameters["delay"] = int(float(data["delay"])) cfg.parameters["friction"] = float(data["friction"]) cfg.status_dict["message"] = "parameters updated" if cfg.debug > 0: print("parameters loaded: " + str(data)+"\n") # serve POST request await picoweb.start_response(resp) await resp.awrite(req_data) elif method == "PUT": # walking planning values received if content_lenght == 0: # PUT request without content # send "received Post without data" html error await picoweb.http_error(resp, "400") else: # process request await req.read_form_data() # extract the request data from request form req_data = ujson.loads(list(req.form.keys())[0]) if cfg.debug > 0: print("Received data: "+str(req_data)+" req_data type is: " + str(type(req_data))) # store walking planning values data = ujson.loads(req_data) cfg.w_parameter0["target"] = float(data["target"]) cfg.w_parameter0["step"] = float(data["step"]) cfg.w_parameter0["leading"] = data["leading"] cfg.status_dict["message"] = "walking planning values updated" if cfg.debug > 0: print("walking plannning data loaded: " + str(data)) await picoweb.start_response(resp) await resp.awrite(req_data) else: print("method received: "+method) if content_lenght > 0: # clear data cache await req.read_form_data() print("received data from "+method+" is: "+str(req.form)) # send "unsupported method" html error await picoweb.http_error(resp, "405") @app.route("/actions") def set_action(req, resp): gc.collect() method = req.method # check if request has content if b"Content-Length" in req.headers: content_lenght = int(req.headers[b"Content-Length"]) else: content_lenght = 0 # determine action based on method of request if method == "POST": if content_lenght == 0: # POST request without content # send "received Post without data" html error await picoweb.http_error(resp, "400") # process request else: await req.read_form_data() # extract the request received and request data from request form _req_rec = ujson.loads(list(req.form.keys())[0]) req_data = ujson.loads(_req_rec) if cfg.debug > 0: print("req_form is: "+str(req.form)) print("Received data: "+str(req_data)+" req_data type is: " + str(type(req_data))) # action on POST request # action "stop" requested? if req_data["action"] == "stop": if cfg.debug > 0: print("\n... performing a stop action...") # send shutdown status to WebUI cfg.status_dict["message"] = "robot controller shut down" cfg.status_dict["action"] = "stop" cfg.status_dict["w_status"] = "stop" await send_status(resp) # shutdown the robot controller now if cfg.debug > 0: print("\n... initiate robot controller shut down ...") # initiate shut down by raising an exception raise Exception("controller shut down requested") else: # normal action requested # action "initialize" requested if req_data["action"] == "init": if cfg.debug > 0: print("\n... performing initialization...") print("memory before thread:", gc.mem_free()) calc_positions("init", cfg.w_parameter0) # action "ready" requested, move to reqested CoM position if req_data["action"] == "ready": if cfg.debug > 0: print("\n... performing a ready action...") print("memory before thread:", gc.mem_free()) calc_positions("ready", cfg.w_parameter0) # action "walk" requested if req_data["action"] == "walk": if cfg.debug > 0: print("\n...robi_comm: action is walk...") print(cfg.w_parameter0) print(cfg.parameters, "\n\n") if cfg.debug > 1: input("press ENTER to start test walk") calc_positions("walk", cfg.w_parameter0) # action "stand" requested" if req_data["action"] == "stand": if cfg.debug > 0: print("\n... performing a stand action...") calc_positions("stand", cfg.w_parameter0) # serve POST request await send_status(resp) else: # invalid method received print("method received: "+method) if content_lenght > 0: # clear data cache await req.read_form_data() print("received data from "+method+" is: "+str(req.form)) # send "unsupported method" error await picoweb.http_error(resp, "405") # end of module