Merge pull request #2 from AndyEveritt/dev

Dev

Author: AndyEveritt

DuetWebAPI/DuetWebAPI.py

@@ -1 +1 @@
-from .DuetWebAPI import DuetWebAPI
+from .dwa_factory import DuetWebAPI

DuetWebAPI/__init__.py

@@ -0,0 +1,4 @@
+from .base import DuetAPI
+from .dsf_api import DSFAPI
+from .dwc_api import DWCAPI
+from .wrapper import DuetAPIWrapper

DuetWebAPI/api/__init__.py

@@ -0,0 +1,47 @@
+from typing import Dict, List
+import requests
+import os
+import logging
+
+
+class DuetAPI:
+    api_name = ''
+
+    def __init__(self, base_url) -> None:
+        self._base_url = base_url
+
+    def get_model(self, key: str = None) -> Dict:
+        """ Get Duet object model. RRF3 only """
+        raise NotImplementedError
+
+    def post_code(self, code: str) -> Dict:
+        """ Send G/M/T-code to Duet """
+        raise NotImplementedError
+
+    def get_file(self, filename: str, directory: str = 'gcodes') -> str:
+        """ Get file from Duet """
+        raise NotImplementedError
+
+    def put_file(self, file: str, directory: str = 'gcodes') -> Dict:
+        """ Upload file to Duet """
+        raise NotImplementedError
+
+    def get_fileinfo(self, filename: str = None, directory: str = 'gcodes') -> Dict:
+        """ Get file info """
+        raise NotImplementedError
+
+    def delete_file(self, filename: str, directory: str = 'gcodes') -> Dict:
+        """ Delete file on Duet """
+        raise NotImplementedError
+
+    def move_file(self, from_path: str, to_path: str, force: bool = False) -> Dict:
+        """ Move file on Duet, can be used to rename files """
+        raise NotImplementedError
+
+    def get_directory(self, directory: str) -> List[Dict]:
+        """ Get a list of all the files & directories in a directory """
+        raise NotImplementedError
+
+    def put_directory(self, directory: str) -> Dict:
+        """ Create a new directory """
+        raise NotImplementedError

DuetWebAPI/api/base.py

@@ -0,0 +1,92 @@
+import logging
+import os
+from typing import Dict, List
+
+import requests
+
+from .base import DuetAPI
+
+
+class DSFAPI(DuetAPI):
+    """
+    Duet Software Framework REST API Interface.
+
+    Used with a Duet 3 + SBC.
+    Must use RRF3.
+    """
+    api_name = 'DSF_REST'
+
+    def get_model(self, **_ignored) -> Dict:
+        url = f'{self._base_url}/machine/status'
+        r = requests.get(url)
+        j = r.json()
+        return j
+
+    def post_code(self, code: str) -> Dict:
+        url = f'{self._base_url}/machine/code'
+        r = requests.post(url, data=code)
+        return {'response': r.text}
+
+    def get_file(self, filename: str, directory: str = 'gcodes') -> str:
+        """
+        filename: name of the file you want to download including extension
+        directory: the folder that the file is in, options are ['gcodes', 'macros', 'sys']
+
+        returns the file as a string
+        """
+        url = f'{self._base_url}/machine/file/{directory}/{filename}'
+        r = requests.get(url)
+        if not r.ok:
+            raise ValueError
+        return r.text
+
+    def put_file(self, file: str, directory: str = 'gcodes') -> Dict:
+        """
+        file: the path to the file you want to upload from your PC
+        directory: the folder that the file is in, options are ['gcodes', 'macros', 'sys']
+
+        returns the file as a string
+        """
+        file = os.path.abspath(file).replace('\\', '/')
+        filename = file.split('/')[-1]
+        url = f'{self._base_url}/machine/file/{directory}/{filename}'
+        with open(file, 'rb') as f:
+            r = requests.put(url, data=f, headers={'Content-Type': 'application/octet-stream'})
+        if not r.ok:
+            raise ValueError
+        return {'err': 0}
+
+    def get_fileinfo(self, filename: str = None, directory: str = 'gcodes') -> Dict:
+        url = f'{self._base_url}/machine/fileinfo/{directory}/{filename}'
+        r = requests.get(url)
+        if not r.ok:
+            raise ValueError
+        return r.json()
+
+    def delete_file(self, filename: str, directory: str = 'gcodes') -> Dict:
+        url = f'{self._base_url}/machine/file/{directory}/{filename}'
+        r = requests.delete(url)
+        if not r.ok:
+            raise ValueError
+        return {'err': 0}
+
+    def move_file(self, from_path: str, to_path: str, force: bool = False) -> Dict:
+        url = f'{self._base_url}/machine/file/move'
+        r = requests.post(url, {'from': f'{from_path}', 'to': f'{to_path}', 'force': force})
+        if not r.ok:
+            raise ValueError
+        return {'err': 0}
+
+    def get_directory(self, directory: str) -> List[Dict]:
+        url = f'{self._base_url}/machine/directory/{directory}'
+        r = requests.get(url)
+        if not r.ok:
+            raise ValueError
+        return r.json()
+
+    def put_directory(self, directory: str) -> Dict:
+        url = f'{self._base_url}/machine/directory/{directory}'
+        r = requests.put(url)
+        if not r.ok:
+            raise ValueError
+        return {'err': 0}