wwshell file management CLI

circup/backends.py

@@ -73,7 +73,7 @@ def get_device_versions(self):
         """
         return self.get_modules(os.path.join(self.device_location, self.LIB_DIR_PATH))
 
-    def _create_library_directory(self, device_path, library_path):
+    def create_directory(self, device_path, directory):
         """
         To be overridden by subclass
         """
@@ -97,6 +97,12 @@ def copy_file(self, target_file, location_to_paste):
         """
         raise NotImplementedError
 
+    def upload_file(self, target_file, location_to_paste):
+        """Paste a copy of the specified file at the location given
+        To be overridden by subclass
+        """
+        raise NotImplementedError
+
     # pylint: disable=too-many-locals,too-many-branches,too-many-arguments,too-many-nested-blocks,too-many-statements
     def install_module(
         self, device_path, device_modules, name, pyext, mod_names, upgrade=False
@@ -189,7 +195,7 @@ def install_module(
                 return
 
             # Create the library directory first.
-            self._create_library_directory(device_path, library_path)
+            self.create_directory(device_path, library_path)
             if local_path is None:
                 if pyext:
                     # Use Python source for module.
@@ -281,7 +287,9 @@ def __init__(  # pylint: disable=too-many-arguments
     ):
         super().__init__(logger)
         if password is None:
-            raise ValueError("--host needs --password")
+            raise ValueError(
+                "Must pass --password or set CIRCUP_WEBWORKFLOW_PASSWORD environment variable"
+            )
 
         # pylint: disable=no-member
         # verify hostname/address
@@ -305,6 +313,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self.library_path = self.device_location + "/" + self.LIB_DIR_PATH
         self.timeout = timeout
         self.version_override = version_override
+        self.FS_URL = urljoin(self.device_location, self.FS_PATH)
 
     def install_file_http(self, source, location=None):
         """
@@ -316,6 +325,7 @@ def install_file_http(self, source, location=None):
         file_name = source.split(os.path.sep)
         file_name = file_name[-2] if file_name[-1] == "" else file_name[-1]
 
+        print(f"inside install_file_http location: '{location}'")
         if location is None:
             target = self.device_location + "/" + self.LIB_DIR_PATH + file_name
         else:
@@ -324,7 +334,10 @@ def install_file_http(self, source, location=None):
         auth = HTTPBasicAuth("", self.password)
 
         with open(source, "rb") as fp:
+            print(f"upload file PUT URL: {target}")
             r = self.session.put(target, fp.read(), auth=auth, timeout=self.timeout)
+            print(f"install_file_http response status: {r.status_code}")
+            print(r.content)
             if r.status_code == 409:
                 _writeable_error()
             r.raise_for_status()
@@ -542,10 +555,9 @@ def _get_modules_http_single_mods(self, auth, result, single_file_mods, url):
             metadata["path"] = sfm_url
             result[sfm[:idx]] = metadata
 
-    def _create_library_directory(self, device_path, library_path):
-        url = urlparse(device_path)
-        auth = HTTPBasicAuth("", url.password)
-        with self.session.put(library_path, auth=auth, timeout=self.timeout) as r:
+    def create_directory(self, device_path, directory):
+        auth = HTTPBasicAuth("", self.password)
+        with self.session.put(directory, auth=auth, timeout=self.timeout) as r:
             if r.status_code == 409:
                 _writeable_error()
             r.raise_for_status()
@@ -556,11 +568,57 @@ def copy_file(self, target_file, location_to_paste):
                 self.device_location,
                 "/".join(("fs", location_to_paste, target_file, "")),
             )
-            self._create_library_directory(self.device_location, create_directory_url)
+            self.create_directory(self.device_location, create_directory_url)
             self.install_dir_http(target_file)
         else:
             self.install_file_http(target_file)
 
+    def upload_file(self, target_file, location_to_paste):
+        """
+        copy a file from the host PC to the microcontroller
+        :param target_file: file on the host PC to copy
+        :param location_to_paste: Location on the microcontroller to paste it.
+        :return:
+        """
+        print(f"inside upload_file location_to_paste: '{location_to_paste}'")
+        if os.path.isdir(target_file):
+            create_directory_url = urljoin(
+                self.device_location,
+                "/".join(("fs", location_to_paste, target_file, "")),
+            )
+            self.create_directory(self.device_location, create_directory_url)
+            self.install_dir_http(target_file, location_to_paste)
+        else:
+            self.install_file_http(target_file, location_to_paste)
+
+    def download_file(self, target_file, location_to_paste):
+        """
+        Download a file from the MCU device to the local host PC
+        :param target_file: The file on the MCU to download
+        :param location_to_paste: The location on the host PC to put the downloaded copy.
+        :return:
+        """
+        auth = HTTPBasicAuth("", self.password)
+        with self.session.get(
+            self.FS_URL + target_file, timeout=self.timeout, auth=auth
+        ) as r:
+            if r.status_code == 404:
+                click.secho(f"{target_file} was not found on the device", "red")
+
+            file_name = target_file.split("/")[-1]
+            if location_to_paste is None:
+                with open(file_name, "wb") as f:
+                    f.write(r.content)
+
+                click.echo(f"Downloaded File: {file_name}")
+            else:
+                with open(os.path.join(location_to_paste, file_name), "wb") as f:
+                    f.write(r.content)
+
+                click.echo(
+                    f"Downloaded File: {os.path.join(location_to_paste, file_name)}"
+                )
+
     def install_module_mpy(self, bundle, metadata):
         """
         :param bundle library bundle.
@@ -636,6 +694,7 @@ def file_exists(self, filepath):
         return True if the file exists, otherwise False.
         """
         auth = HTTPBasicAuth("", self.password)
+        print(f"URL: {self.get_file_path(filepath)}")
         resp = requests.get(
             self.get_file_path(filepath), auth=auth, timeout=self.timeout
         )
@@ -664,11 +723,7 @@ def get_file_path(self, filename):
         """
         retuns the full path on the device to a given file name.
         """
-        return urljoin(
-            urljoin(self.device_location, "fs/", allow_fragments=False),
-            filename,
-            allow_fragments=False,
-        )
+        return "/".join((self.device_location, "fs", filename))
 
     def is_device_present(self):
         """
@@ -739,6 +794,20 @@ def get_free_space(self):
                 return r.json()["free"] * r.json()["block_size"]  # bytes
             sys.exit(1)
 
+    def list_dir(self, dirpath):
+        """
+        Returns the list of files located in the given dirpath.
+        """
+        auth = HTTPBasicAuth("", self.password)
+        with self.session.get(
+            urljoin(self.device_location, f"fs/{dirpath if dirpath else ''}"),
+            auth=auth,
+            headers={"Accept": "application/json"},
+            timeout=self.timeout,
+        ) as r:
+            print(r.content)
+            return r.json()["files"]
+
 
 class DiskBackend(Backend):
     """
@@ -817,9 +886,9 @@ def _get_modules(self, device_lib_path):
         """
         return _get_modules_file(device_lib_path, self.logger)
 
-    def _create_library_directory(self, device_path, library_path):
-        if not os.path.exists(library_path):  # pragma: no cover
-            os.makedirs(library_path)
+    def create_directory(self, device_path, directory):
+        if not os.path.exists(directory):  # pragma: no cover
+            os.makedirs(directory)
 
     def copy_file(self, target_file, location_to_paste):
         target_filename = target_file.split(os.path.sep)[-1]
@@ -834,6 +903,9 @@ def copy_file(self, target_file, location_to_paste):
                 os.path.join(self.device_location, location_to_paste, target_filename),
             )
 
+    def upload_file(self, target_file, location_to_paste):
+        self.copy_file(target_file, location_to_paste)
+
     def install_module_mpy(self, bundle, metadata):
         """
         :param bundle library bundle.

circup/command_utils.py

@@ -625,3 +625,29 @@ def get_device_path(host, password, path):
     else:
         device_path = find_device()
     return device_path
+
+
+def sorted_by_directory_then_alpha(list_of_files):
+    """
+    Sort the list of files into alphabetical seperated
+    with directories grouped together before files.
+    """
+    dirs = {}
+    files = {}
+
+    for cur_file in list_of_files:
+        if cur_file["directory"]:
+            dirs[cur_file["name"]] = cur_file
+        else:
+            files[cur_file["name"]] = cur_file
+
+    sorted_dir_names = sorted(dirs.keys())
+    sorted_file_names = sorted(files.keys())
+
+    sorted_full_list = []
+    for cur_name in sorted_dir_names:
+        sorted_full_list.append(dirs[cur_name])
+    for cur_name in sorted_file_names:
+        sorted_full_list.append(files[cur_name])
+
+    return sorted_full_list

circup/wwshell/README.rst

@@ -0,0 +1,105 @@
+
+wwshell
+=======
+
+.. image:: https://readthedocs.org/projects/circup/badge/?version=latest
+    :target: https://circuitpython.readthedocs.io/projects/circup/en/latest/
+    :alt: Documentation Status
+
+.. image:: https://img.shields.io/discord/327254708534116352.svg
+    :target: https://adafru.it/discord
+    :alt: Discord
+
+
+.. image:: https://github.com/adafruit/circup/workflows/Build%20CI/badge.svg
+    :target: https://github.com/adafruit/circup/actions
+    :alt: Build Status
+
+
+.. image:: https://img.shields.io/badge/code%20style-black-000000.svg
+    :target: https://github.com/psf/black
+    :alt: Code Style: Black
+
+
+A tool to manage files on a CircuitPython device via wireless workflows.
+Currently supports Web Workflow.
+
+.. contents::
+
+Installation
+------------
+
+wwshell is bundled along with Circup. When you install Circup you'll get wwshell automatically.
+
+Circup requires Python 3.5 or higher.
+
+In a `virtualenv <https://virtualenv.pypa.io/en/latest/>`_,
+``pip install circup`` should do the trick. This is the simplest way to make it
+work.
+
+If you have no idea what a virtualenv is, try the following command,
+``pip3 install --user circup``.
+
+.. note::
+
+    If you use the ``pip3`` command to install CircUp you must make sure that
+    your path contains the directory into which the script will be installed.
+    To discover this path,
+
+    * On Unix-like systems, type ``python3 -m site --user-base`` and append
+      ``bin`` to the resulting path.
+    * On Windows, type the same command, but append ``Scripts`` to the
+      resulting path.
+
+What does wwshell do?
+---------------------
+
+It lets you view, delete, upload, and download files from your Circuitpython device
+via wireless workflows. Similar to ampy, but operates over wireless workflow rather
+than USB serial.
+
+Usage
+-----
+
+To use web workflow you need to enable it by putting WIFI credentials and a web workflow
+password into your settings.toml file. `See here <https://learn.adafruit.com/getting-started-with-web-workflow-using-the-code-editor/device-setup>`_,
+
+To get help, just type the command::
+
+    $ wwshell
+    Usage: wwshell [OPTIONS] COMMAND [ARGS]...
+
+      A tool to manage files CircuitPython device over web workflow.
+
+    Options:
+      --verbose          Comprehensive logging is sent to stdout.
+      --path DIRECTORY   Path to CircuitPython directory. Overrides automatic path
+                         detection.
+      --host TEXT        Hostname or IP address of a device. Overrides automatic
+                         path detection.
+      --password TEXT    Password to use for authentication when --host is used.
+                         You can optionally set an environment variable
+                         CIRCUP_WEBWORKFLOW_PASSWORD instead of passing this
+                         argument. If both exist the CLI arg takes precedent.
+      --timeout INTEGER  Specify the timeout in seconds for any network
+                         operations.
+      --version          Show the version and exit.
+      --help             Show this message and exit.
+
+    Commands:
+      get  Download a copy of a file or directory from the device to the...
+      ls   Lists the contents of a directory.
+      put  Upload a copy of a file or directory from the local computer to...
+      rm   Delete a file on the device.
+
+
+.. note::
+
+    If you find a bug, or you want to suggest an enhancement or new feature
+    feel free to create an issue or submit a pull request here:
+
+    https://github.com/adafruit/circup
+
+
+Discussion of this tool happens on the Adafruit CircuitPython
+`Discord channel <https://discord.gg/rqrKDjU>`_.

circup/wwshell/README.rst.license

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2024 Tim Cocks, written for Adafruit Industries
+#
+# SPDX-License-Identifier: MIT

circup/wwshell/__init__.py

@@ -0,0 +1,14 @@
+# SPDX-FileCopyrightText: 2024 Tim Cocks, written for Adafruit Industries
+#
+# SPDX-License-Identifier: MIT
+"""
+wwshell is a CLI utility for managing files on CircuitPython devices via wireless workflows.
+It currently supports Web Workflow.
+"""
+from .commands import main
+
+
+# Allows execution via `python -m circup ...`
+# pylint: disable=no-value-for-parameter
+if __name__ == "__main__":
+    main()