#
# Copyright (C) 2018 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
from vts.proto import AndroidSystemControlMessage_pb2 as ASysCtrlMsg
from vts.proto import VtsResourceControllerMessage_pb2 as ResControlMsg
from vts.proto import ComponentSpecificationMessage_pb2 as CompSpecMsg
from vts.utils.python.mirror import mirror_object
class ResourceFmqMirror(mirror_object.MirrorObject):
"""This is a class that mirrors FMQ resource allocated on the target side.
Attributes:
SUPPORTED_SCALAR_TYPES: set, contains all scalar types supported by FMQ.
If the type of FMQ is one of those, this class
prepares the write data from caller provided
Python data.
_client: VtsTcpClient, the TCP client instance.
_queue_id: int, used to identify the queue object on the target side.
_data_type: type of data in the queue.
_sync: bool, whether the queue is synchronized.
"""
SUPPORTED_SCALAR_TYPES = {
"uint8_t", "int8_t", "uint16_t", "int16_t", "uint32_t", "int32_t",
"uint64_t", "int64_t", "bool_t", "double_t"
}
def __init__(self, data_type, sync, client, queue_id=-1):
"""Initialize a FMQ mirror.
Args:
data_type: string, type of data in the queue
(e.g. "uint32_t", "int16_t").
sync: bool, whether queue is synchronized (only has one reader).
client: VtsTcpClient, specifies the session that this mirror use.
queue_id: int, identifies the queue on the target side.
Optional if caller initializes a new FMQ mirror.
"""
super(ResourceFmqMirror, self).__init__(client)
self._data_type = data_type
self._sync = sync
self._queue_id = queue_id
def _create(self, queue_id, queue_size, blocking, reset_pointers):
"""Initiate a fast message queue object on the target side.
This method registers a FMQ object on the target side, and stores
the queue_id in the class attribute.
Users should not directly call this method because it will overwrite
the original queue_id stored in the mirror object, leaving that
queue object out of reference.
Users should always call InitFmq() in mirror_tracker.py to obtain a
new queue object.
Args:
queue_id: int, identifies the message queue object on the target side.
queue_size: int, size of the queue.
blocking: bool, whether blocking is enabled in the queue.
reset_pointers: bool, whether to reset read/write pointers when
creating a message queue object based on an existing message queue.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_CREATE, queue_id)
request_msg.queue_size = queue_size
request_msg.blocking = blocking
request_msg.reset_pointers = reset_pointers
# Send and receive data.
fmq_response = self._client.SendFmqRequest(request_msg)
if fmq_response is not None and fmq_response.queue_id != -1:
self._queue_id = fmq_response.queue_id
else:
self._queue_id = -1
logging.error("Failed to create a new queue object.")
def read(self, data, data_size):
"""Initiate a non-blocking read request to FMQ driver.
Args:
data: list, data to be filled by this function. The list will
be emptied before the function starts to put read data into
it, which is consistent with the function behavior on the
target side.
data_size: int, length of data to read.
Returns:
bool, true if the operation succeeds,
false otherwise.
"""
# Prepare arguments.
del data[:]
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_READ, self._queue_id)
request_msg.read_data_size = data_size
# Send and receive data.
fmq_response = self._client.SendFmqRequest(request_msg)
if fmq_response is not None and fmq_response.success:
self._extractReadData(fmq_response, data)
return True
return False
# TODO: support long-form blocking read in the future when there is use case.
def readBlocking(self, data, data_size, time_out_nanos=0):
"""Initiate a blocking read request (short-form) to FMQ driver.
Args:
data: list, data to be filled by this function. The list will
be emptied before the function starts to put read data into
it, which is consistent with the function behavior on the
target side.
data_size: int, length of data to read.
time_out_nanos: int, wait time (in nanoseconds) when blocking.
The default value is 0 (no blocking).
Returns:
bool, true if the operation succeeds,
false otherwise.
"""
# Prepare arguments.
del data[:]
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_READ_BLOCKING, self._queue_id)
request_msg.read_data_size = data_size
request_msg.time_out_nanos = time_out_nanos
# Send and receive data.
fmq_response = self._client.SendFmqRequest(request_msg)
if fmq_response is not None and fmq_response.success:
self._extractReadData(fmq_response, data)
return True
return False
def write(self, data, data_size):
"""Initiate a non-blocking write request to FMQ driver.
Args:
data: list, data to be written.
data_size: int, length of data to write.
The function will only write data up until data_size,
i.e. extraneous data will be discarded.
Returns:
bool, true if the operation succeeds,
false otherwise.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_WRITE, self._queue_id)
prepare_result = self._prepareWriteData(request_msg, data[:data_size])
if not prepare_result:
# Prepare write data failure, error logged in _prepareWriteData().
return False
# Send and receive data.
fmq_response = self._client.SendFmqRequest(request_msg)
if fmq_response is not None:
return fmq_response.success
return False
# TODO: support long-form blocking write in the future when there is use case.
def writeBlocking(self, data, data_size, time_out_nanos=0):
"""Initiate a blocking write request (short-form) to FMQ driver.
Args:
data: list, data to be written.
data_size: int, length of data to write.
The function will only write data up until data_size,
i.e. extraneous data will be discarded.
time_out_nanos: int, wait time (in nanoseconds) when blocking.
The default value is 0 (no blocking).
Returns:
bool, true if the operation succeeds,
false otherwise.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_WRITE_BLOCKING, self._queue_id)
prepare_result = self._prepareWriteData(request_msg, data[:data_size])
if not prepare_result:
# Prepare write data failure, error logged in _prepareWriteData().
return False
request_msg.time_out_nanos = time_out_nanos
# Send and receive data.
fmq_response = self._client.SendFmqRequest(request_msg)
if fmq_response is not None:
return fmq_response.success
return False
def availableToWrite(self):
"""Get space available to write in the queue.
Returns:
int, number of slots available.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_AVAILABLE_WRITE, self._queue_id)
# Send and receive data.
return self._processUtilMethod(request_msg)
def availableToRead(self):
"""Get number of items available to read.
Returns:
int, number of items.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_AVAILABLE_READ, self._queue_id)
# Send and receive data.
return self._processUtilMethod(request_msg)
def getQuantumSize(self):
"""Get size of item in the queue.
Returns:
int, size of item.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_GET_QUANTUM_SIZE, self._queue_id)
# send and receive data
return self._processUtilMethod(request_msg)
def getQuantumCount(self):
"""Get number of items that fit in the queue.
Returns:
int, number of items.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_GET_QUANTUM_COUNT, self._queue_id)
# Send and receive data.
return self._processUtilMethod(request_msg)
def isValid(self):
"""Check if the queue is valid.
Returns:
bool, true if the queue is valid.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.FMQ_IS_VALID, self._queue_id)
# Send and receive data.
fmq_response = self._client.SendFmqRequest(request_msg)
if fmq_response is not None:
return fmq_response.success
return False
@property
def queueId(self):
"""Gets the id assigned from the target side.
Returns:
int, id of the queue.
"""
return self._queue_id
@property
def dataType(self):
"""Get the type of data of this FMQ mirror.
Returns:
string, type of data in the queue
"""
return self._data_type
@property
def sync(self):
"""Get the synchronization option of this FMQ mirror.
Returns:
bool, true if the queue is synchronized (only has one reader).
"""
return self._sync
def _createTemplateRequestMessage(self, operation, queue_id):
"""Creates a template FmqRequestMessage with common arguments among
all FMQ operations.
Args:
operation: FmqOp, fmq operations.
(see test/vts/proto/VtsResourceControllerMessage.proto).
queue_id: int, identifies the message queue object on target side.
Returns:
FmqRequestMessage, fmq request message.
(See test/vts/proto/VtsResourceControllerMessage.proto).
"""
request_msg = ResControlMsg.FmqRequestMessage()
request_msg.operation = operation
request_msg.data_type = self._data_type
request_msg.sync = self._sync
request_msg.queue_id = queue_id
return request_msg
def _prepareWriteData(self, request_msg, data):
"""Converts python list to repeated protobuf field.
If the type of data in the queue is a supported scalar, caller can
directly supply the python native value. Otherwise, caller needs to
supply a list of VariableSpecificationMessage.
Args:
request_msg: FmqRequestMessage, arguments for a FMQ operation
request.
data: VariableSpecificationMessage list or a list of scalar values.
If the type of FMQ is scalar type, caller can directly
specify the Python scalar data. Otherwise, caller has to
provide each item as VariableSpecificationMessage.
Returns:
bool, true if preparation succeeds, false otherwise.
This function can fail if caller doesn't provide a list of
VariableSpecificationMessage when type of data in the queue
is not a supported scalar type.
"""
for curr_value in data:
new_message = request_msg.write_data.add()
if isinstance(curr_value,
CompSpecMsg.VariableSpecificationMessage):
new_message.CopyFrom(curr_value)
elif self._data_type in self.SUPPORTED_SCALAR_TYPES:
new_message.type = CompSpecMsg.TYPE_SCALAR
new_message.scalar_type = self._data_type
setattr(new_message.scalar_value, self._data_type, curr_value)
else:
logging.error("Need to provide VariableSpecificationMessage " +
"if type of data in the queue is not a " +
"supported scalar type.")
return False
return True
def _extractReadData(self, response_msg, data):
"""Extracts read data from the response message returned by client.
Args:
response_msg: FmqResponseMessage, contains response from FMQ driver.
data: list, to be filled by this function. data buffer is provided
by caller, so this function will append every element to the
buffer.
"""
for item in response_msg.read_data:
data.append(self._client.GetPythonDataOfVariableSpecMsg(item))
def _processUtilMethod(self, request_msg):
"""Sends request message and process response message for util methods
that return an unsigned integer,
e.g. availableToWrite, availableToRead.
Args:
request_msg: FmqRequestMessage, arguments for a FMQ operation request.
Returns: int, information about the queue,
None if the operation is unsuccessful.
"""
fmq_response = self._client.SendFmqRequest(request_msg)
if fmq_response is not None and fmq_response.success:
return fmq_response.sizet_return_val
return None
class ResourceHidlMemoryMirror(mirror_object.MirrorObject):
"""This class mirrors hidl_memory resource allocated on the target side.
Attributes:
_client: the TCP client instance.
_mem_id: int, used to identify the memory region on the target side.
"""
def __init__(self, client, mem_id=-1):
super(ResourceHidlMemoryMirror, self).__init__(client)
self._mem_id = mem_id
def _allocate(self, mem_size):
"""Initiate a hidl_memory region on the target side.
This method stores the mem_id in the class attribute.
Users should not directly call this method to get a new memory region,
because it will overwrite the original memory object with mem_id,
making that memory object out of reference.
Users should always call InitHidlMemory() in mirror_tracker.py to get
a new memory region.
Args:
mem_size: int, size of the requested memory region.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_ALLOCATE)
request_msg.mem_size = mem_size
# Send and receive data.
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None and response_msg.new_mem_id != -1:
self._mem_id = response_msg.new_mem_id
else:
logging.error("Failed to allocate memory region.")
def read(self):
"""Notify that caller will read the entire memory region.
Before every actual read operation, caller must call this method
or readRange() first.
Returns:
bool, true if the operation succeeds, false otherwise.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_START_READ)
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if not response_msg.success:
logging.error("Failed to find memory region with id %d",
self._mem_id)
return response_msg.success
return False
def readRange(self, start, length):
"""Notify that caller will read only part of memory region.
Notify that caller will read starting at start and
ending at start + length.
Before every actual read operation, caller must call this method
or read() first.
Args:
start: int, offset from the start of memory region to be modified.
length: int, number of bytes to be modified.
Returns:
bool, true if the operation succeeds, false otherwise.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_START_READ_RANGE)
request_msg.start = start
request_msg.length = length
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if not response_msg.success:
logging.error("Failed to find memory region with id %d",
self._mem_id)
return response_msg.success
return False
def update(self):
"""Notify that caller will possibly write to all memory region.
Before every actual write operation, caller must call this method
or updateRange() first.
Returns:
bool, true if the operation succeeds, false otherwise.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_START_UPDATE)
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if not response_msg.success:
logging.error("Failed to find memory region with id %d",
self._mem_id)
return response_msg.success
return False
def updateRange(self, start, length):
"""Notify that caller will only write to part of memory region.
Notify that caller will only write starting at start and
ending at start + length.
Before every actual write operation, caller must call this method
or update() first.
Args:
start: int, offset from the start of memory region to be modified.
length: int, number of bytes to be modified.
Returns:
bool, true if the operation succeeds, false otherwise.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_START_UPDATE_RANGE)
request_msg.start = start
request_msg.length = length
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if not response_msg.success:
logging.error("Failed to find memory region with id %d",
self._mem_id)
return response_msg.success
return False
def readBytes(self, length, start=0):
"""This method performs actual read operation.
This method helps caller perform actual read operation on the
memory region, because host side won't be able to cast c++ pointers.
Args:
length: int, number of bytes to read.
start: int, offset from the start of memory region to read.
Returns:
string, data read from memory.
Caller can perform conversion on the result to obtain the
corresponding data structure in python.
None, indicate if the read fails.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_READ_BYTES)
request_msg.start = start
request_msg.length = length
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if response_msg.success:
return response_msg.read_data
logging.error("Failed to find memory region with id %d",
self._mem_id)
return None
def updateBytes(self, data, length, start=0):
"""This method performs actual write operation.
This method helps caller perform actual write operation on the
memory region, because host side won't be able to cast c++ pointers.
Args:
data: string, bytes to be written into memory.
Caller can use bytearray() function to convert python
data structures into python, and call str() on the resulting
bytearray object.
length: int, number of bytes to write.
start: int, offset from the start of memory region to be modified.
Returns:
bool, true if the operation succeeds, false otherwise.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_UPDATE_BYTES)
request_msg.write_data = data
request_msg.start = start
request_msg.length = length
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if not response_msg.success:
logging.error("Failed to find memory region with id %d",
self._mem_id)
return response_msg.success
return False
def commit(self):
"""Caller signals done with operating on the memory region.
Caller needs to call this method after reading/writing.
Returns:
bool, true if the operation succeeds, false otherwise.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_COMMIT)
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if not response_msg.success:
logging.error("Failed to find memory region with id %d",
self._mem_id)
return response_msg.success
return False
def getSize(self):
"""Gets the size of the memory region.
Returns:
int, size of memory region, -1 to signal operation failure.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.MEM_PROTO_GET_SIZE)
response_msg = self._client.SendHidlMemoryRequest(request_msg)
if response_msg is not None:
if response_msg.success:
return response_msg.mem_size
logging.error("Failed to find memory region with id %d",
self._mem_id)
return -1
@property
def memId(self):
"""Gets the id assigned from the target side.
Returns:
int, id of the memory object.
"""
return self._mem_id
def _createTemplateRequestMessage(self, operation):
"""Creates a template HidlMemoryRequestMessage.
This method creates a message that contains common arguments among
all hidl_memory operations.
Args:
operation: HidlMemoryOp, hidl_memory operations.
(see test/vts/proto/VtsResourceControllerMessage.proto).
Returns:
HidlMemoryRequestMessage, hidl_memory request message.
(See test/vts/proto/VtsResourceControllerMessage.proto).
"""
request_msg = ResControlMsg.HidlMemoryRequestMessage()
request_msg.operation = operation
request_msg.mem_id = self._mem_id
return request_msg
class ResourceHidlHandleMirror(mirror_object.MirrorObject):
"""This class mirrors hidl_handle resource allocated on the target side.
TODO: support more than file types in the future, e.g. socket, pipe.
Attributes:
_client: the TCP client instance.
_handle_id: int, used to identify the handle object on the target side.
"""
def __init__(self, client, handle_id=-1):
super(ResourceHidlHandleMirror, self).__init__(client)
self._handle_id = handle_id
def CleanUp(self):
"""Close open file descriptors on target-side drivers.
Developers can call this method to close open file descriptors
in all handle objects.
Note: This method needs to be called before self._client
is disconnected. self._client is most likely initialized in
one of the hal_mirror.
"""
request_msg = self._createTemplateRequestMessage(
ResControlMsg.HANDLE_PROTO_DELETE)
self._client.SendHidlHandleRequest(request_msg)
def _createHandleForSingleFile(self, filepath, mode, int_data):
"""Initiate a hidl_handle object containing a single file descriptor.
This method stores the handle_id in the class attribute.
Users should not directly call this method to create a new
handle object, because it will overwrite the original handle object,
making that handle object out of reference.
Users should always call InitHidlHandle() in mirror_tracker.py to get
a new handle object.
Args:
filepath: string, path to the file to be opened.
mode: string, specifying the mode to open the file.
int_data: int list, useful integers to store in the handle object.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.HANDLE_PROTO_CREATE_FILE)
request_msg.handle_info.num_fds = 1
request_msg.handle_info.num_ints = len(int_data)
# TODO: support more than one file descriptors at once.
# Add the file information into proto message.
fd_message = request_msg.handle_info.fd_val.add()
fd_message.type = CompSpecMsg.FILE_TYPE
fd_message.file_mode_str = mode
fd_message.file_name = filepath
# Add the integers into proto message.
request_msg.handle_info.int_val.extend(int_data)
# Send and receive data.
response_msg = self._client.SendHidlHandleRequest(request_msg)
if response_msg is not None and response_msg.new_handle_id != -1:
self._handle_id = response_msg.new_handle_id
else:
logging.error("Failed to create handle object.")
def readFile(self, read_data_size, index=0):
"""Reads from a given file in the handle object.
Args:
read_data_size: int, number of bytes to read.
index: int, index of file among all files in the handle object.
Optional if host only wants to read from one file.
Returns:
string, data read from the file.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.HANDLE_PROTO_READ_FILE)
request_msg.read_data_size = read_data_size
# Send and receive data.
response_msg = self._client.SendHidlHandleRequest(request_msg)
if response_msg is not None and response_msg.success:
return response_msg.read_data
# TODO: more detailed error message.
logging.error("Failed to read from the file.")
return None
def writeFile(self, write_data, index=0):
"""Writes to a given file to the handle object.
Args:
write_data: string, data to be written into file.
index: int, index of file among all files in the handle object.
Optional if host only wants to write into one file.
Returns:
int, number of bytes written.
"""
# Prepare arguments.
request_msg = self._createTemplateRequestMessage(
ResControlMsg.HANDLE_PROTO_WRITE_FILE)
request_msg.write_data = write_data
# Send and receive data.
response_msg = self._client.SendHidlHandleRequest(request_msg)
if response_msg is not None and response_msg.success:
return response_msg.write_data_size
# TODO: more detailed error message.
logging.error("Failed to write into the file.")
return None
@property
def handleId(self):
"""Gets the id assigned from the target side.
Returns:
int, id of the handle object.
"""
return self._handle_id
def _createTemplateRequestMessage(self, operation):
"""Creates a template HidlHandleRequestMessage.
This method creates a HidlHandleRequestMessage with common arguments
among all hidl_handle operations.
Args:
operation: HidlHandleOp, hidl_handle operations.
(see test/vts/proto/VtsResourceControllerMessage.proto).
Returns:
HidlHandleRequestMessage, hidl_handle request message.
(See test/vts/proto/VtsResourceControllerMessage.proto).
"""
request_msg = ResControlMsg.HidlHandleRequestMessage()
request_msg.operation = operation
request_msg.handle_id = self._handle_id
return request_msg