107 lines
3.9 KiB
Cython
107 lines
3.9 KiB
Cython
#cython: language_level=3
|
|
|
|
from obitools3.utils cimport bytes2str, str2bytes
|
|
|
|
from .capi.obidms cimport obi_dms, \
|
|
obi_close_dms
|
|
|
|
from .capi.obidmscolumn cimport obi_close_column, \
|
|
OBIDMS_column_p, \
|
|
OBIDMS_column_header_p
|
|
|
|
from .capi.obiutils cimport obi_format_date
|
|
|
|
from .capi.obiview cimport Obiview_p, \
|
|
Obiview_infos_p, \
|
|
Alias_column_pair_p, \
|
|
obi_view_map_file, \
|
|
obi_view_unmap_file
|
|
|
|
from .capi.obitypes cimport const_char_p, \
|
|
OBIType_t, \
|
|
OBI_INT, \
|
|
OBI_FLOAT, \
|
|
OBI_BOOL, \
|
|
OBI_CHAR, \
|
|
OBI_QUAL, \
|
|
OBI_STR, \
|
|
OBI_SEQ, \
|
|
name_data_type, \
|
|
only_ATGC # discuss
|
|
|
|
from ._obitaxo cimport OBI_Taxonomy
|
|
|
|
|
|
|
|
from libc.stdlib cimport malloc
|
|
|
|
|
|
|
|
cdef class OBIDMS :
|
|
|
|
def __init__(self, str dms_name) :
|
|
|
|
# Declarations
|
|
cdef bytes dms_name_b
|
|
|
|
# Format the character string to send to C function
|
|
dms_name_b = str2bytes(dms_name)
|
|
|
|
# Fill structure and create or open the DMS
|
|
self._pointer = obi_dms(<const_char_p> dms_name_b)
|
|
if self._pointer == NULL :
|
|
raise Exception("Failed opening or creating an OBIDMS")
|
|
|
|
|
|
# name property getter
|
|
@property
|
|
def name(self):
|
|
return bytes2str(self._pointer.dms_name)
|
|
|
|
|
|
cpdef close(self) :
|
|
if (obi_close_dms(self._pointer)) < 0 :
|
|
raise Exception("Problem closing an OBIDMS")
|
|
|
|
|
|
cpdef OBI_Taxonomy open_taxonomy(self, str taxo_name) :
|
|
return OBI_Taxonomy(self, taxo_name)
|
|
|
|
|
|
cpdef dict read_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, str2bytes(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
|
|
|
|
|
|
|