MicroPython stuff

code/MicroPython/.gitignore

@@ -0,0 +1,2 @@
+microdot/
+utemplate/

code/MicroPython/README.md

@@ -0,0 +1,37 @@
+#
+
+## 
+
+```shell
+# ...
+esptool.py --port /dev/ttyACM0 erase_flash
+# ...
+esptool.py --chip esp32s2 --port /dev/ttyACM0 write_flash -z 0x1000 ~/Downloads/LOLIN_S2_MINI-20240222-v1.22.2.bin
+```
+
+```shell
+pip3 install rshell   #
+pip3 install mpremote #
+```
+
+### Выполнение кода
+
+```shell
+rshell --port /dev/ttyACM0 cat blink.py
+```
+
+```python
+import machine
+import time
+
+led = machine.Pin(15, machine.Pin.OUT)
+
+led.on()
+time.sleep(1)
+led.off()
+time.sleep(1)
+```
+
+```shell
+mpremote exec 'exec(open("blink.py").read())'
+```

code/MicroPython/blink.py

@@ -0,0 +1,9 @@
+import machine
+import time
+
+led = machine.Pin(15, machine.Pin.OUT)
+
+led.on()
+time.sleep(1)
+led.off()
+time.sleep(1)

code/MicroPython/boot.py

@@ -0,0 +1,4 @@
+"""
+Скрипт, который выполняется при старте EPS32.
+После него выполняется скрипт `main.py`.
+"""

code/MicroPython/ftp.py

@@ -0,0 +1,163 @@
+import socket
+import network
+import os
+import _thread
+
+DATA_PORT = 13333
+
+ftpsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+ftpsocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+ftpsocket.bind(socket.getaddrinfo("0.0.0.0", 21)[0][4])
+datasocket.bind(socket.getaddrinfo("0.0.0.0", DATA_PORT)[0][4])
+
+ftpsocket.listen(1)
+datasocket.listen(1)
+datasocket.settimeout(10)
+
+dataclient = None
+
+
+def send_list_data(cwd, dataclient):
+    for file in os.listdir(cwd):
+        stat = os.stat(get_absolute_path(cwd, file))
+        file_permissions = "drwxr-xr-x" if (stat[0] & 0o170000 == 0o040000) else "-rw-r--r--"
+        file_size = stat[6]
+        description = "{}    1 owner group {:>13} Jan 1  1980 {}\r\n".format(
+            file_permissions, file_size, file
+        )
+        dataclient.sendall(description)
+
+
+def send_file_data(path, dataclient):
+    with open(path) as file:
+        chunk = file.read(128)
+        while len(chunk) > 0:
+            dataclient.sendall(chunk)
+            chunk = file.read(128)
+
+
+def save_file_data(path, dataclient):
+    with open(path, "w") as file:
+        chunk = dataclient.read(128)
+        while len(chunk) > 0:
+            file.write(chunk)
+            chunk = dataclient.read(128)
+
+
+def get_absolute_path(cwd, payload):
+    # if it doesn't start with / consider
+    # it a relative path
+    if not payload.startswith("/"):
+        payload = cwd + "/" + payload
+    # and don't leave any trailing /
+    return payload.rstrip("/")
+
+
+def ftp():
+    try:
+        dataclient = None
+        while True:
+            cwd = "/"
+            cl, remote_addr = ftpsocket.accept()
+            cl.settimeout(300)
+            try:
+                print("FTP connection from:", remote_addr)
+                cl.sendall("220 Hello, this is the ESP8266.\r\n")
+                while True:
+                    data = cl.readline().decode("utf-8").replace("\r\n", "")
+                    if len(data) <= 0:
+                        print("Client is dead")
+                        break
+
+                    command, payload = (data.split(" ") + [""])[:2]
+                    command = command.upper()
+
+                    print("Command={}, Payload={}".format(command, payload))
+
+                    if command == "USER":
+                        cl.sendall("230 Logged in.\r\n")
+                    elif command == "SYST":
+                        cl.sendall("215 ESP8266 MicroPython\r\n")
+                    elif command == "SYST":
+                        cl.sendall("502\r\n")
+                    elif command == "PWD":
+                        cl.sendall('257 "{}"\r\n'.format(cwd))
+                    elif command == "CWD":
+                        path = get_absolute_path(cwd, payload)
+                        try:
+                            files = os.listdir(path)
+                            cwd = path
+                            cl.sendall('250 Directory changed successfully\r\n')
+                        except:
+                            cl.sendall('550 Failed to change directory\r\n')
+                    elif command == "EPSV":
+                        cl.sendall('502\r\n')
+                    elif command == "TYPE":
+                        # probably should switch between binary and not
+                        cl.sendall('200 Transfer mode set\r\n')
+                    elif command == "SIZE":
+                        path = get_absolute_path(cwd, payload)
+                        try:
+                            size = os.stat(path)[6]
+                            cl.sendall('213 {}\r\n'.format(size))
+                        except:
+                            cl.sendall('550 Could not get file size\r\n')
+                    elif command == "QUIT":
+                        cl.sendall('221 Bye.\r\n')
+                    elif command == "PASV":
+                        addr = network.WLAN().ifconfig()[0]
+                        cl.sendall(
+                            '227 Entering Passive Mode ({},{},{}).\r\n'.format(
+                                addr.replace('.', ','), DATA_PORT >> 8, DATA_PORT % 256
+                            )
+                        )
+                        dataclient, data_addr = datasocket.accept()
+                        print("FTP Data connection from:", data_addr)
+                    elif command == "LIST":
+                        try:
+                            send_list_data(cwd, dataclient)
+                            dataclient.close()
+                            cl.sendall("150 Here comes the directory listing.\r\n")
+                            cl.sendall("226 Listed.\r\n")
+                        except:
+                            cl.sendall('550 Failed to list directory\r\n')
+                        finally:
+                            dataclient.close()
+                    elif command == "RETR":
+                        try:
+                            send_file_data(get_absolute_path(cwd, payload), dataclient)
+                            dataclient.close()
+                            cl.sendall("150 Opening data connection.\r\n")
+                            cl.sendall("226 Transfer complete.\r\n")
+                        except:
+                            cl.sendall('550 Failed to send file\r\n')
+                        finally:
+                            dataclient.close()
+                    elif command == "STOR":
+                        try:
+                            cl.sendall("150 Ok to send data.\r\n")
+                            save_file_data(get_absolute_path(cwd, payload), dataclient)
+                            dataclient.close()
+                            cl.sendall("226 Transfer complete.\r\n")
+                        except:
+                            cl.sendall('550 Failed to send file\r\n')
+                        finally:
+                            dataclient.close()
+                    else:
+                        cl.sendall("502 Unsupported command.\r\n")
+                        print("Unsupported command {} with payload {}".format(command, payload))
+
+            finally:
+                cl.close()
+    finally:
+        datasocket.close()
+        ftpsocket.close()
+        if dataclient is not None:
+            dataclient.close()
+
+
+_thread.start_new_thread(ftp, ())

code/MicroPython/get_mac.py

@@ -0,0 +1,13 @@
+"""
+Скрипт, который возвращает MAC адрес
+
+https://git.a2s.su/iiiypuk/snipplets.dev/src/branch/master/code/MicroPython
+Update: May 14 2024
+"""
+
+import network
+import ubinascii
+
+mac = ubinascii.hexlify(network.WLAN().config('mac'), ':').decode()
+
+print(mac)

code/MicroPython/http.py

@@ -0,0 +1,79 @@
+import machine
+import time
+import gc
+
+from microdot import Microdot, send_file, Request, Response
+
+from microdot.utemplate import Template
+
+led = machine.Pin(15, machine.Pin.OUT)
+
+app = Microdot()
+Response.default_content_type = 'text/html'
+Request.max_content_length = 1024 * 1024
+
+
+def cmd_parse(method, args):
+    if method == 'GET':
+        if 'cmd' in args:
+            cmd = args['cmd']
+
+            if 'reboot' == cmd:
+                for x in range(2):
+                    led.on()
+                    time.sleep_ms(200)
+                    led.off()
+                    time.sleep_ms(200)
+
+                machine.reset()
+
+            if 'ftp' == cmd:
+                exec(open('ftp.py').read())
+
+
+@app.route('/static/<path:path>')
+async def static(request, path):
+    if '..' in path:
+        # directory traversal is not allowed
+        return 'Not found', 404
+    return send_file('static/' + path)
+
+
+@app.get('/file')
+async def file(request):
+    return send_file('templates/upload.html')
+
+
+@app.post('/upload')
+async def upload(request):
+    filename = request.headers['Content-Disposition'].split('filename=')[1].strip('"')
+    size = int(request.headers['Content-Length'])
+
+    filename = filename.replace('/', '_')
+
+    with open('/' + filename, 'wb') as f:
+        while size > 0:
+            chunk = await request.stream.read(min(size, 1024))
+            f.write(chunk)
+            size -= len(chunk)
+
+    print('Successfully saved file: ' + filename)
+    return ''
+
+
+@app.route('/')
+async def index(request):
+    # print(request.args, request.path, request.method, sep='\n')
+    # print(request.url, request.query_string, request.url_args, sep='\n')
+    print(request.args)
+
+    cmd_parse(request.method, request.args)
+
+    ram_f = gc.mem_free()
+    ram_a = gc.mem_alloc()
+    ram_t = ram_a + ram_f
+    return Template('index.html').render(ram_free=int(ram_f / 1024), ram_total=int(ram_t / 1024))
+
+
+print('HTTP server started')
+app.run(port=80)

code/MicroPython/main.py

@@ -0,0 +1,17 @@
+"""
+Скрипт, который выполняется при старте EPS32.
+После выполнения скрипта `boot.py`.
+"""
+
+import time
+import network_ap
+import network_sta
+
+# Connecting to Wifi
+wifi_connect = network_sta.connect('Keenetic-8647', 'sFJZi7U51')
+if wifi_connect == False:
+    wifi_connect = network_ap.connect(essid='SkyNet', clients=2)
+
+# If WiFi connected run HTTP server
+if wifi_connect == True:
+    exec(open('http.py').read())

code/MicroPython/network_ap.py

@@ -0,0 +1,21 @@
+import time
+import network
+
+
+def connect(essid='ESP', clients=3) -> bool:
+    print('Starting AP: {0}...'.format(essid))
+    ap = network.WLAN(network.AP_IF)
+    ap.active(True)
+    ap.config(essid=essid)
+    ap.config(max_clients=clients)
+
+    time.sleep(3)
+
+    if ap.isconnected():
+        print('AP "{0}" started'.format(essid))
+
+        return True
+    else:
+        print('Starting AP failed!')
+
+        return False

code/MicroPython/network_sta.py

@@ -0,0 +1,38 @@
+import time
+import network
+
+
+def connect(ssid: str, password: str) -> bool:
+    wlan = network.WLAN(network.STA_IF)
+    wlan.active(True)
+    wlan.config(reconnects=2)
+
+    if not wlan.isconnected():
+        print('Scanning networks...')
+        wlan.scan()
+        print('Connecting to {0}...'.format(ssid))
+        # try:
+        wlan.connect(ssid, password)
+
+        while not wlan.isconnected():
+            status = wlan.status()
+            if status in (network.STAT_IDLE, network.STAT_WRONG_PASSWORD, network.STAT_NO_AP_FOUND):
+                print('\nConnection to "{0}" failed!'.format(ssid))
+                wlan.active(False)
+
+                return False
+
+            print('.', end='')
+            time.sleep(0.1)
+
+        print('\nConnection success!')
+        ifconfig = wlan.ifconfig()
+        print('IP: {0}'.format(ifconfig[0]))
+
+        return True
+    else:
+        print('Already connected!')
+        ifconfig = wlan.ifconfig()
+        print('IP: {0}'.format(ifconfig[0]))
+
+        return True

code/MicroPython/pyproject.toml

@@ -0,0 +1,3 @@
+[tool.black]
+skip-string-normalization = true
+line-length = 100

code/MicroPython/read_write_file.py

@@ -0,0 +1,13 @@
+import os
+
+print('Listing filesystem...')
+print(os.listdir(), '\n')
+
+print('Write data...')
+with open('data.txt', 'w') as f:
+    count = f.write('ololo\n')
+    print('Write', count, 'bytes')
+
+print('Read data...')
+with open('data.txt', 'r') as f:
+    print('Data:', f.read())

code/MicroPython/static/styles.css

@@ -0,0 +1,3 @@
+body {
+  background-color: #ccc;
+}

code/MicroPython/templates/index.html

@@ -0,0 +1,18 @@
+{% args ram_free, ram_total %}
+<!doctype html>
+<html>
+  <head>
+    <title>S2F Terminal</title>
+    <meta charset="utf-8">
+    <meta name="viewport" content="width=device-width,initial-scale=1">
+    <link href="/static/styles.css" rel="stylesheet">
+  </head>
+  <body>
+    <h1>by Alexander Popov</h1>
+    <p>Free RAM: {{ ram_free }} of {{ ram_total }}</p>
+    <div>
+      <a href="/?cmd=ftp">Поднять FTP сервер</a><br>
+      <a href="/?cmd=reboot">Перезапустить</a><br>
+    </div>
+  </body>
+</html>

code/MicroPython/templates/index_html.py

@@ -0,0 +1,23 @@
+# Autogenerated file
+def render(ram_free, ram_total):
+    yield """<!doctype html>
+<html>
+  <head>
+    <title>S2F Terminal</title>
+    <meta charset=\"utf-8\">
+    <meta name=\"viewport\" content=\"width=device-width,initial-scale=1\">
+    <link href=\"/static/styles.css\" rel=\"stylesheet\">
+  </head>
+  <body>
+    <h1>by Alexander Popov</h1>
+    <p>Free RAM: """
+    yield str(ram_free)
+    yield """ of """
+    yield str(ram_total)
+    yield """</p>
+    <div>
+      <a href=\"/?cmd=ftp\">Поднять FTP сервер</a><br>
+      <a href=\"/?cmd=reboot\">Перезапустить</a><br>
+    </div>
+  </body>
+</html>"""

code/MicroPython/templates/upload.html

@@ -0,0 +1,35 @@
+<!doctype html>
+<html>
+  <head>
+    <title>Microdot Upload Example</title>
+    <meta charset="UTF-8">
+  </head>
+  <body>
+    <h1>Microdot Upload Example</h1>
+    <form id="form">
+      <input type="file" id="file" name="file" />
+      <input type="submit" value="Upload" />
+    </form>
+    <script>
+      async function upload(ev) {
+        ev.preventDefault();
+        const file = document.getElementById('file').files[0];
+        if (!file) {
+          return;
+        }
+        await fetch('/upload', {
+          method: 'POST',
+          body: file,
+          headers: {
+            'Content-Type': 'application/octet-stream',
+            'Content-Disposition': `attachment; filename="${file.name}"`,
+          },
+        }).then(res => {
+          console.log('Upload accepted');
+          window.location.href = '/';
+        });
+      }
+      document.getElementById('form').addEventListener('submit', upload);
+    </script>
+  </body>
+</html>