You've already forked MicroPythonOS
mirror of
https://github.com/m5stack/MicroPythonOS.git
synced 2026-05-20 11:51:27 -07:00
WiFiConfig: use mpos.config facilities
This commit is contained in:
@@ -6,8 +6,9 @@ import _thread
|
||||
|
||||
from mpos.apps import Activity, Intent
|
||||
import mpos.ui
|
||||
import mpos.config
|
||||
|
||||
# Global variables because they're shared between activities:
|
||||
# Global variables because they're used by multiple Activities:
|
||||
access_points={}
|
||||
last_tried_ssid = ""
|
||||
last_tried_result = ""
|
||||
@@ -61,7 +62,9 @@ class WiFiConfig(Activity):
|
||||
self.start_scan_networks()
|
||||
|
||||
def onResume(self, screen):
|
||||
load_config()
|
||||
global access_points
|
||||
access_points = mpos.config.SharedPreferences("com.micropythonos.wificonf").get_dict("access_points")
|
||||
|
||||
|
||||
def show_error(self, message):
|
||||
print(f"show_error: Displaying error: {message}")
|
||||
@@ -103,7 +106,7 @@ class WiFiConfig(Activity):
|
||||
self.scan_button_label.set_text(self.scan_button_scanning_text)
|
||||
_thread.stack_size(mpos.apps.good_stack_size())
|
||||
_thread.start_new_thread(self.scan_networks_thread, ())
|
||||
|
||||
|
||||
def refresh_list(self):
|
||||
print("refresh_list: Clearing current list")
|
||||
self.aplist.clean() # this causes an issue with lost taps if an ssid is clicked that has been removed
|
||||
@@ -203,6 +206,7 @@ class PasswordPage(Activity):
|
||||
connect_button=None
|
||||
cancel_button=None
|
||||
|
||||
|
||||
def onCreate(self):
|
||||
self.selected_ssid = self.getIntent().extras.get("selected_ssid")
|
||||
print("PasswordPage: Creating new password page")
|
||||
@@ -218,11 +222,9 @@ class PasswordPage(Activity):
|
||||
self.password_ta.set_one_line(True)
|
||||
self.password_ta.align_to(label, lv.ALIGN.OUT_BOTTOM_MID, 5, 0)
|
||||
self.password_ta.add_event_cb(self.password_ta_cb,lv.EVENT.CLICKED,None)
|
||||
# try to find saved password:
|
||||
for apssid,password in access_points.items():
|
||||
if self.selected_ssid == apssid:
|
||||
self.password_ta.set_text(password)
|
||||
break
|
||||
pwd = self.findSavedPassword(self.selected_ssid)
|
||||
if pwd:
|
||||
self.password_ta.set_text(pwd)
|
||||
self.password_ta.set_placeholder_text("Password")
|
||||
print("PasswordPage: Creating keyboard (hidden by default)")
|
||||
self.keyboard=lv.keyboard(password_page)
|
||||
@@ -291,9 +293,11 @@ class PasswordPage(Activity):
|
||||
print("connect_cb: Connect button clicked")
|
||||
password=self.password_ta.get_text()
|
||||
print(f"connect_cb: Got password: {password}")
|
||||
access_points[self.selected_ssid]=password
|
||||
self.setPassword(self.selected_ssid, password)
|
||||
print(f"connect_cb: Updated access_points: {access_points}")
|
||||
save_config()
|
||||
editor = mpos.config.SharedPreferences("com.micropythonos.wificonf").edit()
|
||||
editor.put_dict("access_points", access_points)
|
||||
editor.commit()
|
||||
self.setResult(True, {"ssid": self.selected_ssid, "password": password})
|
||||
print("connect_cb: Restoring main_screen")
|
||||
self.finish()
|
||||
@@ -302,47 +306,21 @@ class PasswordPage(Activity):
|
||||
print("cancel_cb: Cancel button clicked")
|
||||
self.finish()
|
||||
|
||||
@staticmethod
|
||||
def setPassword(ssid, password):
|
||||
global access_points
|
||||
ap = access_points.get(ssid)
|
||||
if ap:
|
||||
ap["password"] = password
|
||||
return
|
||||
# if not found, then add it:
|
||||
access_points[ssid] = { "password": password }
|
||||
|
||||
|
||||
|
||||
# Non-class functions:
|
||||
|
||||
|
||||
def load_config(): # Maybe this should call a framework function
|
||||
print("load_config: Checking for /data directory")
|
||||
try:
|
||||
os.stat('data')
|
||||
print("load_config: /data exists")
|
||||
except OSError:
|
||||
print("load_config: Creating /data directory")
|
||||
os.mkdir('data')
|
||||
print("load_config: Checking for /data/com.example.wificonf directory")
|
||||
try:
|
||||
os.stat('data/com.example.wificonf')
|
||||
print("load_config: /data/com.example.wificonf exists")
|
||||
except OSError:
|
||||
print("load_config: Creating /data/com.example.wificonf directory")
|
||||
os.mkdir('data/com.example.wificonf')
|
||||
print("load_config: Loading config from conf.json")
|
||||
try:
|
||||
with open('data/com.example.wificonf/conf.json','r') as f:
|
||||
global access_points
|
||||
access_points=ujson.load(f)
|
||||
print(f"load_config: Loaded access_points: {access_points}")
|
||||
except OSError:
|
||||
access_points={}
|
||||
print("load_config: No config file found, using empty access_points")
|
||||
|
||||
|
||||
def save_config(): # Maybe this should call a framework function
|
||||
print("save_config: Saving access_points to conf.json")
|
||||
try:
|
||||
with open('data/com.example.wificonf/conf.json','w') as f:
|
||||
ujson.dump(access_points,f)
|
||||
print(f"save_config: Saved access_points: {access_points}")
|
||||
except OSError:
|
||||
self.show_error("Failed to save config")
|
||||
print("save_config: Failed to save config")
|
||||
|
||||
|
||||
|
||||
@staticmethod
|
||||
def findSavedPassword(ssid):
|
||||
if not access_points:
|
||||
return None
|
||||
ap = access_points.get(ssid)
|
||||
if ap:
|
||||
return ap.get("password")
|
||||
return None
|
||||
|
||||
@@ -8,23 +8,19 @@ class SharedPreferences:
|
||||
self.filename = filename
|
||||
self.filepath = f"data/{self.appname}/{self.filename}"
|
||||
self.data = {}
|
||||
self.load() # Load existing preferences
|
||||
self.load()
|
||||
|
||||
def make_folder_structure(self):
|
||||
"""Create directory structure if it doesn't exist."""
|
||||
#print("Checking if data/ exists")
|
||||
try:
|
||||
os.stat('data')
|
||||
#print("data/ exists")
|
||||
except OSError:
|
||||
print("Creating data/ directory")
|
||||
os.mkdir('data')
|
||||
#print(f"Checking if data/{self.appname}/ exists")
|
||||
try:
|
||||
os.stat(f"data/{self.appname}")
|
||||
#print(f"data/{self.appname}/ exists")
|
||||
except OSError:
|
||||
#print(f"Creating data/{self.appname} directory")
|
||||
print(f"Creating data/{self.appname} directory")
|
||||
os.mkdir(f"data/{self.appname}")
|
||||
|
||||
def load(self):
|
||||
@@ -35,7 +31,7 @@ class SharedPreferences:
|
||||
print(f"load: Loaded preferences: {self.data}")
|
||||
except Exception as e:
|
||||
print(f"load: Got exception {e}, assuming empty or non-existent preferences.")
|
||||
self.data = {} # Default to empty dict on error
|
||||
self.data = {}
|
||||
|
||||
def get_string(self, key, default=None):
|
||||
"""Retrieve a string value for the given key, with a default if not found."""
|
||||
@@ -55,13 +51,21 @@ class SharedPreferences:
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
def get_list(self, key, default=None):
|
||||
"""Retrieve a list for the given key, with a default if not found."""
|
||||
return self.data.get(key, default if default is not None else [])
|
||||
|
||||
def get_dict(self, key, default=None):
|
||||
"""Retrieve a dictionary for the given key, with a default if not found."""
|
||||
return self.data.get(key, default if default is not None else {})
|
||||
|
||||
def edit(self):
|
||||
"""Return an Editor object to modify preferences."""
|
||||
return Editor(self)
|
||||
|
||||
def save_config(self):
|
||||
"""Save preferences to the JSON file."""
|
||||
self.make_folder_structure() # Ensure directories exist
|
||||
self.make_folder_structure()
|
||||
print(f"save_config: Saving preferences to {self.filepath}")
|
||||
try:
|
||||
with open(self.filepath, 'w') as f:
|
||||
@@ -70,11 +74,48 @@ class SharedPreferences:
|
||||
except Exception as e:
|
||||
print(f"save_config: Got exception {e}")
|
||||
|
||||
# Methods for list-based structures
|
||||
def get_list_item(self, list_key, index, item_key, default=None):
|
||||
"""Retrieve a specific item's value from a list of dictionaries."""
|
||||
try:
|
||||
return self.data.get(list_key, [])[index].get(item_key, default)
|
||||
except (IndexError, KeyError, TypeError):
|
||||
return default
|
||||
|
||||
def get_list_item_dict(self, list_key, index, default=None):
|
||||
"""Retrieve an entire dictionary from a list at the specified index."""
|
||||
try:
|
||||
return self.data.get(list_key, [])[index]
|
||||
except (IndexError, TypeError):
|
||||
return default if default is not None else {}
|
||||
|
||||
# Generic methods for dictionary-based structures
|
||||
def get_dict_item_field(self, dict_key, item_key, field, default=None):
|
||||
"""Retrieve a specific field for an item in a dictionary by item_key."""
|
||||
try:
|
||||
return self.data.get(dict_key, {}).get(item_key, {}).get(field, default)
|
||||
except (KeyError, TypeError):
|
||||
return default
|
||||
|
||||
def get_dict_item(self, dict_key, item_key, default=None):
|
||||
"""Retrieve the entire configuration for an item in a dictionary by item_key."""
|
||||
try:
|
||||
return self.data.get(dict_key, {}).get(item_key, default if default is not None else {})
|
||||
except (KeyError, TypeError):
|
||||
return default if default is not None else {}
|
||||
|
||||
def get_dict_keys(self, dict_key):
|
||||
"""Retrieve a list of all keys in a dictionary at dict_key."""
|
||||
try:
|
||||
return list(self.data.get(dict_key, {}).keys())
|
||||
except (KeyError, TypeError):
|
||||
return []
|
||||
|
||||
class Editor:
|
||||
def __init__(self, preferences):
|
||||
"""Initialize Editor with a reference to SharedPreferences."""
|
||||
self.preferences = preferences
|
||||
self.temp_data = preferences.data.copy() # Work on a copy of the data
|
||||
self.temp_data = preferences.data.copy()
|
||||
|
||||
def put_string(self, key, value):
|
||||
"""Store a string value."""
|
||||
@@ -91,6 +132,64 @@ class Editor:
|
||||
self.temp_data[key] = bool(value)
|
||||
return self
|
||||
|
||||
def put_list(self, key, value):
|
||||
"""Store a list value."""
|
||||
if isinstance(value, list):
|
||||
self.temp_data[key] = value
|
||||
return self
|
||||
|
||||
def put_dict(self, key, value):
|
||||
"""Store a dictionary value."""
|
||||
if isinstance(value, dict):
|
||||
self.temp_data[key] = value
|
||||
return self
|
||||
|
||||
def append_to_list(self, list_key, item):
|
||||
"""Append a dictionary to a list in the preferences."""
|
||||
if list_key not in self.temp_data:
|
||||
self.temp_data[list_key] = []
|
||||
if isinstance(item, dict):
|
||||
self.temp_data[list_key].append(item)
|
||||
return self
|
||||
|
||||
def update_list_item(self, list_key, index, item):
|
||||
"""Update a dictionary at a specific index in a list."""
|
||||
try:
|
||||
if list_key in self.temp_data and isinstance(self.temp_data[list_key], list):
|
||||
if index < len(self.temp_data[list_key]) and isinstance(item, dict):
|
||||
self.temp_data[list_key][index] = item
|
||||
except (IndexError, TypeError):
|
||||
pass
|
||||
return self
|
||||
|
||||
def remove_from_list(self, list_key, index):
|
||||
"""Remove an item from a list at the specified index."""
|
||||
try:
|
||||
if list_key in self.temp_data and isinstance(self.temp_data[list_key], list):
|
||||
if index < len(self.temp_data[list_key]):
|
||||
self.temp_data[list_key].pop(index)
|
||||
except (IndexError, TypeError):
|
||||
pass
|
||||
return self
|
||||
|
||||
# Generic methods for dictionary-based structures
|
||||
def put_dict_item(self, dict_key, item_key, config):
|
||||
"""Add or update an item in a dictionary by item_key."""
|
||||
if dict_key not in self.temp_data:
|
||||
self.temp_data[dict_key] = {}
|
||||
if isinstance(config, dict):
|
||||
self.temp_data[dict_key][item_key] = config
|
||||
return self
|
||||
|
||||
def remove_dict_item(self, dict_key, item_key):
|
||||
"""Remove an item from a dictionary by item_key."""
|
||||
try:
|
||||
if dict_key in self.temp_data and isinstance(self.temp_data[dict_key], dict):
|
||||
self.temp_data[dict_key].pop(item_key, None)
|
||||
except (KeyError, TypeError):
|
||||
pass
|
||||
return self
|
||||
|
||||
def apply(self):
|
||||
"""Save changes to the file asynchronously (emulated)."""
|
||||
self.preferences.data = self.temp_data.copy()
|
||||
@@ -102,37 +201,78 @@ class Editor:
|
||||
self.preferences.save_config()
|
||||
return True
|
||||
|
||||
# Example usage
|
||||
# Example usage with access_points as a dictionary
|
||||
def main():
|
||||
# Initialize SharedPreferences with an appname
|
||||
# Initialize SharedPreferences
|
||||
prefs = SharedPreferences("com.example.test_shared_prefs")
|
||||
|
||||
print("Getting preferences:")
|
||||
print(f"theme: {prefs.get_string('theme')}")
|
||||
print(f"volume: {prefs.get_int('volume')}")
|
||||
print(f"notifications: {prefs.get_bool('notifications')}")
|
||||
print("Done getting preferences.")
|
||||
|
||||
# Save some settings using Editor
|
||||
# Save some simple settings and a dictionary-based access_points
|
||||
editor = prefs.edit()
|
||||
editor.put_string("theme", "dark")
|
||||
editor.put_int("volume", 75)
|
||||
editor.put_bool("notifications", True)
|
||||
editor.apply() # Asynchronous save (emulated)
|
||||
editor.put_string("someconfig", "somevalue")
|
||||
editor.put_int("othervalue", 54321)
|
||||
editor.put_dict("access_points", {
|
||||
"example_ssid1": {"password": "examplepass1", "detail": "yes please", "numericalconf": 1234},
|
||||
"example_ssid2": {"password": "examplepass2", "detail": "no please", "numericalconf": 9875}
|
||||
})
|
||||
editor.apply()
|
||||
|
||||
# Read back the settings
|
||||
print("Theme:", prefs.get_string("theme", "light"))
|
||||
print("Volume:", prefs.get_int("volume", 50))
|
||||
print("Notifications:", prefs.get_bool("notifications", False))
|
||||
print("Simple settings:")
|
||||
print("someconfig:", prefs.get_string("someconfig", "default_value"))
|
||||
print("othervalue:", prefs.get_int("othervalue", 0))
|
||||
|
||||
# Modify a setting
|
||||
print("\nAccess points (dictionary-based):")
|
||||
ssids = prefs.get_dict_keys("access_points")
|
||||
for ssid in ssids:
|
||||
print(f"Access Point SSID: {ssid}")
|
||||
print(f" Password: {prefs.get_dict_item_field('access_points', ssid, 'password', 'N/A')}")
|
||||
print(f" Detail: {prefs.get_dict_item_field('access_points', ssid, 'detail', 'N/A')}")
|
||||
print(f" Numerical Conf: {prefs.get_dict_item_field('access_points', ssid, 'numericalconf', 0)}")
|
||||
print(f" Full config: {prefs.get_dict_item('access_points', ssid)}")
|
||||
|
||||
# Add a new access point
|
||||
editor = prefs.edit()
|
||||
editor.put_string("theme", "light")
|
||||
success = editor.commit() # Synchronous save
|
||||
print("Save successful:", success)
|
||||
editor.put_dict_item("access_points", "example_ssid3", {
|
||||
"password": "examplepass3",
|
||||
"detail": "maybe",
|
||||
"numericalconf": 5555
|
||||
})
|
||||
editor.commit()
|
||||
|
||||
# Read updated setting
|
||||
print("Updated Theme:", prefs.get_string("theme", "light"))
|
||||
# Update an existing access point
|
||||
editor = prefs.edit()
|
||||
editor.put_dict_item("access_points", "example_ssid1", {
|
||||
"password": "newpass1",
|
||||
"detail": "updated please",
|
||||
"numericalconf": 4321
|
||||
})
|
||||
editor.commit()
|
||||
|
||||
# Remove an access point
|
||||
editor = prefs.edit()
|
||||
editor.remove_dict_item("access_points", "example_ssid2")
|
||||
editor.commit()
|
||||
|
||||
# Read updated access points
|
||||
print("\nUpdated access points (dictionary-based):")
|
||||
ssids = prefs.get_dict_keys("access_points")
|
||||
for ssid in ssids:
|
||||
print(f"Access Point SSID: {ssid}: {prefs.get_dict_item('access_points', ssid)}")
|
||||
|
||||
# Demonstrate compatibility with list-based configs
|
||||
editor = prefs.edit()
|
||||
editor.put_list("somelist", [
|
||||
{"a": "ok", "numericalconf": 1111},
|
||||
{"a": "not ok", "numericalconf": 2222}
|
||||
])
|
||||
editor.apply()
|
||||
|
||||
print("\List-based config:")
|
||||
somelist = prefs.get_list("somelist")
|
||||
for i, ap in enumerate(somelist):
|
||||
print(f"List item {i}:")
|
||||
print(f" a: {prefs.get_list_item('somelist', i, 'a', 'N/A')}")
|
||||
print(f" Full dict: {prefs.get_list_item_dict('somelist', i)}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user