Commit 0d926dba authored by Arnaud Blanchard's avatar Arnaud Blanchard

Binding with python

parent daf6f244
import os
import ctypes
from ctypes import *
from ctypes.util import find_library
class BlcDim(Structure):
_fields_ = [("length", c_size_t), ("step", c_size_t)]
class BlcChannel:
"""
blc_channel wrapper.
"""
# Bind blc_channel methods
lib_path = find_library("blc_channel")
lib = CDLL(lib_path)
lib.blc_channel_new.restype = c_void_p
lib.blc_channel_delete.argtypes = [ c_void_p ]
lib.blc_channel_get_data.argtypes = [ c_void_p ]
lib.blc_channel_get_data.restype = c_void_p
lib.blc_channel_get_type.argtypes = [ c_void_p ]
lib.blc_channel_get_type.restype = c_uint32
lib.blc_channel_get_format.argtypes = [ c_void_p ]
lib.blc_channel_get_format.restype = c_uint32
lib.blc_channel_get_dims_nb.argtypes= [ c_void_p ]
lib.blc_channel_get_dims_nb.restype = c_int
lib.blc_channel_get_dims.argtypes = [ c_void_p ]
lib.blc_channel_wait_new_data.argtypes = [ c_void_p ]
lib.blc_channel_wait_ack_data.argtypes = [ c_void_p ]
lib.blc_channel_post_new_data.argtypes = [ c_void_p ]
lib.blc_channel_post_ack_data.argtypes = [ c_void_p ]
types={ 'UIN8': c_uint8, 'INT8': c_int8, 'UI16': c_uint16, 'INT16': c_int16, 'UI32': c_uint32, 'IN32': c_int32, 'UI64': c_uint64, 'IN64': c_int64, 'FL32': c_float, 'FL64': c_double}
def __init__(
self, name, mode, data_type=None,format_type=None, dims=None
):
"""
Creates or reopens a given blc_channel.
:param name: Name of the blc_channel (with sync/async identifier). Example: "/data"
:param mode: RW mode. Example: os.O_RDWR
TODO
"""
self.name = name
self.mode = mode
self.type = data_type
self.format = format_type
self.dims = dims
if (dims): #Creates
dim_types = [c_int] * len(dims)
BlcChannel.lib.blc_channel_new.argtypes = [
c_char_p,
c_int,
c_uint32,
c_uint32,
c_int,
*dim_types
]
# Create a new channel
self.channel = BlcChannel.lib.blc_channel_new(
name.encode(),
mode,
int.from_bytes(data_type.encode(), "big"),
int.from_bytes(format_type.encode(), "big"),
len(dims),
*dims
)
else: #Open
BlcChannel.lib.blc_channel_new.argtypes = [
c_char_p,
c_int,
c_uint32,
c_uint32,
c_int,
c_int
]
self.channel = BlcChannel.lib.blc_channel_new(
name.encode(),
mode,
0,
0,
0,
0
)
self.type=BlcChannel.lib.blc_channel_get_type(self.channel).to_bytes(4, "big").decode()
self.format=BlcChannel.lib.blc_channel_get_format(self.channel).to_bytes(4, "big").decode()
dims_nb=BlcChannel.lib.blc_channel_get_dims_nb(self.channel)
BlcChannel.lib.blc_channel_get_dims.restype = POINTER(BlcDim * dims_nb)
blc_dims=BlcChannel.lib.blc_channel_get_dims(self.channel)
dims=[]
for i in range(dims_nb):
dims.append(blc_dims[0][i].length) #dims[i]=
# Specify types for channel
self.total_length=1
for length in dims:
self.total_length *= length
self.ctype = BlcChannel.types[self.type]
self.array = cast(BlcChannel.lib.blc_channel_get_data(self.channel), POINTER(self.ctype))
def wait_new_data(self):
BlcChannel.lib.blc_channel_wait_new_data(self.channel)
def wait_ack_data(self):
BlcChannel.lib.blc_channel_wait_ack_data(self.channel)
def post_new_data(self):
BlcChannel.lib.blc_channel_post_new_data(self.channel)
def post_ack_data(self):
BlcChannel.lib.blc_channel_post_ack_data(self.channel)
def __del__(self):
BlcChannel.lib.blc_channel_delete(self.channel)
@property
def list(self):
#cast it from
data_list = list(self.array[:self.total_length])
return data_list
if __name__ == "__main__":
# EXEMPLE
channel = BlcChannel("/python_test", os.O_RDWR, 'UIN8', 'RGB3', [3, 2])
#channel.wait_new_data() #This needs data to be named .data
channel_read = BlcChannel("/python_test", os.O_RDONLY)
print("type",channel_read.type, "format", channel_read.format, "length", channel_read.total_length)
print(channel_read.array[:channel.total_length])
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment