A gRPC client implementation for firmware updates, handling connection management, firmware uploads, device reboot verification, and update validation.

Key Components

FirmwareUpdateClient Class

Located in service_client/fw_update.py, this class encapsulates all server communicatino logic including cnonection handling and version verificatoin.

Technical Implementation

gRPC Connection Management

def establish_connection(self) -> bool:
    """Create secure channel and initialize stubs"""
    self.logger.info(f"Connecting to {self.target_address}")
    
    for attempt in range(3):
        try:
            self.comm_channel = grpc.secure_channel(
                self.target_address,
                options=self.CHANNEL_OPTIONS
            )
            
            # Initialize service stubs
            self.fw_update_service = (
                firmware_pb2_grpc.FirmwareUpdateServiceStub(self.comm_channel)
            )
            
            # Verify channel readiness
            grpc.channel_ready_future(self.comm_channel).result(timeout=15)
            return self._verify_service_connection()
            
        except Exception as err:
            self.logger.warning(f"Connection attempt {attempt+1} failed: {err}")
            time.sleep(2)
    
    self.logger.error("Connection failed after maximum attempts")
    return False

Streaming File Transfer

def transfer_firmware(self, file_path: str, product_id: str, 
                     release_state: str, position: int = 0) -> bool:
    """Stream firmware file to server"""
    if not os.path.isfile(file_path):
        self.logger.error(f"File not found: {file_path}")
        return False

    file_metadata = self._get_file_metadata(file_path)
    
    try:
        def generate_chunks():
            # Send metadata first
            yield firmware_pb2.UpdateRequest(
                metadata=firmware_pb2.FileMetadata(
                    product_id=product_id,
                    release_state=release_state,
                    file_name=file_metadata.name,
                    file_hash=file_metadata.hash_value,
                    total_size=file_metadata.size
                )
            )
            
            # Stream file content
            with open(file_path, 'rb') as f:
                while chunk := f.read(self.CHUNK_SIZE):
                    yield firmware_pb2.UpdateRequest(
                        data_chunk=firmware_pb2.DataChunk(content=chunk)
                    )

        response = self.fw_update_service.TransferFirmware(
            generate_chunks(),
            timeout=300
        )
        return response.success
        
    except grpc.RpcError as e:
        self.logger.error(f"Transfer failed: {e.details()}")
        return False