120 lines
5.8 KiB
Cython
120 lines
5.8 KiB
Cython
#cython: language_level=3
|
|
|
|
from pathlib import Path
|
|
|
|
from .capidms cimport *
|
|
|
|
from obitools3.obidms.obidmscolumn.obidmscolumn_int.capidmscolumn_int import *
|
|
from obitools3.obidms.obidmscolumn.obidmscolumn_float.capidmscolumn_float import *
|
|
from obitools3.obidms.obidmscolumn.obidmscolumn_bool.capidmscolumn_bool import *
|
|
from obitools3.obidms.obidmscolumn.obidmscolumn_char.capidmscolumn_char import *
|
|
from obitools3.obidms.obidmscolumn.obidmscolumn_idx.capidmscolumn_idx import *
|
|
|
|
|
|
cdef class OBIDMS :
|
|
|
|
def __init__(self, dms_name) :
|
|
dms_name_b = dms_name.encode(encoding='UTF-8')
|
|
self.dms_name = dms_name
|
|
self.pointer = obi_dms(dms_name_b)
|
|
|
|
# def __del__(self) :
|
|
# obi_close_dms(self.pointer)
|
|
|
|
def list(self):
|
|
p = Path(self.dms_name+'.obidms')
|
|
#dms = {}
|
|
print("{:<25} {:<25} {:<25} {:<25}".format('-Column name-','-Data type-','-Latest version number-', '-Line count of latest version-'))
|
|
for entry in p.iterdir():
|
|
if entry.suffix == ".obicol":
|
|
column_name = entry.stem
|
|
column_name_b = column_name.encode('utf-8')
|
|
#dms[column_name] = {}
|
|
data_type = (name_data_type(obi_column_get_data_type_from_name(self.pointer, column_name_b))).decode('utf-8')
|
|
latest_version = obi_column_get_latest_version_from_name(self.pointer, column_name_b)
|
|
line_count = obi_column_get_line_count_from_name(self.pointer, column_name_b)
|
|
#dms[column_name]['data_type'] = data_type
|
|
#dms[column_name]['latest_version'] = latest_version
|
|
#dms[column_name]['line_count'] = line_count
|
|
print("{:<25} {:<25} {:<25} {:<25}".format(column_name, data_type, latest_version, line_count))
|
|
|
|
|
|
def open_column(self,
|
|
column_name,
|
|
bint create=False,
|
|
bint clone=False, bint clone_data=True,
|
|
obiversion_t version_number=-1,
|
|
type=0,
|
|
size_t nb_lines=0,
|
|
size_t nb_elements_per_line=1,
|
|
str elements_names=None):
|
|
|
|
column_name_b = column_name.encode(encoding='UTF-8')
|
|
|
|
if not type :
|
|
if create :
|
|
print("A data type must be specified")
|
|
else :
|
|
type = obi_column_get_data_type_from_name(self.pointer, column_name_b)
|
|
|
|
if type == 1 :
|
|
if (create or clone) :
|
|
column = OBIDMS_column_int_writable(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
else :
|
|
column = OBIDMS_column_int_read(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
|
|
elif type == 2 :
|
|
if (create or clone) :
|
|
column = OBIDMS_column_float_writable(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
else :
|
|
column = OBIDMS_column_float_read(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
|
|
elif type == 3 :
|
|
if (create or clone) :
|
|
column = OBIDMS_column_bool_writable(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
else :
|
|
column = OBIDMS_column_bool_read(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
|
|
elif type == 4 :
|
|
if (create or clone) :
|
|
column = OBIDMS_column_char_writable(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
else :
|
|
column = OBIDMS_column_char_read(self, column_name,
|
|
create, clone, clone_data,
|
|
version_number, type,
|
|
nb_lines, nb_elements_per_line,
|
|
elements_names)
|
|
|
|
else :
|
|
raise Exception("Problem with the data type")
|
|
|
|
return column
|