feat(pydevice): file integrity support, simple run in pydevice.

Signed-off-by: Braden Mars <bradenmars@bradenmars.me>
BradenM · Apr 17, 2023 · 8b7a255 · 8b7a255
1 parent a3cbf31
commit 8b7a255
Show file tree

Hide file tree

Showing 2 changed files with 61 additions and 15 deletions.
diff --git a/micropy/pyd/abc.py b/micropy/pyd/abc.py
@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import abc
+from io import BytesIO, StringIO
 from pathlib import Path
-from typing import Any, AnyStr, NewType, Protocol
+from typing import Any, AnyStr, Generic, NewType, Protocol, TypeVar
 
 HostPath = NewType("HostPath", str)
 DevicePath = NewType("DevicePath", str)
@@ -136,8 +137,11 @@ def eval_script(
         ...
 
 
-class MetaPyDevice(abc.ABC):
-    pydevice: MetaPyDeviceBackend
+AnyBackend = TypeVar("AnyBackend", bound=MetaPyDeviceBackend)
+
+
+class MetaPyDevice(abc.ABC, Generic[AnyBackend]):
+    pydevice: AnyBackend
     stream_consumer: StreamConsumer | None
     message_consumer: MessageConsumer | None
 
@@ -154,9 +158,17 @@ def copy_to(self, source_path: HostPath, target_path: DevicePath) -> None:
         ...
 
     @abc.abstractmethod
-    def copy_from(self, source_path: DevicePath, target_path: HostPath) -> None:
+    def copy_from(
+        self, source_path: DevicePath, target_path: HostPath, *, verify_integrity: bool = True
+    ) -> None:
         ...
 
     @abc.abstractmethod
-    def run_script(self, content: AnyStr, target_path: DevicePath | None = None):
+    def remove(self, target_path: DevicePath) -> None:
+        ...
+
+    @abc.abstractmethod
+    def run_script(
+        self, content: AnyStr | StringIO | BytesIO, target_path: DevicePath | None = None
+    ):
         ...
diff --git a/micropy/pyd/pydevice.py b/micropy/pyd/pydevice.py
@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+from io import BytesIO, StringIO
 from pathlib import Path
-from typing import AnyStr, Type
+from typing import AnyStr, Generic, Optional, Type
 
 from .abc import (
+    AnyBackend,
     DevicePath,
     HostPath,
     MessageConsumer,
@@ -15,8 +17,8 @@
 from .consumers import ConsumerDelegate
 
 
-class PyDevice(MetaPyDevice):
-    pydevice: MetaPyDeviceBackend
+class PyDevice(MetaPyDevice[AnyBackend], Generic[AnyBackend]):
+    pydevice: AnyBackend
     consumer: ConsumerDelegate
 
     def __init__(
@@ -34,27 +36,59 @@ def __init__(
         if auto_connect and self.pydevice:
             self.pydevice.connect()
 
-    def copy_from(self, source_path: DevicePath, target_path: HostPath) -> None:
+    def copy_from(
+        self,
+        source_path: DevicePath,
+        target_path: HostPath,
+        *,
+        verify_integrity: bool = True,
+        exclude_integrity: Optional[set[str]] = None,
+    ) -> None:
         src_path = Path(str(source_path))
         # 'is_dir/file' only works on existing paths.
         if not src_path.suffix:
             return self.pydevice.copy_dir(
-                DevicePath(source_path), target_path, consumer=self.consumer
+                DevicePath(source_path),
+                target_path,
+                consumer=self.consumer,
+                verify_integrity=verify_integrity,
+                exclude_integrity=exclude_integrity,
             )
-        return self.pydevice.pull_file(DevicePath(source_path), target_path, consumer=self.consumer)
+        return self.pydevice.pull_file(
+            DevicePath(source_path),
+            target_path,
+            consumer=self.consumer,
+            verify_integrity=verify_integrity,
+        )
 
-    def copy_to(self, source_path: HostPath, target_path: DevicePath) -> None:
+    def copy_to(self, source_path: HostPath, target_path: DevicePath, **kwargs) -> None:
         src_path = Path(str(source_path))
         host_exists = src_path.exists()
         if (host_exists and src_path.is_dir()) or (not host_exists and not src_path.suffix):
             raise RuntimeError("Copying dirs to device is not yet supported!")
-        return self.pydevice.push_file(source_path, target_path, consumer=self.consumer)
+        return self.pydevice.push_file(source_path, target_path, consumer=self.consumer, **kwargs)
 
     def connect(self):
         return self.pydevice.connect()
 
     def disconnect(self):
         return self.pydevice.disconnect()
 
-    def run_script(self, content: AnyStr, target_path: DevicePath | None = None):
-        return self.pydevice.eval_script(content, target_path, consumer=self.consumer)
+    def run_script(
+        self, content: AnyStr | StringIO | BytesIO, target_path: DevicePath | None = None
+    ):
+        _content = (
+            content
+            if isinstance(
+                content,
+                (
+                    str,
+                    bytes,
+                ),
+            )
+            else content.read()
+        )
+        return self.pydevice.eval_script(_content, target_path, consumer=self.consumer)
+
+    def run(self, content: str) -> str | None:
+        return self.pydevice.eval(content, consumer=self.consumer)