obitools/obitools3
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 66 Wiki Activity

Cython API: compiling but not working

Browse Source
This commit is contained in:
Celine Mercier
2017-03-06 16:07:02 +01:00
parent 778acc48cd
commit 381194194c
19 changed files with 1404 additions and 1531 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
python/obitools3/commands/export.pyx
View File

@ -1,5 +1,5 @@
from obitools3.apps.progress cimport ProgressBar # @UnresolvedImport
from obitools3.obidms._obidms import OBIDMS # TODO cimport doesn't work
from obitools3.dms.dms import OBIDMS # TODO cimport doesn't work
from obitools3.utils cimport bytes2str
import time

59
python/obitools3/commands/grep.pyx
View File

@ -1,5 +1,6 @@
from obitools3.apps.progress cimport ProgressBar # @UnresolvedImport
from obitools3.obidms._obidms import OBIDMS, OBIView, OBIView_line_selection # TODO cimport doesn't work
from obitools3.dms.dms import OBIDMS # TODO cimport doesn't work
#, OBIView, OBIView_line_selection # TODO cimport doesn't work
from functools import reduce
import time
@ -56,34 +57,34 @@ def run(config):
d = OBIDMS(config['obi']['defaultdms'])
# Open input view 1
iview = d.open_view(config['obi']['inputview'])
# Initialize the progress bar
pb = ProgressBar(len(iview), config, seconde=5)
# Apply filter
selection = OBIView_line_selection(iview)
for i in range(len(iview)) :
pb(i)
line = iview[i]
loc_env = {'sequence': line, 'line': line} # TODO add taxonomy
good = (reduce(lambda bint x, bint y: x and y,
(bool(eval(p, loc_env, line))
for p in config['grep']['predicates']), True))
if good :
selection.append(i)
# Create output view with the line selection
oview = d.clone_view_with_line_selection(config['obi']['outputview'], selection, comments="obi grep: "+str(config['grep']['predicates'])+"\n")
#print("\n")
#print(repr(oview))
iview.close()
oview.close()
# iview = d.open_view(config['obi']['inputview'])
#
# # Initialize the progress bar
# pb = ProgressBar(len(iview), config, seconde=5)
#
# # Apply filter
# selection = OBIView_line_selection(iview)
# for i in range(len(iview)) :
# pb(i)
# line = iview[i]
#
# loc_env = {'sequence': line, 'line': line} # TODO add taxonomy
#
# good = (reduce(lambda bint x, bint y: x and y,
# (bool(eval(p, loc_env, line))
# for p in config['grep']['predicates']), True))
#
# if good :
# selection.append(i)
#
# # Create output view with the line selection
# oview = d.clone_view_with_line_selection(config['obi']['outputview'], selection, comments="obi grep: "+str(config['grep']['predicates'])+"\n")
#
# #print("\n")
# #print(repr(oview))
#
# iview.close()
# oview.close()
d.close()

36
python/obitools3/commands/import.pyx
View File

@ -2,7 +2,7 @@ from obitools3.apps.progress cimport ProgressBar # @UnresolvedImport
from obitools3.files.universalopener cimport uopen
from obitools3.parsers.fasta import fastaIterator
from obitools3.parsers.fastq import fastqIterator
from obitools3.obidms._obidms import OBIDMS
from obitools3.dms.dms import OBIDMS # TODO cimport doesn't work
import time
@ -109,22 +109,22 @@ def run(config):
d = OBIDMS(config['obi']['defaultdms'])
# Create view
view = d.new_view(config['import']['destview'], view_type=view_type, quality_column=get_quality)
i = 0
for seq in iseq:
pb(i)
view[i].id = seq['id']
view[i].definition = seq['definition']
view[i].nuc_seq = seq['sequence']
if get_quality :
view[i].quality = seq['quality']
for tag in seq['tags'] :
view[i][tag] = seq['tags'][tag]
i+=1
#print(view.__repr__())
view.close()
# view = d.new_view(config['import']['destview'], view_type=view_type, quality_column=get_quality)
#
# i = 0
# for seq in iseq:
# pb(i)
# view[i].id = seq['id']
# view[i].definition = seq['definition']
# view[i].nuc_seq = seq['sequence']
# if get_quality :
# view[i].quality = seq['quality']
# for tag in seq['tags'] :
# view[i][tag] = seq['tags'][tag]
# i+=1
#
# #print(view.__repr__())
#
# view.close()
d.close()

164
python/obitools3/commands/lcs.pyx
View File

@ -1,11 +1,11 @@
#cython: language_level=3
from obitools3.apps.progress cimport ProgressBar # @UnresolvedImport
from obitools3.obidms._obidms cimport OBIDMS # TODO cimport doesn't work
from obitools3.dms.dms import OBIDMS # TODO cimport doesn't work
from obitools3.utils cimport str2bytes
from obitools3.obidms.capi.obialign cimport obi_lcs_align_one_column, \
obi_lcs_align_two_columns
from obitools3.dms.capi.obialign cimport obi_lcs_align_one_column, \
obi_lcs_align_two_columns
import time
@ -155,82 +155,82 @@ def addOptions(parser):
help="Number of threads to use for the computation. Default: one.")
cpdef align(str dms_n,
str input_view_1_n, str output_view_n,
str input_view_2_n="",
str input_column_1_n="", str input_column_2_n="",
str input_elt_1_n="", str input_elt_2_n="",
str id_column_1_n="", str id_column_2_n="",
double threshold=0.0, bint normalize=True,
int reference=0, bint similarity_mode=True,
bint print_seq=False, bint print_count=False,
comments="",
int thread_count=1) :
cdef OBIDMS d
d = OBIDMS(dms_n)
if input_view_2_n == "" and input_column_2_n == "" :
if obi_lcs_align_one_column(d._pointer, \
str2bytes(input_view_1_n), \
str2bytes(input_column_1_n), \
str2bytes(input_elt_1_n), \
str2bytes(id_column_1_n), \
str2bytes(output_view_n), \
str2bytes(comments), \
print_seq, \
print_count, \
threshold, normalize, reference, similarity_mode,
thread_count) < 0 :
raise Exception("Error aligning sequences")
else :
if obi_lcs_align_two_columns(d._pointer, \
str2bytes(input_view_1_n), \
str2bytes(input_view_2_n), \
str2bytes(input_column_1_n), \
str2bytes(input_column_2_n), \
str2bytes(input_elt_1_n), \
str2bytes(input_elt_2_n), \
str2bytes(id_column_1_n), \
str2bytes(id_column_2_n), \
str2bytes(output_view_n), \
str2bytes(comments), \
print_seq, \
print_count, \
threshold, normalize, reference, similarity_mode) < 0 :
raise Exception("Error aligning sequences")
d.close()
def run(config):
# TODO: Build formatted comments with all parameters etc
comments = "Obi align"
# Call cython alignment function
align(config['obi']['defaultdms'], \
config['obi']['inputview1'], \
config['obi']['outputview'], \
input_view_2_n = config['obi']['inputview2'], \
input_column_1_n = config['obi']['inputcolumn1'], \
input_column_2_n = config['obi']['inputcolumn2'], \
input_elt_1_n = config['obi']['inputelement1'], \
input_elt_2_n = config['obi']['inputelement2'], \
id_column_1_n = config['obi']['idcolumn1'], \
id_column_2_n = config['obi']['idcolumn2'], \
threshold = config['align']['threshold'], \
normalize = config['align']['normalize'], \
reference = config['align']['reflength'], \
similarity_mode = config['align']['similarity'], \
print_seq = config['align']['printseq'], \
print_count = config['align']['printcount'], \
comments = comments, \
thread_count = config['align']['threadcount'])
print("Done.")
# cpdef align(str dms_n,
# str input_view_1_n, str output_view_n,
# str input_view_2_n="",
# str input_column_1_n="", str input_column_2_n="",
# str input_elt_1_n="", str input_elt_2_n="",
# str id_column_1_n="", str id_column_2_n="",
# double threshold=0.0, bint normalize=True,
# int reference=0, bint similarity_mode=True,
# bint print_seq=False, bint print_count=False,
# comments="",
# int thread_count=1) :
#
# cdef OBIDMS d
# d = OBIDMS(dms_n)
#
# if input_view_2_n == "" and input_column_2_n == "" :
# if obi_lcs_align_one_column(d._pointer, \
# str2bytes(input_view_1_n), \
# str2bytes(input_column_1_n), \
# str2bytes(input_elt_1_n), \
# str2bytes(id_column_1_n), \
# str2bytes(output_view_n), \
# str2bytes(comments), \
# print_seq, \
# print_count, \
# threshold, normalize, reference, similarity_mode,
# thread_count) < 0 :
# raise Exception("Error aligning sequences")
# else :
# if obi_lcs_align_two_columns(d._pointer, \
# str2bytes(input_view_1_n), \
# str2bytes(input_view_2_n), \
# str2bytes(input_column_1_n), \
# str2bytes(input_column_2_n), \
# str2bytes(input_elt_1_n), \
# str2bytes(input_elt_2_n), \
# str2bytes(id_column_1_n), \
# str2bytes(id_column_2_n), \
# str2bytes(output_view_n), \
# str2bytes(comments), \
# print_seq, \
# print_count, \
# threshold, normalize, reference, similarity_mode) < 0 :
# raise Exception("Error aligning sequences")
#
# d.close()
#
#
# def run(config):
#
# # TODO: Build formatted comments with all parameters etc
# comments = "Obi align"
#
# # Call cython alignment function
# align(config['obi']['defaultdms'], \
# config['obi']['inputview1'], \
# config['obi']['outputview'], \
# input_view_2_n = config['obi']['inputview2'], \
# input_column_1_n = config['obi']['inputcolumn1'], \
# input_column_2_n = config['obi']['inputcolumn2'], \
# input_elt_1_n = config['obi']['inputelement1'], \
# input_elt_2_n = config['obi']['inputelement2'], \
# id_column_1_n = config['obi']['idcolumn1'], \
# id_column_2_n = config['obi']['idcolumn2'], \
# threshold = config['align']['threshold'], \
# normalize = config['align']['normalize'], \
# reference = config['align']['reflength'], \
# similarity_mode = config['align']['similarity'], \
# print_seq = config['align']['printseq'], \
# print_count = config['align']['printcount'], \
# comments = comments, \
# thread_count = config['align']['threadcount'])
#
# print("Done.")
#
#
#
#
#

760
python/obitools3/commands/test.pyx
View File

@ -1,383 +1,383 @@
from obitools3.apps.progress cimport ProgressBar # TODO I absolutely don't understand why it doesn't work without that line
from obitools3.obidms._obidms import OBIDMS # TODO cimport doesn't work
from obitools3.obidms._obidms import OBIView_line_selection
from obitools3.utils cimport str2bytes
import shutil
import string
import random
VIEW_TYPES = ["", "NUC_SEQS_VIEW"]
COL_TYPES = ["OBI_BOOL", "OBI_CHAR", "OBI_FLOAT", "OBI_INT", "OBI_SEQ", "OBI_STR"]
NUC_SEQUENCE_COLUMN = "NUC_SEQ"
ID_COLUMN = "ID"
DEFINITION_COLUMN = "DEFINITION"
QUALITY_COLUMN = "QUALITY"
SPECIAL_COLUMNS = [NUC_SEQUENCE_COLUMN, ID_COLUMN, DEFINITION_COLUMN, QUALITY_COLUMN]
NAME_MAX_LEN = 200
COL_COMMENTS_MAX_LEN = 2048
MAX_INT = 2147483647 # used to generate random float values
__title__="Tests if the obitools are working properly"
default_config = {
}
def random_length(max_len):
return random.randint(1, max_len)
def random_bool(config):
return random.choice([True, False])
def random_char(config):
return random.choice(string.ascii_lowercase)
def random_float(config):
return random.randint(0, MAX_INT) + random.random()
def random_int(config):
return random.randint(0, config['test']['maxlinenb'])
def random_seq(config):
return str2bytes(''.join(random.choice(['a','t','g','c']) for i in range(random_length(config['test']['seqmaxlen']))))
def random_str(config):
return random_str_with_max_len(config['test']['strmaxlen'])
def random_str_with_max_len(max_len):
return ''.join(random.choice(string.ascii_lowercase) for i in range(random_length(max_len)))
def random_column(infos):
return random.choice(sorted(list(infos['view'].columns)))
def random_unique_name(infos):
name = ""
while name == "" or name in infos['unique_names'] :
name = random_str_with_max_len(NAME_MAX_LEN)
infos['unique_names'].append(name)
return name
def random_unique_element_name(config, infos):
name = ""
while name == "" or name in infos['unique_names'] :
name = random_str_with_max_len(config['test']['elt_name_max_len'])
infos['unique_names'].append(name)
return name
def print_test(config, sentence):
if config['test']['verbose'] :
print(sentence)
def test_set_and_get(config, infos):
print_test(config, ">>> Set and get test")
col_name = random_column(infos)
col = infos['view'][col_name]
element_names = col.elements_names
data_type = col.data_type
if data_type == "OBI_QUAL" :
print_test(config, "-")
return
idx = random_int(config)
value = infos['random_generator'][data_type](config)
if col.nb_elements_per_line > 1 :
elt = random.choice(element_names)
col[idx][elt] = value
assert col[idx][elt] == value, "Set value != gotten value "+str(col[idx][elt])+" != "+str(value)
else:
col[idx] = value
assert col[idx] == value, "Set value != gotten value "+str(col[idx])+" != "+str(value)
print_test(config, ">>> Set and get test OK")
def test_add_col(config, infos):
print_test(config, ">>> Add column test")
#existing_col = random_bool(config) # TODO doesn't work because of line count problem. See obiview.c line 1737
#if existing_col and infos["view_names"] != [] :
# random_view = infos['dms'].open_view(random.choice(infos["view_names"]))
# random_column = random_view[random.choice(sorted(list(random_view.columns))]
# random_column_refs = random_column.refs
# if random_column_refs['name'] in infos['view'] :
# alias = random_unique_name(infos)
# else :
# alias = ''
# infos['view'].add_column(random_column_refs['name'], version_number=random_column_refs['version'], alias=alias, create=False)
# random_view.close()
#else :
create_random_column(config, infos)
print_test(config, ">>> Add column test OK")
def test_delete_col(config, infos):
print_test(config, ">>> Delete column test")
if len(list(infos['view'].columns)) <= 1 :
print_test(config, "-")
return
col_name = random_column(infos)
if col_name in SPECIAL_COLUMNS :
print_test(config, "-")
return
infos['view'].delete_column(col_name)
print_test(config, ">>> Delete column test OK")
def test_col_alias(config, infos):
print_test(config, ">>> Changing column alias test")
col_name = random_column(infos)
if col_name in SPECIAL_COLUMNS :
print_test(config, "-")
return
infos['view'][col_name].alias = random_unique_name(infos)
print_test(config, ">>> Changing column alias test OK")
def test_new_view(config, infos):
print_test(config, ">>> New view test")
random_new_view(config, infos)
print_test(config, ">>> New view test OK")
def random_test(config, infos):
return random.choice(infos['tests'])(config, infos)
def random_view_type():
return random.choice(VIEW_TYPES)
def random_col_type():
return random.choice(COL_TYPES)
def fill_column(config, infos, col) :
data_type = col.data_type
element_names = col.elements_names
if len(element_names) > 1 :
for i in range(random_int(config)) :
for j in range(len(element_names)) :
col[i][element_names[j]] = infos['random_generator'][data_type](config)
else :
for i in range(random_int(config)) :
col[i] = infos['random_generator'][data_type](config)
def create_random_column(config, infos) :
alias = random.choice(['', random_unique_name(infos)])
nb_elements_per_line=random.randint(1, config['test']['maxelts'])
elements_names = []
for i in range(nb_elements_per_line) :
elements_names.append(random_unique_element_name(config, infos))
elements_names = random.choice([None, elements_names])
name = random_unique_name(infos)
infos['view'].add_column(name,
alias=alias,
type=random_col_type(),
nb_elements_per_line=nb_elements_per_line,
elements_names=elements_names,
indexer_name=random.choice(['', random_unique_name(infos)]),
comments=random_str_with_max_len(COL_COMMENTS_MAX_LEN),
create=True
)
if alias != '' :
return infos['view'][alias]
else :
return infos['view'][name]
def fill_view(config, infos):
for i in range(random.randint(1, config['test']['maxinicolcount'])) :
col = create_random_column(config, infos)
fill_column(config, infos, col)
def random_new_view(config, infos, first=False):
v_to_clone = None
line_selection = None
quality_col = False # TODO
if not first:
infos['view_names'].append(infos['view'].name)
infos['view'].close()
v_to_clone = infos['dms'].open_view(random.choice(infos["view_names"]))
v_type = ""
print_test(config, "View to clone: ")
print_test(config, repr(v_to_clone))
create_line_selection = random_bool(config)
if create_line_selection and v_to_clone.line_count > 0:
print_test(config, "New view with new line selection.")
line_selection = OBIView_line_selection(v_to_clone)
for i in range(random.randint(1, v_to_clone.line_count)) :
line_selection.append(random.randint(0, v_to_clone.line_count-1))
#print_test(config, "New line selection: "+str(line_selection))
else :
v_type = random_view_type()
if line_selection is not None :
infos['view'] = infos['dms'].clone_view_with_line_selection(random_unique_name(infos), line_selection, comments=random_str_with_max_len(config['test']['commentsmaxlen']))
elif v_to_clone is not None :
infos['view'] = infos['dms'].clone_view(random_unique_name(infos), v_to_clone, comments=random_str_with_max_len(config['test']['commentsmaxlen']))
else :
infos['view'] = infos['dms'].new_view(random_unique_name(infos), view_type=v_type, comments=random_str_with_max_len(config['test']['commentsmaxlen']), quality_column=quality_col)
print_test(config, repr(infos['view']))
if v_to_clone is not None :
if line_selection is None:
assert v_to_clone.line_count == infos['view'].line_count, "New view and cloned view don't have the same line count"
else :
assert len(line_selection) == infos['view'].line_count, "New view with new line selection does not have the right line count"
v_to_clone.close()
if first :
fill_view(config, infos)
def create_test_obidms(config, infos):
infos['dms'] = OBIDMS(config['obi']['defaultdms'])
def ini_dms_and_first_view(config, infos):
create_test_obidms(config, infos)
random_new_view(config, infos, first=True)
infos['view_names'] = []
def addOptions(parser):
# TODO put this common group somewhere else but I don't know where
group=parser.add_argument_group('DMS and view options')
group.add_argument('--default-dms','-d',
action="store", dest="obi:defaultdms",
metavar='<DMS NAME>',
default="/tmp/test_dms",
type=str,
help="Name of the default DMS for reading and writing data. "
"Default: /tmp/test_dms")
group=parser.add_argument_group('obi test specific options')
group.add_argument('--nb_tests','-n',
action="store", dest="test:nbtests",
metavar='<NB_TESTS>',
default=1000,
type=int,
help="Number of tests to carry out. "
"Default: 1000")
group.add_argument('--seq_max_len','-s',
action="store", dest="test:seqmaxlen",
metavar='<SEQ_MAX_LEN>',
default=200,
type=int,
help="Maximum length of DNA sequences. "
"Default: 200")
group.add_argument('--str_max_len','-t',
action="store", dest="test:strmaxlen",
metavar='<STR_MAX_LEN>',
default=200,
type=int,
help="Maximum length of character strings. "
"Default: 200")
group.add_argument('--comments_max_len','-c',
action="store", dest="test:commentsmaxlen",
metavar='<COMMENTS_MAX_LEN>',
default=10000,
type=int,
help="Maximum length of view comments. "
"Default: 10000")
group.add_argument('--max_ini_col_count','-o',
action="store", dest="test:maxinicolcount",
metavar='<MAX_INI_COL_COUNT>',
default=10,
type=int,
help="Maximum number of columns in the initial view. "
"Default: 10")
group.add_argument('--max_line_nb','-l',
action="store", dest="test:maxlinenb",
metavar='<MAX_LINE_NB>',
default=10000,
type=int,
help="Maximum number of lines in a column. "
"Default: 10000")
group.add_argument('--max_elts_per_line','-e',
action="store", dest="test:maxelts",
metavar='<MAX_ELTS_PER_LINE>',
default=20,
type=int,
help="Maximum number of elements per line in a column. "
"Default: 20")
group.add_argument('--verbose','-v',
action="store_true", dest="test:verbose",
default=False,
help="Print the tests. "
"Default: Don't print the tests")
group.add_argument('--seed','-g',
action="store", dest="test:seed",
metavar='<SEED>',
default=None,
help="Seed (use for reproducible tests). "
"Default: Seed is determined by Python")
def run(config):
if 'seed' in config['test'] :
random.seed(config['test']['seed'])
infos = {'dms': None,
'view': None,
'view_names': None,
'unique_names': [],
'random_generator': {"OBI_BOOL": random_bool, "OBI_CHAR": random_char, "OBI_FLOAT": random_float, "OBI_INT": random_int, "OBI_SEQ": random_seq, "OBI_STR": random_str},
'tests': [test_set_and_get, test_add_col, test_delete_col, test_col_alias, test_new_view]
}
config['test']['elt_name_max_len'] = int((COL_COMMENTS_MAX_LEN - config['test']['maxelts']) / config['test']['maxelts'])
print("Initializing the DMS and the first view...")
shutil.rmtree(config['obi']['defaultdms']+'.obidms', ignore_errors=True)
ini_dms_and_first_view(config, infos)
print_test(config, repr(infos['view']))
i = 0
for t in range(config['test']['nbtests']):
random_test(config, infos)
print_test(config, repr(infos['view']))
i+=1
if (i%(config['test']['nbtests']/10)) == 0 :
print("Testing......"+str(i*100/config['test']['nbtests'])+"%")
#print(infos)
infos['view'].close()
infos['dms'].close()
shutil.rmtree(config['obi']['defaultdms']+'.obidms', ignore_errors=True)
print("Done.")
from obitools3.dms.dms import OBIDMS # TODO cimport doesn't work
# from obitools3.obidms._obidms import OBIView_line_selection
# from obitools3.utils cimport str2bytes
#
# import shutil
# import string
# import random
#
#
# VIEW_TYPES = ["", "NUC_SEQS_VIEW"]
# COL_TYPES = ["OBI_BOOL", "OBI_CHAR", "OBI_FLOAT", "OBI_INT", "OBI_SEQ", "OBI_STR"]
# NUC_SEQUENCE_COLUMN = "NUC_SEQ"
# ID_COLUMN = "ID"
# DEFINITION_COLUMN = "DEFINITION"
# QUALITY_COLUMN = "QUALITY"
# SPECIAL_COLUMNS = [NUC_SEQUENCE_COLUMN, ID_COLUMN, DEFINITION_COLUMN, QUALITY_COLUMN]
#
#
# NAME_MAX_LEN = 200
# COL_COMMENTS_MAX_LEN = 2048
# MAX_INT = 2147483647 # used to generate random float values
#
#
# __title__="Tests if the obitools are working properly"
#
#
# default_config = {
# }
#
#
# def random_length(max_len):
# return random.randint(1, max_len)
#
#
# def random_bool(config):
# return random.choice([True, False])
#
#
# def random_char(config):
# return random.choice(string.ascii_lowercase)
#
#
# def random_float(config):
# return random.randint(0, MAX_INT) + random.random()
#
#
# def random_int(config):
# return random.randint(0, config['test']['maxlinenb'])
#
#
# def random_seq(config):
# return str2bytes(''.join(random.choice(['a','t','g','c']) for i in range(random_length(config['test']['seqmaxlen']))))
#
#
# def random_str(config):
# return random_str_with_max_len(config['test']['strmaxlen'])
#
#
# def random_str_with_max_len(max_len):
# return ''.join(random.choice(string.ascii_lowercase) for i in range(random_length(max_len)))
#
#
# def random_column(infos):
# return random.choice(sorted(list(infos['view'].columns)))
#
#
# def random_unique_name(infos):
# name = ""
# while name == "" or name in infos['unique_names'] :
# name = random_str_with_max_len(NAME_MAX_LEN)
# infos['unique_names'].append(name)
# return name
#
#
# def random_unique_element_name(config, infos):
# name = ""
# while name == "" or name in infos['unique_names'] :
# name = random_str_with_max_len(config['test']['elt_name_max_len'])
# infos['unique_names'].append(name)
# return name
#
#
# def print_test(config, sentence):
# if config['test']['verbose'] :
# print(sentence)
#
#
# def test_set_and_get(config, infos):
# print_test(config, ">>> Set and get test")
# col_name = random_column(infos)
# col = infos['view'][col_name]
# element_names = col.elements_names
# data_type = col.data_type
# if data_type == "OBI_QUAL" :
# print_test(config, "-")
# return
# idx = random_int(config)
# value = infos['random_generator'][data_type](config)
# if col.nb_elements_per_line > 1 :
# elt = random.choice(element_names)
# col[idx][elt] = value
# assert col[idx][elt] == value, "Set value != gotten value "+str(col[idx][elt])+" != "+str(value)
# else:
# col[idx] = value
# assert col[idx] == value, "Set value != gotten value "+str(col[idx])+" != "+str(value)
#
# print_test(config, ">>> Set and get test OK")
#
#
# def test_add_col(config, infos):
# print_test(config, ">>> Add column test")
# #existing_col = random_bool(config) # TODO doesn't work because of line count problem. See obiview.c line 1737
# #if existing_col and infos["view_names"] != [] :
# # random_view = infos['dms'].open_view(random.choice(infos["view_names"]))
# # random_column = random_view[random.choice(sorted(list(random_view.columns))]
# # random_column_refs = random_column.refs
# # if random_column_refs['name'] in infos['view'] :
# # alias = random_unique_name(infos)
# # else :
# # alias = ''
# # infos['view'].add_column(random_column_refs['name'], version_number=random_column_refs['version'], alias=alias, create=False)
# # random_view.close()
# #else :
# create_random_column(config, infos)
# print_test(config, ">>> Add column test OK")
#
#
# def test_delete_col(config, infos):
# print_test(config, ">>> Delete column test")
# if len(list(infos['view'].columns)) <= 1 :
# print_test(config, "-")
# return
# col_name = random_column(infos)
# if col_name in SPECIAL_COLUMNS :
# print_test(config, "-")
# return
# infos['view'].delete_column(col_name)
# print_test(config, ">>> Delete column test OK")
#
#
# def test_col_alias(config, infos):
# print_test(config, ">>> Changing column alias test")
# col_name = random_column(infos)
# if col_name in SPECIAL_COLUMNS :
# print_test(config, "-")
# return
# infos['view'][col_name].alias = random_unique_name(infos)
# print_test(config, ">>> Changing column alias test OK")
#
#
# def test_new_view(config, infos):
# print_test(config, ">>> New view test")
# random_new_view(config, infos)
# print_test(config, ">>> New view test OK")
#
#
# def random_test(config, infos):
# return random.choice(infos['tests'])(config, infos)
#
#
# def random_view_type():
# return random.choice(VIEW_TYPES)
#
#
# def random_col_type():
# return random.choice(COL_TYPES)
#
#
# def fill_column(config, infos, col) :
# data_type = col.data_type
# element_names = col.elements_names
#
# if len(element_names) > 1 :
# for i in range(random_int(config)) :
# for j in range(len(element_names)) :
# col[i][element_names[j]] = infos['random_generator'][data_type](config)
# else :
# for i in range(random_int(config)) :
# col[i] = infos['random_generator'][data_type](config)
#
#
# def create_random_column(config, infos) :
# alias = random.choice(['', random_unique_name(infos)])
# nb_elements_per_line=random.randint(1, config['test']['maxelts'])
# elements_names = []
# for i in range(nb_elements_per_line) :
# elements_names.append(random_unique_element_name(config, infos))
# elements_names = random.choice([None, elements_names])
# name = random_unique_name(infos)
# infos['view'].add_column(name,
# alias=alias,
# type=random_col_type(),
# nb_elements_per_line=nb_elements_per_line,
# elements_names=elements_names,
# indexer_name=random.choice(['', random_unique_name(infos)]),
# comments=random_str_with_max_len(COL_COMMENTS_MAX_LEN),
# create=True
# )
# if alias != '' :
# return infos['view'][alias]
# else :
# return infos['view'][name]
#
#
# def fill_view(config, infos):
# for i in range(random.randint(1, config['test']['maxinicolcount'])) :
# col = create_random_column(config, infos)
# fill_column(config, infos, col)
#
#
# def random_new_view(config, infos, first=False):
# v_to_clone = None
# line_selection = None
# quality_col = False # TODO
# if not first:
# infos['view_names'].append(infos['view'].name)
# infos['view'].close()
# v_to_clone = infos['dms'].open_view(random.choice(infos["view_names"]))
# v_type = ""
# print_test(config, "View to clone: ")
# print_test(config, repr(v_to_clone))
# create_line_selection = random_bool(config)
# if create_line_selection and v_to_clone.line_count > 0:
# print_test(config, "New view with new line selection.")
# line_selection = OBIView_line_selection(v_to_clone)
# for i in range(random.randint(1, v_to_clone.line_count)) :
# line_selection.append(random.randint(0, v_to_clone.line_count-1))
# #print_test(config, "New line selection: "+str(line_selection))
# else :
# v_type = random_view_type()
#
# if line_selection is not None :
# infos['view'] = infos['dms'].clone_view_with_line_selection(random_unique_name(infos), line_selection, comments=random_str_with_max_len(config['test']['commentsmaxlen']))
# elif v_to_clone is not None :
# infos['view'] = infos['dms'].clone_view(random_unique_name(infos), v_to_clone, comments=random_str_with_max_len(config['test']['commentsmaxlen']))
# else :
# infos['view'] = infos['dms'].new_view(random_unique_name(infos), view_type=v_type, comments=random_str_with_max_len(config['test']['commentsmaxlen']), quality_column=quality_col)
#
# print_test(config, repr(infos['view']))
# if v_to_clone is not None :
# if line_selection is None:
# assert v_to_clone.line_count == infos['view'].line_count, "New view and cloned view don't have the same line count"
# else :
# assert len(line_selection) == infos['view'].line_count, "New view with new line selection does not have the right line count"
# v_to_clone.close()
# if first :
# fill_view(config, infos)
#
#
# def create_test_obidms(config, infos):
# infos['dms'] = OBIDMS(config['obi']['defaultdms'])
#
#
# def ini_dms_and_first_view(config, infos):
# create_test_obidms(config, infos)
# random_new_view(config, infos, first=True)
# infos['view_names'] = []
#
#
# def addOptions(parser):
#
# # TODO put this common group somewhere else but I don't know where
# group=parser.add_argument_group('DMS and view options')
#
# group.add_argument('--default-dms','-d',
# action="store", dest="obi:defaultdms",
# metavar='<DMS NAME>',
# default="/tmp/test_dms",
# type=str,
# help="Name of the default DMS for reading and writing data. "
# "Default: /tmp/test_dms")
#
# group=parser.add_argument_group('obi test specific options')
#
# group.add_argument('--nb_tests','-n',
# action="store", dest="test:nbtests",
# metavar='<NB_TESTS>',
# default=1000,
# type=int,
# help="Number of tests to carry out. "
# "Default: 1000")
#
# group.add_argument('--seq_max_len','-s',
# action="store", dest="test:seqmaxlen",
# metavar='<SEQ_MAX_LEN>',
# default=200,
# type=int,
# help="Maximum length of DNA sequences. "
# "Default: 200")
#
# group.add_argument('--str_max_len','-t',
# action="store", dest="test:strmaxlen",
# metavar='<STR_MAX_LEN>',
# default=200,
# type=int,
# help="Maximum length of character strings. "
# "Default: 200")
#
# group.add_argument('--comments_max_len','-c',
# action="store", dest="test:commentsmaxlen",
# metavar='<COMMENTS_MAX_LEN>',
# default=10000,
# type=int,
# help="Maximum length of view comments. "
# "Default: 10000")
#
# group.add_argument('--max_ini_col_count','-o',
# action="store", dest="test:maxinicolcount",
# metavar='<MAX_INI_COL_COUNT>',
# default=10,
# type=int,
# help="Maximum number of columns in the initial view. "
# "Default: 10")
#
# group.add_argument('--max_line_nb','-l',
# action="store", dest="test:maxlinenb",
# metavar='<MAX_LINE_NB>',
# default=10000,
# type=int,
# help="Maximum number of lines in a column. "
# "Default: 10000")
#
# group.add_argument('--max_elts_per_line','-e',
# action="store", dest="test:maxelts",
# metavar='<MAX_ELTS_PER_LINE>',
# default=20,
# type=int,
# help="Maximum number of elements per line in a column. "
# "Default: 20")
#
# group.add_argument('--verbose','-v',
# action="store_true", dest="test:verbose",
# default=False,
# help="Print the tests. "
# "Default: Don't print the tests")
#
# group.add_argument('--seed','-g',
# action="store", dest="test:seed",
# metavar='<SEED>',
# default=None,
# help="Seed (use for reproducible tests). "
# "Default: Seed is determined by Python")
#
# def run(config):
#
# if 'seed' in config['test'] :
# random.seed(config['test']['seed'])
#
# infos = {'dms': None,
# 'view': None,
# 'view_names': None,
# 'unique_names': [],
# 'random_generator': {"OBI_BOOL": random_bool, "OBI_CHAR": random_char, "OBI_FLOAT": random_float, "OBI_INT": random_int, "OBI_SEQ": random_seq, "OBI_STR": random_str},
# 'tests': [test_set_and_get, test_add_col, test_delete_col, test_col_alias, test_new_view]
# }
#
# config['test']['elt_name_max_len'] = int((COL_COMMENTS_MAX_LEN - config['test']['maxelts']) / config['test']['maxelts'])
#
# print("Initializing the DMS and the first view...")
#
# shutil.rmtree(config['obi']['defaultdms']+'.obidms', ignore_errors=True)
#
# ini_dms_and_first_view(config, infos)
# print_test(config, repr(infos['view']))
#
# i = 0
# for t in range(config['test']['nbtests']):
# random_test(config, infos)
# print_test(config, repr(infos['view']))
# i+=1
# if (i%(config['test']['nbtests']/10)) == 0 :
# print("Testing......"+str(i*100/config['test']['nbtests'])+"%")
#
# #print(infos)
#
# infos['view'].close()
# infos['dms'].close()
# shutil.rmtree(config['obi']['defaultdms']+'.obidms', ignore_errors=True)
#
# print("Done.")

96
python/obitools3/commands/uniq.pyx
View File

@ -1,96 +0,0 @@
from obitools3.apps.progress cimport ProgressBar # @UnresolvedImport
from obitools3.obidms._obidms import OBIDMS, OBIView, OBIView_line_selection # TODO cimport doesn't work
from functools import reduce
import time
__title__="Grep view lines that match the given predicates"
default_config = { 'inputview' : None,
'outputview' : None
}
def addOptions(parser):
# TODO put this common group somewhere else but I don't know where
group=parser.add_argument_group('DMS and view options')
group.add_argument('--default-dms','-d',
action="store", dest="obi:defaultdms",
metavar='<DMS NAME>',
default=None,
type=str,
help="Name of the default DMS for reading and writing data.")
group.add_argument('--input-view','-i',
action="store", dest="obi:inputview",
metavar='<INPUT VIEW NAME>',
default=None,
type=str,
help="Name of the input view, either raw if the view is in the default DMS,"
" or in the form 'dms:view' if it is in another DMS.")
group.add_argument('--output-view','-o',
action="store", dest="obi:outputview",
metavar='<OUTPUT VIEW NAME>',
default=None,
type=str,
help="Name of the output view, either raw if the view is in the default DMS,"
" or in the form 'dms:view' if it is in another DMS.")
group=parser.add_argument_group('obi grep specific options')
group.add_argument('--predicate','-p',
action="append", dest="grep:predicates",
metavar='<PREDICATE>',
default=None,
type=str,
help="Grep lines that match the given python expression on <line> or <sequence>.")
def run(config):
# Open DMS
d = OBIDMS(config['obi']['defaultdms'])
# Open input view 1
iview = d.open_view(config['obi']['inputview'])
# Initialize the progress bar
pb = ProgressBar(len(iview), config, seconde=5)
# Apply filter
selection = OBIView_line_selection(iview)
for i in range(len(iview)) :
pb(i)
line = iview[i]
loc_env = {'sequence': line, 'line': line} # TODO add taxonomy
good = (reduce(lambda bint x, bint y: x and y,
(bool(eval(p, loc_env, line))
for p in config['grep']['predicates']), True))
if good :
selection.append(i)
# Create output view with the line selection
oview = d.new_view(config['obi']['outputview'], line_selection=selection, comments="obi grep: "+str(config['grep']['predicates'])+"\n")
#print("\n")
#print(repr(oview))
iview.close()
oview.close()
d.close()

4
python/obitools3/dms/capi/obialign.pxd
View File

@ -1,7 +1,7 @@
#cython: language_level=3
from obitools3.obidms.capi.obidms cimport OBIDMS_p
from obitools3.obidms.capi.obitypes cimport const_char_p
from obitools3.dms.capi.obidms cimport OBIDMS_p
from obitools3.dms.capi.obitypes cimport const_char_p
cdef extern from "obi_align.h" nogil:

25
python/obitools3/dms/column.pxd
View File

@ -1,25 +0,0 @@
#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.pyx
View File

@ -1,180 +0,0 @@
#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__):
'''
Creates 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)

107
python/obitools3/dms/column/bool.pxd
View File

@ -1,56 +1,53 @@
#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)
# #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 ..view.view cimport View
#
# from obitools3.utils cimport str2bytes, bytes2str
#
# cdef class Column_line_bool(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(Column):
#
# cpdef object get_line(self, index_t line_nb)
# cpdef set_line(self, index_t line_nb, object value)

585
python/obitools3/dms/column/bool.pyx
View File

@ -1,294 +1,305 @@
#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 Column_line_bool(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_pp
# # column_pp = <OBIDMS_column_p*>self._pointer
# # cdef OBIDMS_column_p column_p = 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, 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
# """
#
# 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()
#
# cdef OBIDMS_column_p* column_pp
# column_pp = <OBIDMS_column_p*>self._pointer
# cdef OBIDMS_column_p column_p = column_pp[0]
#
# value = obi_column_get_obibool_with_elt_name(column_p,
# self._index,
# cname)
#
# if obi_errno > 0 :
# obi_errno = 0
# raise KeyError("Cannot access to key %s" % bytes2str(element_name))
#
# return 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
#
# cdef OBIDMS_column_p* column_pp
# column_pp = <OBIDMS_column_p*>self._pointer
# cdef OBIDMS_column_p column_p = column_pp[0]
#
# self.update_pointer()
#
# value = obi_column_get_obibool_with_elt_idx(column_p,
# self._index,
# index)
#
# if obi_errno > 0 :
# obi_errno = 0
# raise IndexError("Cannot access to element %d" % index)
#
# return 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((<OBIDMS_column_p*>self._pointer)[0],
# 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((<OBIDMS_column_p*>self._pointer)[0],
# 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 Column_bool(Column):
# cdef class Column_bool(Column):
#
# @staticmethod
# def new(View view,
# object column_name,
# index_t nb_elements_per_line=1,
# object elements_names=None,
# object comments=b""):
#
# cdef bytes column_name_b = tobytes(column_name)
# cdef bytes comments_b
# cdef bytes elements_names_b
# cdef char* elements_names_p
# cdef Column new_column
#
# if comments is not None:
# comments_b = tobytes(comments)
# else:
# comments_b = b''
#
# 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_b,
# 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_b,
# 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
#
# def add_to_view(self,
# View view,
# object column_name=None,
# object comments=b""):
#
# cdef OBIDMS_column_p column_p = (<OBIDMS_column_p*>self._pointer)[0]
# cdef bytes alias
#
# if (column_name is None):
# alias = self._alias
# else:
# alias = tobytes(column_name)
#
# if (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) < 0):
# raise RuntimeError("Cannot insert column %s (%s@%d) into view %s" %
# ( bytes2str(alias),
# bytes2str(column_p.header.name),
# column_p.header.version,
# bytes2str(view.name)
# ))
#
# 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(<Obiview_p>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, Column_bool, bool)
@staticmethod
def new(OBIView view,
object column_name,
index_t nb_elements_per_line=1,
object elements_names=None,
object comments=b""):
cdef bytes column_name_b = tobytes(column_name)
cdef bytes comments_b
cdef bytes elements_names_b
cdef char* elements_names_p
cdef OBIDMS_column new_column
if comments is not None:
comments_b = tobytes(comments)
else:
comments_b = b''
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_b,
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_b,
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
def 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)
if (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) < 0):
raise RuntimeError("Cannot insert column %s (%s@%d) into view %s" %
( bytes2str(alias),
bytes2str(column_p.header.name),
column_p.header.version,
bytes2str(view.name)
))
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,Column_bool,bool)

40
python/obitools3/dms/column/column.pxd
View File

@ -1,26 +1,34 @@
#cython: language_level=3
from ..capi.obitypes cimport index_t, \
obitype_t
from ..capi.obidmscolumn cimport OBIDMS_column_p
from ..view cimport View
from ..view.view cimport View
from ..object cimport OBIWrapper
cdef class Column:
cdef OBIDMS_column_p* _pointer
cdef View _view
cdef class Column(OBIWrapper) :
cdef View _view
cpdef close(self)
cdef class Column_line:
@staticmethod
cdef type get_column_class(obitype_t obitype)
cdef Column _column
cdef index_t _index
cpdef update(self, data)
#
# 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 matrix,
type python)

426
python/obitools3/dms/column/column.pyx
View File

@ -1,23 +1,22 @@
#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 ..capi.obitypes cimport name_data_type
from ..capi.obidmscolumn cimport OBIDMS_column_header_p, \
obi_close_column
from ..capi.obiutils cimport obi_format_date
from ..dms cimport __OBIDMS_COLUMN_CLASS__
from obitools3.utils cimport bytes2str
cdef class Column :
cdef class Column(OBIWrapper) :
"""
The obitools3.dms.column.Column class wraps a c instance of a column in the context of a View
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,
@ -25,192 +24,235 @@ cdef class Column :
'''
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:
if __internalCall__ != 987654:
raise RuntimeError('OBIView constructor cannot be called directly')
OBIWrapper.__init__(self)
# 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
self._pointer = NULL
self._view = view
@staticmethod
def new(OBIView view,
object column_name,
index_t nb_elements_per_line=1,
object elements_names=None,
object comments=b""):
cdef bytes column_name_b = tobytes(column_name)
cdef bytes comments_b
cdef bytes elements_names_b
cdef char* elements_names_p
cdef OBIDMS_column new_column
if comments is not None:
comments_b = tobytes(comments)
else:
comments_b = b''
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_b,
version_number = -1,
alias = NULL,
data_type = self.pointer.header.returned_data_type,
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_b,
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
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 type get_column_class(obitype_t obitype):
"""
Internal function returning the python class representing
a column for a given obitype.
"""
return __OBIDMS_COLUMN_CLASS__[obitype][0]
# @staticmethod
# def new(OBIView view,
# object column_name,
# index_t nb_elements_per_line=1,
# object elements_names=None,
# object comments=b""):
#
# cdef bytes column_name_b = tobytes(column_name)
# cdef bytes comments_b
# cdef bytes elements_names_b
# cdef char* elements_names_p
# cdef OBIDMS_column new_column
#
# if comments is not None:
# comments_b = tobytes(comments)
# else:
# comments_b = b''
#
# 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_b,
# version_number = -1,
# alias = NULL,
# data_type = self.pointer.header.returned_data_type,
# 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_b,
# create = True)<0):
# raise RuntimeError("Cannot create column %s in view %s" % (bytes2str(column_name),
# bytes2str(view.name)))
#
# return Column.open(view,column_name)
#
#
# @staticmethod
# def open(OBIView view,
# 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(view._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)(view, 987654)
# column._pointer = column_pp
#
# return column
#
#
# 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):
@ -219,7 +261,7 @@ cdef register_column_class(obitype_t obitype,
to declare its relationship with an `OBIType_t`
"""
global __OBIDMS_COLUMN_CLASS__
assert issubclass(classe,Column)
__OBIDMS_COLUMN_CLASS__[obitype]=(classe,python)

18
python/obitools3/dms/dms.pxd
View File

@ -5,21 +5,21 @@ from .capi.obitypes cimport obiversion_t, \
obitype_t, \
index_t
from .object cimport OBIObject
from .object cimport OBIWrapper
cdef dict __OBIDMS_COLUMN_CLASS__
cdef dict __OBIDMS_VIEW_CLASS__
cdef class DMS(OBIObject):
cdef OBIDMS_p _pointer
@staticmethod
cdef type get_column_class(obitype_t obitype)
cdef class DMS(OBIWrapper):
@staticmethod
cdef type get_python_type(obitype_t obitype)
@staticmethod
cdef DMS new(object dms_name)
cpdef close(self)
cpdef int view_count(self)
cpdef bint is_view_writable(self, object view_name)

96
python/obitools3/dms/dms.pyx
View File

@ -20,18 +20,9 @@ from pathlib import Path
__OBIDMS_COLUMN_CLASS__ = {}
__OBIDMS_VIEW_CLASS__= {}
cdef class DMS(OBIObject):
@staticmethod
cdef type get_column_class(obitype_t obitype):
"""
Internal function returning the python class representing
a column for a given obitype.
"""
return __OBIDMS_COLUMN_CLASS__[obitype][0]
cdef class DMS(OBIWrapper):
@staticmethod
cdef type get_python_type(obitype_t obitype):
@ -40,31 +31,31 @@ cdef class DMS(OBIObject):
an instance for a given obitype.
"""
return __OBIDMS_COLUMN_CLASS__[obitype][1]
@staticmethod
cdef type get_view_class(bytes viewtype):
return __OBIDMS_VIEW_CLASS__[viewtype]
def __init__(self, object dms_name) :
'''
Constructor of a obitools3.dms.DMS instance.
@param dms_name: The name of the DMS
@type dms_name: a `str` or a `bytes` instance
'''
OBIObject.__init__(self)
# Declarations
# def __init__(self, size_t pointer) :
# '''
# Constructor of a obitools3.dms.DMS instance.
#
# @param dms_name: The name of the DMS
# @type dms_name: a `str` or a `bytes` instance
# '''
#
# print("hmmmm")
@staticmethod
cdef DMS new(object dms_name) :
cdef OBIDMS_p pointer
cdef DMS dms
cdef bytes dms_name_b = tobytes(dms_name)
# Fill structure and create or open the DMS
self._pointer = obi_dms(<const_char_p> dms_name_b)
if self._pointer == NULL :
pointer = obi_dms(<const_char_p> dms_name_b)
if pointer == NULL :
raise Exception("Failed opening or creating an OBIDMS")
dms = OBIWrapper.new_wrapper(DMS, pointer)
return dms
# name property getter
@property
def name(self):
@ -73,27 +64,32 @@ cdef class DMS(OBIObject):
@rtype: bytes
"""
return <bytes> self._pointer.dms_name
cdef OBIDMS_p pointer = <OBIDMS_p>(self._pointer)
return <bytes> pointer.dms_name
def close(self) :
"""
Closes the DNS instance and free the associated memory
Closes the DMS instance and free the associated memory
the `close` method is automatically called by the object destructor.
The `close` method is automatically called by the object destructor.
"""
cdef OBIDMS_p pointer = self._pointer
cdef OBIDMS_p pointer = <OBIDMS_p>(self._pointer)
OBIObject.close(self)
self._pointer=NULL
if pointer!=NULL:
OBIWrapper.close(self)
if (obi_close_dms(pointer)) < 0 :
raise Exception("Problem closing an OBIDMS")
self._pointer=NULL
else:
raise OBIObjectClosedInstance()
def keys(self):
cdef const_char_p path = obi_dms_get_full_path(self._pointer,
cdef OBIDMS_p pointer = <OBIDMS_p>(self._pointer)
cdef const_char_p path = obi_dms_get_full_path(pointer,
b"VIEWS"
)
@ -106,12 +102,14 @@ cdef class DMS(OBIObject):
for v in p.glob("*.obiview"):
yield str2bytes(v.stem)
def values(self):
cdef bytes view_name
for view_name in self.keys():
yield self.get_view(view_name)
def items(self):
cdef bytes view_name
@ -119,9 +117,14 @@ cdef class DMS(OBIObject):
for view_name in self.keys():
yield (view_name,self.get_view(view_name))
def __contains__(self,key):
def __contains__(self, key):
cdef OBIDMS_p pointer = <OBIDMS_p>(self._pointer)
cdef str key_s = tostr(key)
cdef const_char_p path = obi_dms_get_full_path(self._pointer,
cdef const_char_p path = obi_dms_get_full_path(pointer,
b"VIEWS"
)
p = Path(bytes2str(path),key_s)
@ -130,23 +133,28 @@ cdef class DMS(OBIObject):
return p.with_suffix(".obiview").is_file()
cpdef int view_count(self):
return PyList_Size(list(self.keys()))
def get_view(self):
raise NotImplemented
def __len__(self):
return self.view_count()
def __getitem__(self, object view_name):
return self.get_view(view_name)
def __iter__(self):
return self.keys()
cpdef bint is_view_writable(self, object view_name):
raise NotImplemented

21
python/obitools3/dms/object.pxd
View File

@ -1,13 +1,28 @@
#cython: language_level=3
cdef dict __c_cython_mapping__
cdef class OBIObject:
cdef dict _dependent_object
cpdef register(self, OBIObject object)
cpdef unregister(self, OBIObject object)
cdef register(self, OBIObject object)
cdef unregister(self, OBIObject object)
cpdef close(self)
cdef class OBIWrapper(OBIObject):
cdef void* _pointer
cdef size_t cid(self)
cpdef bint active(self)
@staticmethod
cdef object new_wrapper(type constructor, void* pointer)
cdef class OBIObjectClosedInstance(Exception):
pass

72
python/obitools3/dms/object.pyx
View File

@ -1,29 +1,63 @@
#cython: language_level=3
__c_cython_mapping__={}
cdef class OBIObject:
cpdef register(self, OBIObject object):
cdef register(self, OBIObject object):
self._dependent_object[id(object)]=object
cpdef unregister(self, OBIObject object):
cdef unregister(self, OBIObject object):
del self._dependent_object[id(object)]
def close(self):
cpdef close(self):
cdef OBIObject object
cdef list toclose = list(self._dependent_object.values())
for object in toclose:
object.close()
assert len(dependent_object.values)==0
def __init__(self):
assert len(self._dependent_object.values)==0
def __init__(self, __internalCall__):
if __internalCall__ != 987654 :
raise RuntimeError('OBIObject constructor can not be called directly')
if type(self) == OBIObject or type(self) == OBIWrapper or not isinstance(self, OBIObject) :
raise RuntimeError('OBIObject constructor can not be called directly')
self._dependent_object={}
cdef class OBIWrapper(OBIObject):
"""
The OBIWrapper class enables to wrap a C object representing a DMS or an element from a DMS
"""
cdef size_t cid(self) :
return <size_t>(self._pointer)
cpdef close(self):
if (self._pointer != NULL):
OBIObject.close(self)
self._pointer = NULL
assert len(self._dependent_object.values) == 0
cpdef bint active(self):
return self._pointer != NULL
def __dealloc__(self):
"""
Destructor of any OBI instance.
@ -31,9 +65,23 @@ cdef class OBIObject:
The destructor automatically calls the `close` method and
therefore frees all the associated memory.
"""
self.close()
cdef class OBIObjectClosedInstance(Exception):
@staticmethod
cdef object new_wrapper(type constructor, void* pointer):
cdef object o
if (<size_t>pointer in __c_cython_mapping__):
return __c_cython_mapping__[<size_t>pointer]
else:
o = constructor(<size_t>pointer, 987654)
o._pointer = <size_t>pointer
__c_cython_mapping__[<size_t>pointer] = o
return o
cdef class OBIDeactivatedInstanceError(Exception):
pass

22
python/obitools3/dms/view/view.pxd
View File

@ -4,18 +4,18 @@ from ..capi.obiview cimport Obiview_p
from ..capi.obitypes cimport index_t, \
obitype_t
from ..object cimport OBIObject
from ..object cimport OBIWrapper
from ..dms cimport DMS
from ..column.column cimport Column
cdef class View(OBIObject):
cdef class View(OBIWrapper):
cdef DMS _dms
cdef Obiview_p _pointer
cdef DMS _dms
cdef inline Obiview_p pointer(self)
cpdef delete_column(self,
object column_name)
@ -23,8 +23,16 @@ cdef class View(OBIObject):
object current_name,
object new_name)
cpdef View_line_selection new_selection(self,
list lines=*)
cpdef Line_selection new_selection(self,
list lines=*)
@staticmethod
cdef type get_view_class(bytes viewtype)
@staticmethod
cdef register_view_class(bytes viewtype,
type classe)
cdef class Line_selection(list):

222
python/obitools3/dms/view/view.pyx
View File

@ -2,6 +2,9 @@
from libc.stdlib cimport malloc
cdef dict __VIEW_CLASS__= {}
from ..capi.obiview cimport Alias_column_pair_p, \
obi_new_view, \
obi_open_view, \
@ -11,8 +14,7 @@ from ..capi.obiview cimport Alias_column_pair_p, \
obi_view_create_column_alias
from ..capi.obidmscolumn cimport OBIDMS_column_p
from .dms cimport __OBIDMS_VIEW_CLASS__
from ..capi.obidms cimport OBIDMS_p
from obitools3.utils cimport tobytes, \
bytes2str
@ -20,19 +22,77 @@ from obitools3.utils cimport tobytes, \
from ..object cimport OBIObjectClosedInstance
cdef class View(OBIObject) :
cdef class View(OBIWrapper) :
cdef inline Obiview_p pointer(self):
return <Obiview_p>(self.pointer)
@staticmethod
cdef register_view_class(bytes viewtype,
type classe):
"""
Each sub class of `dms.View` needs to be registered after its declaration
to declare its relationship with an `OBIType_t`
"""
global __VIEW_CLASS__
assert issubclass(classe,View)
__VIEW_CLASS__[viewtype]=classe
@staticmethod
cdef type get_view_class(bytes viewtype):
global __VIEW_CLASS__
return __VIEW_CLASS__.get(viewtype,View)
def __init__(self,dms,int __internalCall__):
OBIObject.__init__(self)
OBIWrapper.__init__(self,__internalCall__)
if __internalCall__!=987654:
raise RuntimeError('OBIView constructor cannot be called directly')
self._dms = dms
self._pointer = NULL
@staticmethod
def new(DMS dms,
object view_name,
object comments=None,
type viewclass=View):
cdef bytes view_name_b = tobytes(view_name)
cdef bytes comments_b
cdef str message
cdef void* pointer
cdef View view # @DuplicatedSignature
if comments is not None:
comments_b = tobytes(comments)
else:
comments_b = b''
pointer = <void*>obi_new_view(<OBIDMS_p>dms._pointer,
view_name_b,
NULL,
NULL,
comments_b)
if pointer == NULL :
message = "Error : Cannot create view %s" % bytes2str(view_name_b)
raise RuntimeError(message)
view = OBIWrapper.new_wrapper(viewclass, pointer)
view._dms = dms
dms.register(view)
return view
def clone(self,
object view_name,
object comments=None):
@ -40,95 +100,69 @@ cdef class View(OBIObject) :
cdef bytes view_name_b = tobytes(view_name)
cdef bytes comments_b
cdef View view = View(self._dms,
987654)
cdef void* pointer
cdef View
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)
pointer = <void*> obi_new_view(<OBIDMS_p>(self._dms._pointer),
view_name_b,
<Obiview_p>self._pointer,
NULL,
comments_b)
if view._pointer == NULL :
if pointer == NULL :
raise RuntimeError("Error : Cannot clone view %s into view %s"
% (str(self.name),
bytes2str(view_name_b))
)
view = OBIWrapper.new_wrapper(type(self),pointer)
view._dms = self._dms
self._dms.register(view)
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)
dms.register(view)
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)
cdef void* pointer
cdef View view
pointer = <void*> obi_open_view(<OBIDMS_p>dms._pointer,
view_name_b)
if view._pointer == NULL :
if pointer == NULL :
raise RuntimeError("Error : Cannot open view %s" % bytes2str(view_name_b))
view = OBIWrapper.new_wrapper(View.get_view_class((<Obiview_p>pointer).infos.view_type),
pointer)
view._dms = dms
dms.register(view)
return view
def close(self):
cdef Obiview_p pointer = self._pointer
cpdef close(self):
cdef Obiview_p pointer = <Obiview_p>self._pointer
if (pointer != NULL):
self._dms.unregister(self)
OBIObject.close(self)
self._pointer = NULL
if obi_save_and_close_view(self._pointer) < 0 :
OBIWrapper.close(self)
if obi_save_and_close_view(pointer) < 0 :
raise Exception("Problem closing view %s" %
bytes2str(self.name))
else:
raise OBIObjectClosedInstance()
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),
@ -138,16 +172,19 @@ cdef class View(OBIObject) :
# 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
cdef Obiview_p pointer = self.pointer()
cdef int nb_column = pointer.infos.column_count
cdef Alias_column_pair_p column_p = pointer.infos.column_references
for i in range(nb_column) :
col_alias = bytes2str(self._pointer.infos.column_references[i].alias)
col_alias = bytes2str(pointer.infos.column_references[i].alias)
yield col_alias
def get_column(self,
object column_name):
cdef bytes column_name_b = tobytes(column_name)
@ -155,8 +192,9 @@ cdef class View(OBIObject) :
cdef OBIDMS_column_p column_p
cdef Column column
cdef obitype_t column_type
cdef Obiview_p pointer = self.pointer()
column_pp = obi_view_get_pointer_on_column_in_view(self._pointer,
column_pp = obi_view_get_pointer_on_column_in_view(pointer,
column_name_b)
if column_pp == NULL:
@ -173,16 +211,16 @@ cdef class View(OBIObject) :
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 :
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,
@ -192,7 +230,7 @@ cdef class View(OBIObject) :
cdef bytes current_name_b = tobytes(current_name)
cdef bytes new_name_b = tobytes(new_name)
if (obi_view_create_column_alias(self._pointer,
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" % (
@ -200,15 +238,16 @@ cdef class View(OBIObject) :
bytes2str(new_name_b)))
cpdef View_line_selection new_selection(self,list lines=None):
return View_line_selection(self,lines)
cpdef Line_selection new_selection(self,list lines=None):
return Line_selection(self,lines)
def __iter__(self):
# Iteration on each line of all columns
# Declarations
cdef index_t line_nb
cdef View_line line
cdef Line line
# Yield each line
for line_nb in range(self.line_count) :
@ -220,7 +259,7 @@ cdef class View(OBIObject) :
if type(item) == str :
return (self._columns)[item]
elif type(item) == int :
return View_line(self, item)
return Line(self, item)
def __contains__(self, str column_name):
@ -232,7 +271,7 @@ cdef class View(OBIObject) :
def __str__(self) :
cdef View_line line
cdef Line line
cdef str to_print
to_print = ""
for line in self :
@ -244,25 +283,29 @@ cdef class View(OBIObject) :
def dms(self):
return self._dms
# line_count property getter
@property
def line_count(self):
return self._pointer.infos.line_count
return (<Obiview_p>self._pointer).infos.line_count
# name property getter
@property
def name(self):
return <bytes> self._pointer.infos.name
return <bytes> (<Obiview_p>self._pointer).infos.name
# view type property getter
@property
def type(self): # @ReservedAssignment
return <bytes> self._pointer.infos.view_type
return <bytes> (<Obiview_p>self._pointer).infos.view_type
# comments property getter
@property
def comments(self):
return <bytes> self._pointer.infos.comments
return <bytes> (<Obiview_p>self._pointer).infos.comments
# TODO setter that concatenates new comments?
@ -277,6 +320,7 @@ cdef class Line_selection(list):
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
@ -290,6 +334,7 @@ cdef class Line_selection(list):
)
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" %
@ -299,6 +344,7 @@ cdef class Line_selection(list):
)
list.append(self,idx)
cdef index_t* __build_binary_list__(self):
cdef index_t* line_selection_p = NULL
cdef int i
@ -311,6 +357,7 @@ cdef class Line_selection(list):
return line_selection_p
cpdef View materialize(self,
object view_name,
object comments=""):
@ -318,9 +365,9 @@ cdef class Line_selection(list):
cdef View view = View(987654)
cdef bytes view_name_b=tobytes(view_name)
view._pointer = obi_new_view(self._view._pointer.dms,
view._pointer = obi_new_view(<OBIDMS_p>self.view._dms.pointer,
view_name_b,
self._view._pointer,
<Obiview_p>self._view._pointer,
self.__build_binary_list__(),
tobytes(comments))
@ -339,6 +386,7 @@ cdef class Line :
self._index = line_nb
self._view = view
def __getitem__(self, str column_name) :
return ((self._view)._columns)[column_name][self._index]
@ -429,15 +477,3 @@ cdef class Line :
# return view_infos_d
cdef register_view_class(bytes viewtype,
type classe):
"""
Each sub class of `dms.View` needs to be registered after its declaration
to declare its relationship with an `OBIType_t`
"""
global __OBIDMS_VIEW_CLASS__
assert issubclass(classe,View)
__OBIDMS_VIEW_CLASS__[viewtype]=classe