eric: comparison src/eric7/MicroPython/Devices/CircuitPythonDevices.py

-:21803aa6c3e2
+:32c8a5b57332
 """
 Module implementing the device interface class for CircuitPython boards.
 """
 import ast
+import binascii
 import json
 import os
 import shutil
 from PyQt6.QtCore import QUrl, pyqtSlot
 from PyQt6.QtNetwork import QNetworkReply, QNetworkRequest
 from PyQt6.QtWidgets import QMenu
 from eric7 import Globals, Preferences
+from eric7.EricGui.EricOverrideCursor import EricOverrideCursor
 from eric7.EricWidgets import EricFileDialog, EricMessageBox
 from eric7.EricWidgets.EricApplication import ericApp
 from eric7.SystemUtilities import FileSystemUtilities
 from ..MicroPythonWidget import HAS_QTCHART
 @param parent reference to the parent object
 @type QObject
 """
 super().__init__(microPythonWidget, deviceType, parent)
-self.submitMode = "paste"  # use 'paste' mode to avoid loosing state
+self._submitMode = "paste"  # use 'paste' mode to avoid loosing state
 self.__boardName = boardName
 self.__workspace = self.__findWorkspace()
 self.__updater = CircuitPythonUpdaterInterface(self)
 self.__createCPyMenu()
+self.__securityTranslations = {
+"OPEN": self.tr("open", "open WiFi network"),
+"WEP": "WEP",
+"WPA_PSK": "WPA",
+"WPA2_PSK": "WPA2",
+"WPA_WPA2_PSK": "WPA/WPA2",
+"WPA2_ENTERPRISE": "WPA2 (CCMP)",
+"WPA3_PSK": "WPA3",
+"WPA2_WPA3_PSK": "WPA2/WPA3",
+}
+self.__securityCode2AuthModeString = {
+0: "[wifi.AuthMode.OPEN]",
+1: "[wifi.AuthMode.WEP]",
+2: "[wifi.AuthMode.WPA, wifi.AuthMode.PSK]",
+3: "[wifi.AuthMode.WPA2, wifi.AuthMode.PSK]",
+4: "[wifi.AuthMode.WPA, wifi.AuthMode.WPA2, wifi.AuthMode.PSK]",
+5: "[wifi.AuthMode.WPA2, wifi.AuthMode.ENTERPRISE]",
+6: "[wifi.AuthMode.WPA3, wifi.AuthMode.PSK]",
+7: "[wifi.AuthMode.WPA2, wifi.AuthMode.WPA3, wifi.AuthMode.PSK]",
+}
 def setConnected(self, connected):
 """
 Public method to set the connection state.
 def __findWorkspace(self, silent=False):
 """
 Private method to find the workspace directory.
-@param silent flag indicating silent operations
+@param silent flag indicating silent operations (defaults to False)
-@type bool
+@type bool (optional)
 @return workspace directory used for saving files
 @rtype str
 """
 # Attempts to find the paths on the filesystem that represents the
 # plugged in CIRCUITPY boards.
 return False, ''
 print(has_wifi())
 del has_wifi
 """
-out, err = self._interface.execute(command, mode=self.submitMode)
+out, err = self._interface.execute(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
 def getWifiData(self):
 import wifi
 r = wifi.radio
 station = {
-'active': r.enabled and r.ipv4_address_ap is None,
+'active': r.enabled and r.ipv4_address is not None,
 'connected': r.ipv4_address is not None,
 'ifconfig': (
 str(r.ipv4_address) if r.ipv4_address else'0.0.0.0',
 str(r.ipv4_subnet) if r.ipv4_subnet else'0.0.0.0',
 str(r.ipv4_gateway) if r.ipv4_gateway else'0.0.0.0',
 str(r.ipv4_dns) if r.ipv4_dns else'0.0.0.0',
 ),
 'mac': binascii.hexlify(r.mac_address, ':').decode(),
-'txpower': r.tx_power,
 'hostname': r.hostname,
 }
+try:
+station['txpower'] = r.tx_power
+except AttributeError:
+pass
+try:
+if r.ap_info is not None:
+station.update({
+'ap_ssid': r.ap_info.ssid,
+'ap_bssid': binascii.hexlify(r.ap_info.bssid, ':'),
+'ap_rssi': r.ap_info.rssi,
+'ap_channel': r.ap_info.channel,
+'ap_country': r.ap_info.country,
+})
+authmode = r.ap_info.authmode
+station['ap_security'] = (
+'_'.join(str(x).split('.')[-1] for x in authmode)
+if isinstance(authmode, list)
+else authmode
+)
+except (NotImplementedError, AttributeError):
+pass
 print(json.dumps(station))
 ap = {
 'active': r.enabled and r.ipv4_address_ap is not None,
 'connected': r.ipv4_address_ap is not None,
 str(r.ipv4_subnet_ap) if r.ipv4_subnet_ap else'0.0.0.0',
 str(r.ipv4_gateway_ap) if r.ipv4_gateway_ap else'0.0.0.0',
 str(r.ipv4_dns) if r.ipv4_dns else'0.0.0.0',
 ),
 'mac': binascii.hexlify(r.mac_address_ap, ':').decode(),
-'txpower': r.tx_power,
 'hostname': r.hostname,
 }
+try:
+ap['txpower'] = r.tx_power
+except AttributeError:
+pass
 print(json.dumps(ap))
 overall = {
 'active': r.enabled
 }
 wifi_status()
 del wifi_status
 """
-out, err = self._interface.execute(command, mode=self.submitMode)
+out, err = self._interface.execute(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 stationStr, apStr, overallStr = out.decode("utf-8").splitlines()
 station = json.loads(stationStr)
 ap = json.loads(apStr)
 overall = json.loads(overallStr)
+if "ap_security" in station:
+try:
+station["ap_security"] = self.__securityTranslations[
+station["ap_security"]
+]
+except KeyError:
+station["ap_security"] = self.tr("unknown ({0})").format(
+station["ap_security"]
+)
 return station, ap, overall
+def connectWifi(self, ssid, password):
+"""
+Public method to connect a device to a WiFi network.
+@param ssid name (SSID) of the WiFi network
+@type str
+@param password password needed to connect
+@type str
+@return tuple containing the connection status and an error string
+@rtype tuple of (bool, str)
+"""
+command = """
+def connect_wifi(ssid, password):
+import json
+import wifi
+r = wifi.radio
+try:
+r.start_station()
+r.connect(ssid, password)
+status = 'connected'
+except Exception as exc:
+status = str(exc)
+print(json.dumps({{'connected': r.ipv4_address is not None, 'status': status}}))
+connect_wifi({0}, {1})
+del connect_wifi
+""".format(
+repr(ssid),
+repr(password if password else ""),
+)
+with EricOverrideCursor():
+out, err = self._interface.execute(
+command, mode=self._submitMode, timeout=15000
+)
+if err:
+return False, err
+result = json.loads(out.decode("utf-8").strip())
+error = "" if result["connected"] else result["status"]
+return result["connected"], error
+def disconnectWifi(self):
+"""
+Public method to disconnect a device from the WiFi network.
+@return tuple containing a flag indicating success and an error string
+@rtype tuple of (bool, str)
+"""
+command = """
+def disconnect_wifi():
+import json
+import wifi
+r = wifi.radio
+try:
+r.stop_station()
+status = ''
+except Exception as exc:
+status = str(exc)
+print(json.dumps({'success': status == '', 'status': status}))
+disconnect_wifi()
+del disconnect_wifi
+"""
+out, err = self._interface.execute(command, mode=self._submitMode)
+if err:
+return False, err
+result = json.loads(out.decode("utf-8").strip())
+return result["success"], result["status"]
+def writeCredentials(self, ssid, password):
+"""
+Public method to write the given credentials to the connected device and modify
+the start script to connect automatically.
+@param ssid SSID of the network to connect to
+@type str
+@param password password needed to authenticate
+@type str
+@return tuple containing a flag indicating success and an error message
+@rtype tuple of (bool, str)
+"""
+if not self.__deviceVolumeMounted():
+return False, self.tr("The device volume is not available.")
+workspace = self.getWorkspace()
+if Globals.versionToTuple(self._deviceData["release"]) >= (8, 0, 0):
+# CircuitPython >= 8.0.0: generate 'settings.toml' file
+contents = (
+'CIRCUITPY_WIFI_SSID = "{0}"\nCIRCUITPY_WIFI_PASSWORD = "{1}"\n'.format(
+ssid, password
+)
+)
+filename = os.path.join(workspace, "settings.toml")
+if os.path.exists(filename):
+ok = EricMessageBox.yesNo(
+None,
+self.tr("Write WiFi Credentials"),
+self.tr(
+"""<p>The file <b>{0}</b> exists already. Shall it be"""
+""" replaced?</p>"""
+).format(filename),
+icon=EricMessageBox.Warning,
+)
+if not ok:
+return False, self.tr("Aborted")
+try:
+with open(filename, "w") as f:
+f.write(contents)
+return True, ""
+except OSError as err:
+return False, str(err)
+else:
+# CircuitPython < 8.0.0: generate a secrets.py script
+# step 1: generate the secrets.py file
+contents = (
+'secrets = {{\n    "ssid": "{0}",\n    "password": "{1}",\n}}\n'.format(
+ssid, password
+)
+)
+filename = os.path.join(workspace, "secrets.py")
+if os.path.exists(filename):
+ok = EricMessageBox.yesNo(
+None,
+self.tr("Write WiFi Credentials"),
+self.tr(
+"""<p>The file <b>{0}</b> exists already. Shall it be"""
+""" replaced?</p>"""
+).format(filename),
+icon=EricMessageBox.Warning,
+)
+if not ok:
+return False, self.tr("Aborted")
+# step 2: create the auto-connect script (wifi_connect.py)
+try:
+with open(filename, "w") as f:
+f.write(contents)
+except OSError as err:
+return False, str(err)
+scriptFile = os.path.join(
+os.path.dirname(__file__), "MCUScripts", "circuitPy7WiFiConnect.py"
+)
+targetFile = os.path.join(workspace, "wifi_connect.py")
+try:
+shutil.copy2(scriptFile, targetFile)
+except OSError as err:
+return False, str(err)
+# Note: code.py will not be modified because the connection will be
+#       reset anyway
+return True, ""
+def removeCredentials(self):
+"""
+Public method to remove the saved credentials from the connected device.
+@return tuple containing a flag indicating success and an error message
+@rtype tuple of (bool, str)
+"""
+if not self.__deviceVolumeMounted():
+return False, self.tr("The device volume is not available.")
+workspace = self.getWorkspace()
+for name in ("settings.toml", "secrets.py"):
+filename = os.path.join(workspace, name)
+if os.path.exists(filename):
+os.remove(filename)
+return True, ""
+def checkInternet(self):
+"""
+Public method to check, if the internet can be reached.
+@return tuple containing a flag indicating reachability and an error string
+@rtype tuple of (bool, str)
+"""
+command = """
+def check_internet():
+import ipaddress
+import wifi
+r = wifi.radio
+if r.ipv4_address is not None:
+ping = r.ping(ipaddress.IPv4Address("8.8.8.8"))
+print(ping is not None)
+else:
+print(False)
+check_internet()
+del check_internet
+"""
+out, err = self._interface.execute(command, mode=self._submitMode)
+if err:
+return False, err
+return out.decode("utf-8").strip() == "True", ""
+def scanNetworks(self):
+"""
+Public method to scan for available WiFi networks.
+@return tuple containing the list of available networks as a tuple of 'Name',
+'MAC-Address', 'channel', 'RSSI' and 'security' and an error string
+@rtype tuple of (list of tuple of (str, str, int, int, str), str)
+"""
+command = """
+def scan_networks():
+import wifi
+r = wifi.radio
+network_list = []
+enabled = r.enabled
+if not enabled:
+r.enabled = True
+for net in r.start_scanning_networks():
+network_list.append(
+(net.ssid, net.bssid, net.channel, net.rssi,
+'_'.join(str(x).split('.')[-1] for x in net.authmode))
+)
+r.stop_scanning_networks()
+if not enabled:
+r.enabled = False
+print(network_list)
+scan_networks()
+del scan_networks
+"""
+out, err = self._interface.execute(
+command, mode=self._submitMode, timeout=15000
+)
+if err:
+return [], err
+networksList = ast.literal_eval(out.decode("utf-8"))
+networks = []
+seenNetworks = []
+for network in networksList:
+if network[0]:
+ssid = network[0]
+mac = binascii.hexlify(network[1], ":").decode("utf-8")
+channel = network[2]
+rssi = network[3]
+try:
+security = self.__securityTranslations[network[4]]
+except KeyError:
+security = self.tr("unknown ({0})").format(network[4])
+if (ssid, mac, channel) not in seenNetworks:
+seenNetworks.append((ssid, mac, channel))
+networks.append((ssid, mac, channel, rssi, security))
+return networks, ""
+def deactivateInterface(self, interface):
+"""
+Public method to deactivate a given WiFi interface of the connected device.
+Note: With CircuitPython it is not possible to deactivate the station and
+access point interfaces separately.
+@param interface designation of the interface to be deactivated (one of 'AP'
+or 'STA')
+@type str
+@return tuple containg a flag indicating success and an error message
+@rtype tuple of (bool, str)
+@exception ValueError raised to indicate a wrong value for the interface type
+"""
+if interface not in ("STA", "AP"):
+raise ValueError(
+"interface must be 'AP' or 'STA', got '{0}'".format(interface)
+)
+command = """
+def deactivate():
+import wifi
+wifi.radio.enabled = False
+print(not wifi.radio.enabled)
+deactivate()
+del deactivate
+"""
+out, err = self._interface.execute(command, mode=self._submitMode)
+if err:
+return False, err
+else:
+return out.decode("utf-8").strip() == "True", ""
+def startAccessPoint(self, ssid, security=None, password=None, ifconfig=None):
+"""
+Public method to start the access point interface.
+@param ssid SSID of the access point
+@type str
+@param security security method (defaults to None)
+@type int (optional)
+@param password password (defaults to None)
+@type str (optional)
+@param ifconfig IPv4 configuration for the access point if not default
+(IPv4 address, netmask, gateway address, DNS server address)
+@type tuple of (str, str, str, str)
+@return tuple containing a flag indicating success and an error message
+@rtype tuple of (bool, str)
+"""
+if security is None or password is None:
+security = 0
+password = ""
+authmode = self.__securityCode2AuthModeString[security]
+if ifconfig:
+return (
+False,
+self.tr(
+"CircuitPython does not support setting the IPv4 parameters of the"
+" WiFi access point."
+),
+)
+command = """
+def start_ap(ssid, password):
+import wifi
+r = wifi.radio
+try:
+r.start_ap(ssid, password, authmode={2})
+except ValueError as exc:
+print('Error:', str(exc))
+start_ap({0}, {1})
+del start_ap
+""".format(
+repr(ssid), repr(password), authmode
+)
+out, err = self._interface.execute(
+command, mode=self._submitMode, timeout=15000
+)
+if err:
+return False, err
+elif out and out.startswith(b"Error:"):
+return False, out.decode("utf-8").split(None, 1)[-1]
+else:
+return True, ""
+def stopAccessPoint(self):
+"""
+Public method to stop the access point interface.
+@return tuple containg a flag indicating success and an error message
+@rtype tuple of (bool, str)
+"""
+command = """
+def stop_ap():
+import wifi
+r = wifi.radio
+try:
+r.stop_ap()
+except NotImplementedError as exc:
+print('Error:', str(exc))
+stop_ap()
+del stop_ap
+"""
+out, err = self._interface.execute(command, mode=self._submitMode)
+if err:
+return False, err
+elif out and out.startswith(b"Error:"):
+return False, out.decode("utf-8").split(None, 1)[-1]
+else:
+return True, ""
+def getConnectedClients(self):
+"""
+Public method to get a list of connected clients.
+@return a tuple containing a list of tuples containing the client MAC-Address
+and the RSSI (if supported and available) and an error message
+@rtype tuple of ([(bytes, int)], str)
+"""
+return (
+[],
+self.tr("CircuitPython does not support reporting of connected clients"),
+)
 def createDevice(microPythonWidget, deviceType, vid, pid, boardName, serialNumber):
 """
 Function to instantiate a MicroPython device object.

Mercurial Repositories > eric / file comparison

comparison: src/eric7/MicroPython/Devices/CircuitPythonDevices.py

src/eric7/MicroPython/Devices/CircuitPythonDevices.py