pw_transfer: Minor python improvements

- Allows the data/parameter chunk processing to ignore handshake chunks
   - transfer can sometimes recieve duplicate handshake chunks due to
     timeouts
- Allows individual transfers to set timeouts on python clients

Change-Id: Ia3c834f03c0007c8dd698cadfda74e04c30ffeea
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/184152
Reviewed-by: Alexei Frolov <frolv@google.com>
Commit-Queue: Jordan Brauer <jtbrauer@google.com>

pw_transfer/py/pw_transfer/client.py

@@ -120,6 +120,8 @@ def read(
         resource_id: int,
         progress_callback: Optional[ProgressCallback] = None,
         protocol_version: Optional[ProtocolVersion] = None,
+        chunk_timeout_s: Optional[float] = None,
+        initial_timeout_s: Optional[float] = None,
     ) -> bytes:
         """Receives ("downloads") data from the server.
 
@@ -147,13 +149,19 @@ def read(
             else self.assign_session_id()
         )
 
+        if chunk_timeout_s is None:
+            chunk_timeout_s = self._default_response_timeout_s
+
+        if initial_timeout_s is None:
+            initial_timeout_s = self._initial_response_timeout_s
+
         transfer = ReadTransfer(
             session_id,
             resource_id,
             self._send_read_chunk,
             self._end_read_transfer,
-            self._default_response_timeout_s,
-            self._initial_response_timeout_s,
+            chunk_timeout_s,
+            initial_timeout_s,
             self.max_retries,
             self.max_lifetime_retries,
             protocol_version,
@@ -174,6 +182,8 @@ def write(
         data: Union[bytes, str],
         progress_callback: Optional[ProgressCallback] = None,
         protocol_version: Optional[ProtocolVersion] = None,
+        chunk_timeout_s: Optional[Any] = None,
+        initial_timeout_s: Optional[Any] = None,
     ) -> None:
         """Transmits ("uploads") data to the server.
 
@@ -205,14 +215,20 @@ def write(
             else self.assign_session_id()
         )
 
+        if chunk_timeout_s is None:
+            chunk_timeout_s = self._default_response_timeout_s
+
+        if initial_timeout_s is None:
+            initial_timeout_s = self._initial_response_timeout_s
+
         transfer = WriteTransfer(
             session_id,
             resource_id,
             data,
             self._send_write_chunk,
             self._end_write_transfer,
-            self._default_response_timeout_s,
-            self._initial_response_timeout_s,
+            chunk_timeout_s,
+            initial_timeout_s,
             self.max_retries,
             self.max_lifetime_retries,
             protocol_version,

pw_transfer/py/pw_transfer/transfer.py

@@ -255,8 +255,12 @@ async def handle_chunk(self, chunk: Chunk) -> None:
                 # Expecting a completion ACK but didn't receive one. Go through
                 # the retry process.
                 self._on_timeout()
-        else:
+        # Only ignoring START_ACK, tests were unhappy with other non-data chunks
+        elif chunk.type not in [Chunk.Type.START_ACK]:
             await self._handle_data_chunk(chunk)
+        else:
+            _LOG.warning("Ignoring extra START_ACK chunk")
+            return
 
         # Start the timeout for the server to send a chunk in response.
         self._response_timer.start()

pw_transfer/py/tests/transfer_test.py

@@ -417,7 +417,8 @@ def test_read_transfer_server_error(self) -> None:
 
     def test_write_transfer_basic(self) -> None:
         manager = pw_transfer.Manager(
-            self._service, default_response_timeout_s=DEFAULT_TIMEOUT_S
+            self._service,
+            default_response_timeout_s=DEFAULT_TIMEOUT_S,
         )
 
         self._enqueue_server_responses(