# copilot_console_json8a.py # # Generated by copilot using prompts: # # Write a python script using nicegui to display an editable dict that # is stored in nicegui general persistent storage. Add a second endpoint # for a client to push updates to the dict using plain json string, using # http get. The dict will have a fixed set of keys. # # Now allow the numerical values in the dict to be floats, and use a # checkbox for the 'run' boolean element. # # Modify the nicegui script you wrote to add another endpoint that # returns the complete dict as a simple json string. # # RLS 20250422 # Provide a web-based console for controlling # the 3GBP system for GC demo. # # Endpoint '/' is the console # Endpoint '/sp/raw' is the setpoints as a json string # Endpoint '/pv/update' updates process values from a json string # Endpoint '/pv/raw' returns process values as json string from nicegui import ui, app import json from fastapi import Response, Query import base64 import urllib.parse import asyncio asyncio.set_event_loop(asyncio.new_event_loop()) # Initialize a dictionary with float values and a boolean checkbox def initialize_dict(): app.storage.general.enable_backup = False app.storage.general.backup_interval = None if hasattr(app.storage.general, 'backup'): del app.storage.general.backup # if 'sp_store' not in app.storage.general: app.storage.general['sp_store'] = { 'te': 0.0, 'tb': 0.0, 'rate': 0.0, 'vel': 0.0, 'hv': 0.0, 'zoff': 0.0, 'te_p': 0.0, 'te_i': 0.0, 'te_d': 0.0, 'rate_scaling': 96.75, 'run': False, 'tune_te': False, 'tune_tb': False, } if 'pv_store' not in app.storage.general: app.storage.general['pv_store'] = { 'te': 0.0, 'tb': 0.0, 'te_p': 0.0, 'te_i': 0.0, 'te_d': 0.0, 'tb_p': 0.0, 'tb_i': 0.0, 'tb_d': 0.0, 'ball': False, 'tune_te_done': False, } initialize_dict() # Ensure storage is initialized at app startup @ui.page('/') def main_page(): stored_dict = app.storage.general['sp_store'] ui.label('3GBP Console').classes('text-2xl font-bold mb-4') input_fields = {} checkbox1 = ui.checkbox('Run', value=stored_dict['run']) # Checkbox for boolean field checkbox2 = ui.checkbox('Hotend Autotune', value=stored_dict['tune_te']) # Checkbox for boolean field def update_values(): for key, input_field in input_fields.items(): stored_dict[key] = float(input_field.value) # Convert to float stored_dict['run'] = checkbox1.value # Update boolean value stored_dict['tune_te'] = checkbox2.value # Update boolean value app.storage.general['sp_store'] = stored_dict ui.notify('Values updated successfully!') with ui.card(): for key, value in stored_dict.items(): if key != 'run': # Numeric fields as inputs input_fields[key] = ui.input(f'{key}', value=str(value)).props('type=number') ui.button('Save Changes', on_click=update_values).classes('bg-blue-500 text-white mt-4') BASE64_CHARS ="ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" def base64_decode(data): """Decode Base64 string back to bytes (MicroPython-compatible).""" buffer = bytearray() data = data.rstrip("=") # Remove padding i = 0 while i < len(data): byte1 = BASE64_CHARS.index(data[i]) << 2 byte2 = BASE64_CHARS.index(data[i+1]) >> 4 if i+1 < len(data) else 0 byte3 = (BASE64_CHARS.index(data[i+1]) << 4) & 0xF0 if i+1 < len(data) else 0 byte4 = BASE64_CHARS.index(data[i+2]) >> 2 if i+2 < len(data) else 0 byte5 = (BASE64_CHARS.index(data[i+2]) << 6) & 0xC0 if i+2 < len(data) else 0 byte6 = BASE64_CHARS.index(data[i+3]) if i+3 < len(data) else 0 i += 4 buffer.append(byte1 | byte2) if i-2 < len(data): buffer.append(byte3 | byte4) if i-1 < len(data): buffer.append(byte5 | byte6) return bytes(buffer) @app.get('/pv/update') def update_dict_json(data: str = Query(None)): try: # Decode base64 JSON string # json_data = base64.b64decode(data).decode('utf-8') json_data = base64_decode(data) update_data = json.loads(json_data) print("json_data=", json_data) print("update_data=", update_data) if not isinstance(update_data, dict): return Response(content=json.dumps({"error": "Data must be a JSON object"}), media_type="application/json") stored_dict = app.storage.general.get('data_store', {}) updated_keys = [] for key, value in update_data.items(): if key in stored_dict: stored_dict[key] = bool(value) if key == 'run' else float(value) # Convert appropriately updated_keys.append(key) app.storage.general['data_store'] = stored_dict return Response(content=json.dumps({"status": "success", "updated_keys": updated_keys, "current_data": stored_dict}), media_type="application/json") except json.JSONDecodeError as e: return Response(content=json.dumps({"error": f"Invalid JSON format: {str(e)}"}), media_type="application/json") except Exception as e: return Response(content=json.dumps({"error": str(e)}), media_type="application/json") #def update_dict_json(data: str = Query(None), encoded: bool = True): # try: # json_data = base64.b64decode(data).decode('utf-8') if encoded else urllib.parse.unquote(data) # update_data = json.loads(json_data) # print("json_data=", json_data) # print("update_data=", update_data) # # if not isinstance(update_data, dict): # return {"error": "Data must be a JSON object"} # # stored_dict = app.storage.general.get('pv_store', {}) # updated_keys = [] # # for key, value in update_data.items(): # if key in stored_dict: # stored_dict[key] = bool(value) if key == 'ball' else float(value) # Handle float conversion & boolean # updated_keys.append(key) # # # turn off autotune if done # # app.storage.general['pv_store'] = stored_dict # # pv_dict = app.storage.general['pv_store'] # sp_dict = app.storage.general['sp_store'] # if (pv_dict["tune_te_done"] is True and sp_dict["tune_te"] is True): # app.storage.general['sp_store']["tune_te"] = False # app.storage.general['sp_store']["te_p"] = pv_dict["te_p"] # app.storage.general['sp_store']["te_i"] = pv_dict["te_i"] # app.storage.general['sp_store']["te_d"] = pv_dict["te_d"] # # return {"status": "success", "updated_keys": updated_keys, "current_data": stored_dict} # # except json.JSONDecodeError as e: # return {"error": f"Invalid JSON format: {str(e)}"} # except Exception as e: # return {"error": str(e)} # sp raw endpoint #@app.get('/sp/raw') #def get_raw_sp_dict(): # return Response( # content=json.dumps(app.storage.general.get('sp_store', {})), # media_type="application/json" # ) # pv raw endpoint #@app.get('/pv/raw') #def get_raw_pv_dict(): # return Response( # content=json.dumps(app.storage.general.get('pv_store', {})), # media_type="application/json" # ) #app.add_api_route('/sp/raw', lambda: Response(content=json.dumps(app.storage.general.get('sp_store', {})), media_type="application/json"), methods=['GET']) @app.get('/sp/raw') def get_raw_sp_dict(): sp_dict = app.storage.general.get('sp_store', {}) return Response(content=json.dumps(sp_dict), media_type="application/json") app.add_api_route('/pv/raw', lambda: Response(content=json.dumps(app.storage.general.get('pv_store', {})), media_type="application/json"), methods=['GET']) ui.run()