Save work in progress...

picowatch_d/picowatch.py

@@ -26,89 +26,97 @@ import os
 import sys
 import time
 import json
-import serial
 import signal
 import binascii
 import textwrap
 
+from serial import Serial
 from watchdog.observers import Observer
 from watchdog.events import PatternMatchingEventHandler
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple, Union
 
 
 BUFFER_SIZE: int = 256
 
 
-# class TelnetToSerial:
+class Telnet:
-#     def __init__(self, ip, user, password, read_timeout=None):
-#         import telnetlib
-#         self.tn = telnetlib.Telnet(ip, timeout=15)
-#         self.read_timeout = read_timeout
-#         if b'Login as:' in self.tn.read_until(b'Login as:', timeout=read_timeout):
-#             self.tn.write(bytes(user, 'ascii') + b"\r\n")
 
-#             if b'Password:' in self.tn.read_until(b'Password:', timeout=read_timeout):
+    def __init__(self, IP: str, login: str, password: str):
-#                 # needed because of internal implementation details of the telnet server
+        import telnetlib
-#                 time.sleep(0.2)
+        from collections import deque
-#                 self.tn.write(bytes(password, 'ascii') + b"\r\n")
 
-#                 if b'for more information.' in self.tn.read_until(b'Type "help()" for more information.', timeout=read_timeout):
+        self.tn = telnetlib.Telnet(IP, timeout=15)
-#                     # login succesful
+        self.fifo = deque()
-#                     from collections import deque
-#                     self.fifo = deque()
-#                     return
 
-#         raise Exception('Failed to establish a telnet connection with the board')
+        if b'Login as:' in self.tn.read_until(b'Login as:'):
+            self.tn.write(bytes(login, 'ascii') + b'\r\n')
 
-#     def __del__(self):
+            if b'Password:' in self.tn.read_until(b'Password:'):
-#         self.close()
+                time.sleep(0.2)
+                self.tn.write(bytes(password, 'ascii') + b'\r\n')
 
-#     def close(self):
+                if b'for more information.' in self.tn.read_until(b'Type "help()" for more information.'):
-#         try:
+                    return
-#             self.tn.close()
-#         except:
-#             # the telnet object might not exist yet, so ignore this one
-#             pass
 
-#     def read(self, size=1):
+        raise Exception('Failed to establish a Telnet connection with the board')
-#         while len(self.fifo) < size:
-#             timeout_count = 0
-#             data = self.tn.read_eager()
-#             if len(data):
-#                 self.fifo.extend(data)
-#                 timeout_count = 0
-#             else:
-#                 time.sleep(0.25)
-#                 if self.read_timeout is not None and timeout_count > 4 * self.read_timeout:
-#                     break
-#                 timeout_count += 1
 
-#         data = b''
+    def __del__(self):
-#         while len(data) < size and len(self.fifo) > 0:
+        self.close()
-#             data += bytes([self.fifo.popleft()])
-#         return data
 
-#     def write(self, data):
+    def close(self):
-#         self.tn.write(data)
+        try:
-#         return len(data)
+            self.tn.close()
+        except:
+            pass
 
-#     def in_waiting(self):
+    def read(self, size: int = 1) -> bytes:
-#         n_waiting = len(self.fifo)
+        timeout = 0
-#         if not n_waiting:
+
-#             data = self.tn.read_eager()
+        while len(self.fifo) < size and not timeout >= 8:
-#             self.fifo.extend(data)
+            timeout = 0
-#             return len(data)
+            data = self.tn.read_eager()
-#         else:
+
-#             return n_waiting
+            if len(data):
+                self.fifo.extend(data)
+                timeout = 0
+            else:
+                timeout += 1
+                time.sleep(0.25)
+
+        data = b''
+
+        while len(data) < size and len(self.fifo) > 0:
+            data += bytes([self.fifo.popleft()])
+
+        return data
+
+    def write(self, data: bytes):
+        self.tn.write(data)
+        return len(data)
+
+    def inWaiting(self) -> int:
+        n_waiting = len(self.fifo)
+
+        if not n_waiting:
+            data = self.tn.read_eager()
+            self.fifo.extend(data)
+            n_waiting = len(data)
+        
+        return n_waiting
 
 
 class Pyboard(object):
-    serial: serial.Serial
+    serial: Union[Serial, Telnet]
+
+    def __init__(self, device: str, baudrate: int = 115200, login: str = '', password: str = ''):
+        is_telnet = device and device.count('.') == 3
 
-    def __init__(self, device: str, baudrate: int = 115200):
         for _ in range(0, 3):
             try:
-                self.serial = serial.Serial(device, baudrate=baudrate, interCharTimeout=1)
+                if is_telnet:
+                    self.serial = Telnet(device, login, password)
+                else:
+                    self.serial = Serial(device, baudrate=baudrate, interCharTimeout=1)
                 break
             except:
                 time.sleep(1)
@@ -530,12 +538,12 @@ class Picowatch(object):
     def upload(self, filepath: str):
         local = filepath
 
+        if local.startswith('/'):
+            local = '.' + local
+
         if not local.startswith('./'):
             local = './' + local
 
-        if not local.startswith('.'):
-            local = '.' + local
-
         queue = []
 
         if os.path.isdir(local):