Refactoring ...
This commit is contained in:
1
python/obitools3/dms/column/__init__.py
Normal file
1
python/obitools3/dms/column/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from .column import column # @UnresolvedImport
|
56
python/obitools3/dms/column/bool.pxd
Normal file
56
python/obitools3/dms/column/bool.pxd
Normal file
@ -0,0 +1,56 @@
|
||||
#cython: language_level=3
|
||||
|
||||
from cpython.bool cimport bool, PyBool_FromLong
|
||||
|
||||
from ..capi.obitypes cimport index_t, \
|
||||
const_char_p, \
|
||||
OBIType_t, \
|
||||
obibool_t, \
|
||||
OBI_BOOL, \
|
||||
OBIBool_NA
|
||||
|
||||
from ..capi.obiview cimport obi_get_bool_with_elt_name_and_col_p_in_view, \
|
||||
obi_get_bool_with_elt_idx_and_col_p_in_view, \
|
||||
obi_set_bool_with_elt_name_and_col_p_in_view, \
|
||||
obi_set_bool_with_elt_idx_and_col_p_in_view
|
||||
|
||||
from ..capi.obidmscolumn cimport obi_column_get_obibool_with_elt_name, \
|
||||
obi_column_get_obibool_with_elt_idx, \
|
||||
obi_column_set_obibool_with_elt_name, \
|
||||
obi_column_set_obibool_with_elt_idx, \
|
||||
OBIDMS_column_p
|
||||
|
||||
from ..capi.obierrno cimport obi_errno
|
||||
|
||||
from .column cimport Column, \
|
||||
Column_line, \
|
||||
register_column_class
|
||||
|
||||
from obitools3.utils cimport str2bytes, bytes2str
|
||||
|
||||
cdef class Column_line_bool(OBIDMS_column_line) :
|
||||
|
||||
@staticmethod
|
||||
cdef bool obibool_t2bool(obibool_t value)
|
||||
|
||||
@staticmethod
|
||||
cdef bool2obibool_t(bool value)
|
||||
|
||||
cpdef bool get_bool_item_by_name(self,bytes element_name)
|
||||
cpdef bool get_bool_item_by_idx(self,index_t index)
|
||||
cpdef set_bool_item_by_name(self,bytes element_name,bool value)
|
||||
cpdef set_bool_item_by_idx(self,index_t index,bool value)
|
||||
|
||||
|
||||
# cdef obibool_t [:] _data_view
|
||||
|
||||
cdef class Column_bool(OBIDMS_column):
|
||||
cdef Column _new(OBIView view,
|
||||
bytes column_name,
|
||||
index_t nb_elements_per_line=1,
|
||||
object elements_names=None,
|
||||
bytes comments=b""):
|
||||
|
||||
cpdef object get_line(self, index_t line_nb)
|
||||
cpdef set_line(self, index_t line_nb, object value)
|
||||
|
294
python/obitools3/dms/column/bool.pyx
Normal file
294
python/obitools3/dms/column/bool.pyx
Normal file
@ -0,0 +1,294 @@
|
||||
#cython: language_level=3
|
||||
|
||||
|
||||
cdef class OBIDMS_column_line_bool(OBIDMS_column_line) :
|
||||
|
||||
cdef update_pointer(self):
|
||||
"""
|
||||
Checks if the obicolumn address changed since the last call and update
|
||||
if need the `_column_p` and `_data_view` data structure fields.
|
||||
"""
|
||||
cdef OBIDMS_column_p column_p = self._column_pp[0]
|
||||
|
||||
if column_p != self._column_p:
|
||||
self._column_p = column_p
|
||||
# self._data_view = (<obibool_t*> (column_p.data)) + \
|
||||
# self._index * column_p.header.nb_elements_per_line
|
||||
|
||||
@staticmethod
|
||||
cdef bool obibool_t2bool(obibool_t value):
|
||||
cdef bool result
|
||||
|
||||
if value == OBIBool_NA :
|
||||
result = None
|
||||
else :
|
||||
result = PyBool_FromLong(value)
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
cdef bool2obibool_t(bool value):
|
||||
cdef obibool_t result
|
||||
|
||||
if value is None:
|
||||
result=OBIBool_NA
|
||||
else:
|
||||
result= <obibool_t> <int> value
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def __init__(self, OBIDMS_column column, index_t line_nb) :
|
||||
"""
|
||||
Creates a new `OBIDMS_column_line_bool`
|
||||
|
||||
@param column: an OBIDMS_column instance
|
||||
@param line_nb: the line in the column
|
||||
"""
|
||||
|
||||
OBIDMS_column_line.__init__(self,column,line_nb)
|
||||
self.update_pointer()
|
||||
|
||||
|
||||
|
||||
cpdef bool get_bool_item_by_name(self,bytes element_name):
|
||||
"""
|
||||
Returns the value associated to the name `element_name` of the current line
|
||||
|
||||
@param element_name: a `bytes` instance containing the name of the element
|
||||
|
||||
@return: the `bool` value corresponding to the name
|
||||
"""
|
||||
cdef char* cname = element_name
|
||||
cdef obibool_t value
|
||||
global obi_errno
|
||||
|
||||
self.update_pointer()
|
||||
|
||||
value = obi_column_get_obibool_with_elt_name(self._column_p,
|
||||
self._index,
|
||||
cname)
|
||||
|
||||
if obi_errno > 0 :
|
||||
obi_errno = 0
|
||||
raise KeyError("Cannot access to key %s" % bytes2str(element_name))
|
||||
|
||||
return OBIDMS_column_line_bool.obibool_t2bool(value)
|
||||
|
||||
cpdef bool get_bool_item_by_idx(self,index_t index):
|
||||
"""
|
||||
Returns the value associated to the name `element_name` of the current line
|
||||
|
||||
@param index: a `int` instance containing the index of the element
|
||||
|
||||
@return: the `bool` value corresponding to the name
|
||||
"""
|
||||
cdef obibool_t value # @DuplicatedSignature
|
||||
global obi_errno
|
||||
|
||||
|
||||
self.update_pointer()
|
||||
|
||||
value = obi_column_get_obibool_with_elt_idx(self._column_p,
|
||||
self._index,
|
||||
index)
|
||||
|
||||
if obi_errno > 0 :
|
||||
obi_errno = 0
|
||||
raise IndexError("Cannot access to element %d" % index)
|
||||
|
||||
return OBIDMS_column_line_bool.obibool_t2bool(value)
|
||||
|
||||
|
||||
def __getitem__(self, object element_name) :
|
||||
cdef bytes name
|
||||
cdef int cindex
|
||||
cdef obibool_t value
|
||||
cdef type typearg = type(element_name)
|
||||
cdef bool result
|
||||
|
||||
|
||||
if typearg == int:
|
||||
cindex=element_name
|
||||
if cindex < 0:
|
||||
cindex = self._len - cindex
|
||||
result=self.get_bool_item_by_idx(cindex)
|
||||
elif typearg == bytes:
|
||||
result=self.get_bool_item_by_name(element_name)
|
||||
elif typearg == str:
|
||||
name = str2bytes(element_name)
|
||||
result=self.get_bool_item_by_name(name)
|
||||
|
||||
return result
|
||||
|
||||
cpdef set_bool_item_by_name(self,bytes element_name,bool value):
|
||||
"""
|
||||
Sets the value associated to the name `element_name` of the current line
|
||||
|
||||
@param element_name: a `bytes` instance containing the name of the element
|
||||
@param value: a `bool` instance of the new value
|
||||
|
||||
@return: the `bool` value corresponding to the name
|
||||
"""
|
||||
cdef char* cname = element_name
|
||||
cdef obibool_t cvalue
|
||||
|
||||
self.update_pointer()
|
||||
cvalue = OBIDMS_column_line_bool.bool2obibool_t(value)
|
||||
|
||||
if ( obi_column_set_obibool_with_elt_name(self._column_p,
|
||||
self._index,
|
||||
cname,
|
||||
cvalue) < 0 ):
|
||||
raise KeyError("Cannot access to key %s" % bytes2str(element_name))
|
||||
|
||||
cpdef set_bool_item_by_idx(self,index_t index,bool value):
|
||||
"""
|
||||
Sets the value associated to the name `element_name` of the current line
|
||||
|
||||
@param index: a `int` instance containing the index of the element
|
||||
@param value: a `bool` instance of the new value
|
||||
|
||||
@return: the `bool` value corresponding to the name
|
||||
"""
|
||||
cdef obibool_t cvalue # @DuplicatedSignature
|
||||
|
||||
self.update_pointer()
|
||||
cvalue = OBIDMS_column_line_bool.bool2obibool_t(value)
|
||||
|
||||
if ( obi_column_set_obibool_with_elt_idx(self._column_p,
|
||||
self._index,
|
||||
index,
|
||||
cvalue) < 0 ):
|
||||
raise IndexError("Cannot access to item index %d" % index)
|
||||
|
||||
|
||||
|
||||
def __setitem__(self, object element_name, object value):
|
||||
cdef bytes name
|
||||
cdef int cindex
|
||||
cdef type typearg = type(element_name)
|
||||
cdef bool result
|
||||
|
||||
|
||||
if typearg == int:
|
||||
cindex=element_name
|
||||
if cindex < 0:
|
||||
cindex = self._len - cindex
|
||||
self.set_bool_item_by_idx(cindex,value)
|
||||
elif typearg == bytes:
|
||||
self.set_bool_item_by_name(element_name,value)
|
||||
elif typearg == str:
|
||||
name = str2bytes(element_name)
|
||||
self.set_bool_item_by_name(name,value)
|
||||
|
||||
def __repr__(self) :
|
||||
return str(self._column.get_line(self._index))
|
||||
|
||||
def __len__(self):
|
||||
return self._len
|
||||
|
||||
|
||||
|
||||
cdef class OBIDMS_column_bool(OBIDMS_column):
|
||||
|
||||
@staticmethod
|
||||
cdef OBIDMS_column _new(OBIView view,
|
||||
bytes column_name,
|
||||
index_t nb_elements_per_line=1,
|
||||
object elements_names=None,
|
||||
bytes comments=b""):
|
||||
|
||||
cdef bytes elements_names_b
|
||||
cdef char* elements_names_p
|
||||
cdef OBIDMS_column new_column
|
||||
|
||||
if elements_names is not None:
|
||||
elements_names_b = b''.join([tobytes(x) for x in elements_names])
|
||||
elements_names_p = elements_names_b
|
||||
else:
|
||||
elements_names_p = NULL
|
||||
|
||||
if (obi_view_add_column(view = view._pointer,
|
||||
column_name = column_name,
|
||||
version_number = -1,
|
||||
alias = NULL,
|
||||
data_type = OBI_BOOL,
|
||||
nb_lines = len(view),
|
||||
nb_elements_per_line = nb_elements_per_line,
|
||||
elements_names = elements_names_p,
|
||||
indexer_name = NULL,
|
||||
associated_column_name = NULL,
|
||||
associated_column_version = -1,
|
||||
comments = comments,
|
||||
create = True)<0):
|
||||
raise RuntimeError("Cannot create column %s in view %s" % (bytes2str(column_name),
|
||||
bytes2str(view.name)))
|
||||
|
||||
view.__init_columns__()
|
||||
new_column = self._columns[column_name]
|
||||
|
||||
return new_column
|
||||
|
||||
@staticmethod
|
||||
def new(OBIView view,
|
||||
object column_name,
|
||||
index_t nb_elements_per_line=1,
|
||||
object elements_names=None,
|
||||
object comments=b""):
|
||||
|
||||
return OBIDMS_column_bool._new(view,
|
||||
tobytes(column_name),
|
||||
nb_elements_per_line,
|
||||
elements_names,
|
||||
tobytes(comments))
|
||||
|
||||
cpdef add_to_view(self,
|
||||
OBIView view,
|
||||
object column_name=None,
|
||||
object comments=b""):
|
||||
|
||||
cdef OBIDMS_column_p column_p = self._column_pp[0]
|
||||
cdef bytes alias
|
||||
|
||||
if (column_name is None):
|
||||
alias = self._alias
|
||||
else:
|
||||
alias = tobytes(column_name)
|
||||
|
||||
obi_view_add_column(view = view._ponter,
|
||||
column_name = column_p.header.name,
|
||||
version_number = column_p.header.version,
|
||||
alias = alias,
|
||||
data_type = OBI_BOOL,
|
||||
nb_lines = column_p.header.lines_used,
|
||||
nb_elements_per_line = column_p.header.nb_elements_per_line,
|
||||
elements_names = column_p.header.elements_names,
|
||||
indexer_name = NULL,
|
||||
associated_column_name = NULL,
|
||||
associated_column_version = NULL,
|
||||
comments = tobytes(comments),
|
||||
create = False)
|
||||
|
||||
cpdef object get_line(self, index_t line_nb):
|
||||
cdef obibool_t value
|
||||
cdef object result
|
||||
global obi_errno
|
||||
|
||||
value = obi_get_bool_with_elt_idx_and_col_p_in_view(self._view._pointer, (self._pointer)[0], line_nb, 0)
|
||||
if obi_errno > 0 :
|
||||
raise IndexError(line_nb)
|
||||
if value == OBIBool_NA :
|
||||
result = None
|
||||
else :
|
||||
result = PyBool_FromLong(value)
|
||||
return result
|
||||
|
||||
cpdef set_line(self, index_t line_nb, object value):
|
||||
if value is None :
|
||||
value = OBIBool_NA
|
||||
if obi_set_bool_with_elt_idx_and_col_p_in_view(self._view._pointer, (self._pointer)[0], line_nb, 0, <obibool_t> value) < 0:
|
||||
raise Exception("Problem setting a value in a column")
|
||||
|
||||
|
||||
register_column_class(OBI_BOOL,OBIDMS_column_bool)
|
25
python/obitools3/dms/column/column.pxd
Normal file
25
python/obitools3/dms/column/column.pxd
Normal file
@ -0,0 +1,25 @@
|
||||
#cython: language_level=3
|
||||
|
||||
from .capi.obitypes cimport index_t, \
|
||||
obitype_t
|
||||
from .capi.obidmscolumn cimport OBIDMS_column_p
|
||||
|
||||
from .view cimport View
|
||||
|
||||
cdef class Column:
|
||||
|
||||
cdef OBIDMS_column_p* _pointer
|
||||
cdef View _view
|
||||
|
||||
cpdef close(self)
|
||||
|
||||
cdef class Column_line:
|
||||
|
||||
cdef Column _column
|
||||
cdef index_t _index
|
||||
|
||||
cpdef update(self, data)
|
||||
|
||||
cdef register_column_class(obitype_t obitype,
|
||||
type classe,
|
||||
type python)
|
180
python/obitools3/dms/column/column.pyx
Normal file
180
python/obitools3/dms/column/column.pyx
Normal file
@ -0,0 +1,180 @@
|
||||
#cython: language_level=3
|
||||
|
||||
from .capi.obitypes cimport name_data_type
|
||||
|
||||
from .capi.obidmscolumn cimport OBIDMS_column_header_p, \
|
||||
obi_close_column, \
|
||||
obi_column_prepare_to_get_value
|
||||
|
||||
from .capi.obiutils cimport obi_format_date
|
||||
|
||||
from .dms cimport __OBIDMS_COLUMN_CLASS__
|
||||
|
||||
from obitools3.utils cimport bytes2str
|
||||
|
||||
|
||||
cdef class Column :
|
||||
"""
|
||||
The obitools3.dms.column.Column class wraps a c instance of a column in the context of a View
|
||||
"""
|
||||
|
||||
# Note: should only be initialized through a subclass
|
||||
def __init__(self,
|
||||
View view,
|
||||
int __internalCall__):
|
||||
'''
|
||||
Create a new OBDMS column objected referring to a already created column
|
||||
in the context of a view.
|
||||
|
||||
This constructor is normally only called by subclass constructor.
|
||||
|
||||
@param view: The view object containing the column.
|
||||
@type view: OBIView
|
||||
'''
|
||||
|
||||
cdef OBIDMS_column_p* column_pp
|
||||
|
||||
if __internalCall__!=987654:
|
||||
raise RuntimeError('OBIView constructor cannot be called directly')
|
||||
|
||||
# Check that the class is only created as a subclass instance
|
||||
if type(self)==Column or not isinstance(self, Column):
|
||||
raise RuntimeError('OBIDMS.Column constructor cannot be called directly')
|
||||
|
||||
# Fill structure
|
||||
self._pointer = NULL
|
||||
self._view = view
|
||||
|
||||
def __len__(self):
|
||||
'''
|
||||
implements the len() function for the Column class
|
||||
|
||||
@rtype: `int`
|
||||
'''
|
||||
return self.lines_used
|
||||
|
||||
def __sizeof__(self):
|
||||
'''
|
||||
returns the size of the C object wrapped by the Column instance
|
||||
'''
|
||||
cdef OBIDMS_column_header_p header = self._pointer[0].header
|
||||
return header.header_size + header.data_size
|
||||
|
||||
def __iter__(self):
|
||||
cdef index_t line_nb
|
||||
|
||||
for line_nb in range(self.lines_used):
|
||||
yield self[line_nb]
|
||||
|
||||
def __str__(self) :
|
||||
cdef str to_print
|
||||
to_print = ''
|
||||
for line in self :
|
||||
to_print = to_print + str(line) + "\n"
|
||||
return to_print
|
||||
|
||||
def __repr__(self) :
|
||||
return b"%s, original name: %s, version %d, data type: %d" % (
|
||||
self._alias,
|
||||
self.original_name,
|
||||
self.version,
|
||||
self.data_type
|
||||
)
|
||||
|
||||
cpdef close(self):
|
||||
if self._pointer != NULL:
|
||||
if obi_close_column(self._pointer[0]) < 0 :
|
||||
raise Exception("Problem closing column %s" % bytes2str(self.name))
|
||||
|
||||
# Column alias property getter and setter
|
||||
@property
|
||||
def name(self):
|
||||
return self._alias
|
||||
@name.setter
|
||||
def name(self, new_alias): # @DuplicatedSignature
|
||||
self._view.change_column_alias(self._alias, new_alias)
|
||||
|
||||
# elements_names property getter
|
||||
@property
|
||||
def elements_names(self):
|
||||
return (((self._pointer)[0].header).elements_names).split(b';')
|
||||
|
||||
# nb_elements_per_line property getter
|
||||
@property
|
||||
def nb_elements_per_line(self):
|
||||
return ((self._pointer)[0].header).nb_elements_per_line
|
||||
|
||||
# data_type property getter
|
||||
@property
|
||||
def data_type(self):
|
||||
return name_data_type(((self._pointer)[0].header).returned_data_type)
|
||||
|
||||
# original_name property getter
|
||||
@property
|
||||
def original_name(self):
|
||||
return ((self._pointer)[0].header).name
|
||||
|
||||
# version property getter
|
||||
@property
|
||||
def version(self):
|
||||
return ((self._pointer)[0].header).version
|
||||
|
||||
# lines_used property getter
|
||||
@property
|
||||
def lines_used(self):
|
||||
return (self._pointer)[0].header.lines_used
|
||||
|
||||
# comments property getter
|
||||
@property
|
||||
def comments(self):
|
||||
return (self._pointer)[0].header.comments
|
||||
|
||||
# creation_date property getter
|
||||
@property
|
||||
def creation_date(self):
|
||||
return obi_format_date((self._pointer)[0].header.creation_date)
|
||||
|
||||
|
||||
######################################################################################################
|
||||
|
||||
|
||||
cdef class Column_line :
|
||||
|
||||
def __init__(self, Column column, index_t line_nb) :
|
||||
self._index = line_nb
|
||||
self._column = column
|
||||
|
||||
if obi_column_prepare_to_get_value(self._column._pointer[0],line_nb) < 0:
|
||||
raise IndexError("Cannot access to the line %d" % line_nb)
|
||||
|
||||
|
||||
def __contains__(self, str element_name):
|
||||
pass
|
||||
#return (element_name in self._column.elements_names)
|
||||
|
||||
def __repr__(self) :
|
||||
return str(self._column.get_line(self._index))
|
||||
|
||||
cpdef update(self, data):
|
||||
if isinstance(data, dict):
|
||||
data=data.items()
|
||||
for key,value in data:
|
||||
if key in self:
|
||||
self[key]=value
|
||||
|
||||
|
||||
######################################################################################################
|
||||
|
||||
|
||||
cdef register_column_class(obitype_t obitype,
|
||||
type classe,
|
||||
type python):
|
||||
"""
|
||||
Each sub class of `OBIDMS_column` needs to be registered after its declaration
|
||||
to declare its relationship with an `OBIType_t`
|
||||
"""
|
||||
global __OBIDMS_COLUMN_CLASS__
|
||||
|
||||
assert issubclass(classe,Column)
|
||||
|
||||
__OBIDMS_COLUMN_CLASS__[obitype]=(classe,python)
|
@ -1,6 +1,7 @@
|
||||
#cython: language_level=3
|
||||
|
||||
from libc.stdlib cimport malloc,free
|
||||
from libc.stdlib cimport malloc,free
|
||||
from cpython.list cimport PyList_Size
|
||||
|
||||
from .capi.obidms cimport obi_dms, \
|
||||
obi_close_dms, \
|
||||
|
3
python/obitools3/dms/view/__init__.py
Normal file
3
python/obitools3/dms/view/__init__.py
Normal file
@ -0,0 +1,3 @@
|
||||
from .view import View # @UnresolvedImport
|
||||
from .view import Line_selection # @UnresolvedImport
|
||||
from .view import Line # @UnresolvedImport
|
42
python/obitools3/dms/view/view.pxd
Normal file
42
python/obitools3/dms/view/view.pxd
Normal file
@ -0,0 +1,42 @@
|
||||
#cython: language_level=3
|
||||
|
||||
from ..capi.obiview cimport Obiview_p
|
||||
from ..capi.obitypes cimport index_t, \
|
||||
obitype_t
|
||||
|
||||
from ..dms cimport DMS
|
||||
from ..column.column cimport Column
|
||||
|
||||
|
||||
|
||||
cdef class View:
|
||||
|
||||
cdef DMS _dms
|
||||
cdef Obiview_p _pointer
|
||||
|
||||
cpdef delete_column(self,
|
||||
object column_name)
|
||||
|
||||
cpdef rename_column(self,
|
||||
object current_name,
|
||||
object new_name)
|
||||
|
||||
cpdef View_line_selection new_selection(self,
|
||||
list lines=*)
|
||||
|
||||
cdef class Line_selection(list):
|
||||
|
||||
cdef View _view
|
||||
|
||||
cdef index_t* __build_binary_list__(self)
|
||||
|
||||
cpdef View materialize(self,
|
||||
object view_name,
|
||||
object comments=*)
|
||||
|
||||
cdef class Line :
|
||||
|
||||
cdef index_t _index
|
||||
cdef View _view
|
||||
|
||||
|
412
python/obitools3/dms/view/view.pyx
Normal file
412
python/obitools3/dms/view/view.pyx
Normal file
@ -0,0 +1,412 @@
|
||||
#cython: language_level=3
|
||||
|
||||
from libc.stdlib cimport malloc
|
||||
|
||||
from ..capi.obiview cimport Alias_column_pair_p, \
|
||||
obi_new_view, \
|
||||
obi_open_view, \
|
||||
obi_save_and_close_view, \
|
||||
obi_view_get_pointer_on_column_in_view, \
|
||||
obi_view_delete_column, \
|
||||
obi_view_create_column_alias
|
||||
|
||||
from ..capi.obidmscolumn cimport OBIDMS_column_p
|
||||
|
||||
from obitools3.utils cimport tobytes, \
|
||||
bytes2str
|
||||
|
||||
|
||||
cdef class View :
|
||||
|
||||
|
||||
def __init__(self,dms,int __internalCall__):
|
||||
|
||||
if __internalCall__!=987654:
|
||||
raise RuntimeError('OBIView constructor cannot be called directly')
|
||||
|
||||
self._dms = dms
|
||||
self._pointer = NULL
|
||||
|
||||
def clone(self,
|
||||
object view_name,
|
||||
object comments=None):
|
||||
|
||||
|
||||
cdef bytes view_name_b = tobytes(view_name)
|
||||
cdef bytes comments_b
|
||||
cdef View view = View(self._dms,
|
||||
987654)
|
||||
|
||||
if comments is not None:
|
||||
comments_b = tobytes(comments)
|
||||
else:
|
||||
comments_b = b''
|
||||
|
||||
view._pointer = obi_new_view(self._pointer.dms,
|
||||
view_name_b,
|
||||
self._pointer,
|
||||
NULL,
|
||||
comments_b)
|
||||
|
||||
if view._pointer == NULL :
|
||||
raise RuntimeError("Error : Cannot clone view %s into view %s"
|
||||
% (str(self.name),
|
||||
bytes2str(view_name_b))
|
||||
)
|
||||
|
||||
return view
|
||||
|
||||
@staticmethod
|
||||
def new(DMS dms,
|
||||
object view_name,
|
||||
object comments=None):
|
||||
|
||||
cdef bytes view_name_b = tobytes(view_name)
|
||||
cdef bytes comments_b
|
||||
cdef str message
|
||||
|
||||
cdef View view = View(dms,
|
||||
987654) # @DuplicatedSignature
|
||||
|
||||
if comments is not None:
|
||||
comments_b = tobytes(comments)
|
||||
else:
|
||||
comments_b = b''
|
||||
|
||||
view._pointer = obi_new_view(dms._pointer,
|
||||
view_name_b,
|
||||
NULL,
|
||||
NULL,
|
||||
comments_b)
|
||||
|
||||
if view._pointer == NULL :
|
||||
message = "Error : Cannot create view %s" % bytes2str(view_name_b)
|
||||
raise RuntimeError(message)
|
||||
|
||||
return view
|
||||
|
||||
@staticmethod
|
||||
def open(DMS dms, # @ReservedAssignment
|
||||
object view_name):
|
||||
|
||||
cdef bytes view_name_b = tobytes(view_name)
|
||||
cdef View view = View(dms,
|
||||
987654) # @DuplicatedSignature
|
||||
|
||||
view._pointer = obi_open_view(dms._pointer,
|
||||
view_name_b)
|
||||
|
||||
if view._pointer == NULL :
|
||||
raise RuntimeError("Error : Cannot open view %s" % bytes2str(view_name_b))
|
||||
|
||||
return view
|
||||
|
||||
def close(self):
|
||||
if (self._pointer != NULL
|
||||
and obi_save_and_close_view(self._pointer) < 0) :
|
||||
raise Exception("Problem closing view %s" %
|
||||
bytes2str(self.name))
|
||||
|
||||
def __dealloc__(self):
|
||||
self.close()
|
||||
|
||||
def __repr__(self) :
|
||||
cdef str s = "{name:s}\n{comments:s}\n{line_count:d} lines\n".format(name = str(self.name),
|
||||
comments = str(self.comments),
|
||||
line_count = self.line_count)
|
||||
|
||||
#for column_name in self._columns :
|
||||
# s = s + repr(self._columns[column_name]) + '\n'
|
||||
|
||||
return s
|
||||
|
||||
def keys(self):
|
||||
cdef int i
|
||||
cdef int nb_column = self._pointer.infos.column_count
|
||||
cdef Alias_column_pair_p column_p = self._pointer.infos.column_references
|
||||
|
||||
for i in range(nb_column) :
|
||||
col_alias = bytes2str(self._pointer.infos.column_references[i].alias)
|
||||
yield col_alias
|
||||
|
||||
def get_column(self,
|
||||
object column_name):
|
||||
cdef bytes column_name_b = tobytes(column_name)
|
||||
cdef OBIDMS_column_p* column_pp
|
||||
cdef OBIDMS_column_p column_p
|
||||
cdef Column column
|
||||
cdef obitype_t column_type
|
||||
|
||||
column_pp = obi_view_get_pointer_on_column_in_view(self._pointer,
|
||||
column_name_b)
|
||||
|
||||
if column_pp == NULL:
|
||||
raise KeyError("Cannot access to column %s in view %s" % (
|
||||
bytes2str(column_name_b),
|
||||
bytes2str(self.name)
|
||||
))
|
||||
|
||||
column_p = column_pp[0]
|
||||
column_type = column_p.header.returned_data_type
|
||||
|
||||
column = DMS.get_column_class(column_type)(self)
|
||||
column._pointer = column_pp
|
||||
|
||||
return column
|
||||
|
||||
cpdef delete_column(self,
|
||||
object column_name) :
|
||||
|
||||
cdef bytes column_name_b = tobytes(column_name)
|
||||
|
||||
if obi_view_delete_column(self._pointer, column_name_b) < 0 :
|
||||
raise Exception("Problem deleting column %s from a view",
|
||||
bytes2str(column_name_b))
|
||||
|
||||
|
||||
|
||||
cpdef rename_column(self,
|
||||
object current_name,
|
||||
object new_name):
|
||||
|
||||
cdef Column column
|
||||
cdef bytes current_name_b = tobytes(current_name)
|
||||
cdef bytes new_name_b = tobytes(new_name)
|
||||
|
||||
if (obi_view_create_column_alias(self._pointer,
|
||||
tobytes(current_name_b),
|
||||
tobytes(new_name_b)) < 0) :
|
||||
raise Exception("Problem in renaming column %s to %s" % (
|
||||
bytes2str(current_name_b),
|
||||
bytes2str(new_name_b)))
|
||||
|
||||
|
||||
cpdef View_line_selection new_selection(self,list lines=None):
|
||||
return View_line_selection(self,lines)
|
||||
|
||||
def __iter__(self):
|
||||
# Iteration on each line of all columns
|
||||
|
||||
# Declarations
|
||||
cdef index_t line_nb
|
||||
cdef View_line line
|
||||
|
||||
# Yield each line
|
||||
for line_nb in range(self.line_count) :
|
||||
line = self[line_nb]
|
||||
yield line
|
||||
|
||||
|
||||
def __getitem__(self, object item) :
|
||||
if type(item) == str :
|
||||
return (self._columns)[item]
|
||||
elif type(item) == int :
|
||||
return View_line(self, item)
|
||||
|
||||
|
||||
def __contains__(self, str column_name):
|
||||
return (column_name in self._columns)
|
||||
|
||||
|
||||
def __len__(self):
|
||||
return(self.line_count)
|
||||
|
||||
|
||||
def __str__(self) :
|
||||
cdef View_line line
|
||||
cdef str to_print
|
||||
to_print = ""
|
||||
for line in self :
|
||||
to_print = to_print + str(line) + "\n"
|
||||
return to_print
|
||||
|
||||
|
||||
@property
|
||||
def dms(self):
|
||||
return self._dms
|
||||
|
||||
# line_count property getter
|
||||
@property
|
||||
def line_count(self):
|
||||
return self._pointer.infos.line_count
|
||||
|
||||
# name property getter
|
||||
@property
|
||||
def name(self):
|
||||
return <bytes> self._pointer.infos.name
|
||||
|
||||
# view type property getter
|
||||
@property
|
||||
def type(self): # @ReservedAssignment
|
||||
return bytes2str(self._pointer.infos.view_type)
|
||||
|
||||
# comments property getter
|
||||
@property
|
||||
def comments(self):
|
||||
return <bytes> self._pointer.infos.comments
|
||||
# TODO setter that concatenates new comments?
|
||||
|
||||
|
||||
cdef class Line_selection(list):
|
||||
|
||||
def __init__(self, View view, lines=None) :
|
||||
if view._pointer == NULL:
|
||||
raise Exception("Error: trying to create a line selection with an invalidated view")
|
||||
self._view = view
|
||||
self._view_name = view.name
|
||||
|
||||
if lines is not None:
|
||||
self.extend(lines)
|
||||
|
||||
def extend(self, iterable):
|
||||
cdef index_t i
|
||||
cdef index_t max_i = self._view.line_count
|
||||
|
||||
for i in iterable:
|
||||
if i > max_i:
|
||||
raise RuntimeError("Error: trying to select line %d beyond the line count %d of view %s" %
|
||||
(i,
|
||||
max_i,
|
||||
self._view_name)
|
||||
)
|
||||
list.append(self,i)
|
||||
|
||||
def append(self, index_t idx) :
|
||||
if idx >= self._view.line_count :
|
||||
raise IndexError("Error: trying to select line %d beyond the line count %d of view %s" %
|
||||
(idx,
|
||||
self._view.line_count,
|
||||
bytes2str(self.name))
|
||||
)
|
||||
list.append(self,idx)
|
||||
|
||||
cdef index_t* __build_binary_list__(self):
|
||||
cdef index_t* line_selection_p = NULL
|
||||
cdef int i
|
||||
cdef size_t l_selection = len(self)
|
||||
|
||||
line_selection_p = <index_t*> malloc((l_selection + 1) * sizeof(index_t)) # +1 for the -1 flagging the end of the array
|
||||
for i in range(l_selection) :
|
||||
line_selection_p[i] = self[i]
|
||||
line_selection_p[l_selection] = -1 # flagging the end of the array
|
||||
|
||||
return line_selection_p
|
||||
|
||||
cpdef View materialize(self,
|
||||
object view_name,
|
||||
object comments=""):
|
||||
|
||||
cdef View view = View(987654)
|
||||
cdef bytes view_name_b=tobytes(view_name)
|
||||
|
||||
view._pointer = obi_new_view(self._view._pointer.dms,
|
||||
view_name_b,
|
||||
self._view._pointer,
|
||||
self.__build_binary_list__(),
|
||||
tobytes(comments))
|
||||
|
||||
if view._pointer == NULL :
|
||||
raise RuntimeError("Error : Cannot clone view %s into view %s"
|
||||
% (str(self.name),
|
||||
view_name)
|
||||
)
|
||||
|
||||
return view
|
||||
|
||||
|
||||
cdef class Line :
|
||||
|
||||
def __init__(self, View view, index_t line_nb) :
|
||||
self._index = line_nb
|
||||
self._view = view
|
||||
|
||||
def __getitem__(self, str column_name) :
|
||||
return ((self._view)._columns)[column_name][self._index]
|
||||
|
||||
# def __setitem__(self, str column_name, object value):
|
||||
# # TODO detect multiple elements (dict type)? put somewhere else? but more risky (in get)
|
||||
# # TODO OBI_QUAL ?
|
||||
# cdef type value_type
|
||||
# cdef str value_obitype
|
||||
# cdef bytes value_b
|
||||
#
|
||||
# if column_name not in self._view :
|
||||
# if value == None :
|
||||
# raise Exception("Trying to create a column from a None value (can't guess type)")
|
||||
# value_type = type(value)
|
||||
# if value_type == int :
|
||||
# value_obitype = 'OBI_INT'
|
||||
# elif value_type == float :
|
||||
# value_obitype = 'OBI_FLOAT'
|
||||
# elif value_type == bool :
|
||||
# value_obitype = 'OBI_BOOL'
|
||||
# elif value_type == str or value_type == bytes :
|
||||
# if value_type == str :
|
||||
# value_b = str2bytes(value)
|
||||
# else :
|
||||
# value_b = value
|
||||
# if only_ATGC(value_b) : # TODO detect IUPAC
|
||||
# value_obitype = 'OBI_SEQ'
|
||||
# elif len(value) == 1 :
|
||||
# value_obitype = 'OBI_CHAR'
|
||||
# elif (len(value) > 1) :
|
||||
# value_obitype = 'OBI_STR'
|
||||
# else :
|
||||
# raise Exception("Could not guess the type of a value to create a new column")
|
||||
# self._view.add_column(column_name, type=value_obitype)
|
||||
#
|
||||
# (((self._view)._columns)[column_name]).set_line(self._index, value)
|
||||
#
|
||||
# def __iter__(self):
|
||||
# for column_name in ((self._view)._columns) :
|
||||
# yield column_name
|
||||
#
|
||||
# def __contains__(self, str column_name):
|
||||
# return (column_name in self._view._columns)
|
||||
|
||||
def __repr__(self):
|
||||
cdef dict line
|
||||
cdef str column_name
|
||||
line = {}
|
||||
# for column_name in self._view._columns :
|
||||
# line[column_name] = self[column_name]
|
||||
return str(line)
|
||||
|
||||
|
||||
# cpdef dict get_view_infos(self, str view_name) :
|
||||
#
|
||||
# cdef Obiview_infos_p view_infos_p
|
||||
# cdef dict view_infos_d
|
||||
# cdef Alias_column_pair_p column_refs
|
||||
# cdef int i, j
|
||||
# cdef str column_name
|
||||
#
|
||||
# view_infos_p = obi_view_map_file(self._pointer,
|
||||
# tobytes(view_name))
|
||||
# view_infos_d = {}
|
||||
# view_infos_d["name"] = bytes2str(view_infos_p.name)
|
||||
# view_infos_d["comments"] = bytes2str(view_infos_p.comments)
|
||||
# view_infos_d["view_type"] = bytes2str(view_infos_p.view_type)
|
||||
# view_infos_d["column_count"] = <int> view_infos_p.column_count
|
||||
# view_infos_d["line_count"] = <int> view_infos_p.line_count
|
||||
# view_infos_d["created_from"] = bytes2str(view_infos_p.created_from)
|
||||
# view_infos_d["creation_date"] = bytes2str(obi_format_date(view_infos_p.creation_date))
|
||||
# if (view_infos_p.all_lines) :
|
||||
# view_infos_d["line_selection"] = None
|
||||
# else :
|
||||
# view_infos_d["line_selection"] = {}
|
||||
# view_infos_d["line_selection"]["column_name"] = bytes2str((view_infos_p.line_selection).column_name)
|
||||
# view_infos_d["line_selection"]["version"] = <int> (view_infos_p.line_selection).version
|
||||
# view_infos_d["column_references"] = {}
|
||||
# column_references = view_infos_p.column_references
|
||||
# for j in range(view_infos_d["column_count"]) :
|
||||
# column_name = bytes2str((column_references[j]).alias)
|
||||
# view_infos_d["column_references"][column_name] = {}
|
||||
# view_infos_d["column_references"][column_name]["original_name"] = bytes2str((column_references[j]).column_refs.column_name)
|
||||
# view_infos_d["column_references"][column_name]["version"] = (column_references[j]).column_refs.version
|
||||
#
|
||||
# obi_view_unmap_file(self._pointer, view_infos_p)
|
||||
#
|
||||
# return view_infos_d
|
||||
|
||||
|
Reference in New Issue
Block a user