"""Map values used in the database to the corresponding python data type.

if val == 'true':

return True

elif val == 'false':

return False

try:

return int(val)

except ValueError:

pass

try:

return float(val)

except ValueError:

pass

"""Map a python data type to the corresponding string used in the database.

if type(val) == bool:

return str(val).lower()

regex = (

r"^Dialect sqlite\+pysqlite does \*not\* support Decimal objects natively\, "

"and SQLAlchemy must convert from floating point - rounding errors and other "

"issues may occur\. Please consider storing Decimal numbers as strings or "

"integers on this platform for lossless storage\.$")

"""Get the iteration variables used in the simulation

iterationvar_stmt = sqa.select([m.runattr.c.attrValue]).where(m.runattr.c.attrName == 'iterationvars')

def iterationvar_values_stmt(var):

return sqa.select([m.runattr.c.attrValue]).where(m.runattr.c.attrName == var).distinct()

with engine.connect() as conn:

iterationvars = ([re.match(r'\$(\w+)=.*', entry).groups(1)[0]

for entry in conn.execute(iterationvar_stmt).fetchone()[0].split(', ')])

return {var: [_map_database_value(entry[0]) for entry in conn.execute(iterationvar_values_stmt(var)).fetchall()]

"""Get a param which is unique for the database

return type(con.execute(sqa.select([m.runparam.c.parValue])

.where(m.runparam.c.parName.like('%{}'.format(name)))

.distinct())

filter_=None, aggregate=None, self_descriptive_result=False):

"""Get OMNeT++ result vectors

_ignore_decimal_warning()

def normalize_by(by):

"""Normalize the different 'by' syntaxes to a map of strings to lists of strings"""

def normalize_filter(filter_):

valid_types = (int, float, bool, str)

if filter_ is None:

return []

if type(filter_) in valid_types:

return [filter_]

if type(filter_) == list and all(type(val) in valid_types for val in filter_):

return filter_

else:

raise TypeError("Filter must be string, int, float or bool, a list of these or None")

if type(by) == str:

return {by: []}

elif type(by) == list and all(type(attr) == str for attr in by):

return {attr: [] for attr in by}

elif type(by) == dict and all(type(attr) == str for attr in by):

return {attr: normalize_filter(filter_) for attr, filter_ in by.items()}

else:

raise TypeError("By must be string, list of strings or dictionary")

def normalize_variable(variable):

"""Normalize the different variable syntaxes to a list of strings"""

return [variable] if type(variable) == str else variable

def simtime(simtime_raw, simtime_exponent):

"""Compute float simtime from fixed-point notation"""

return simtime_raw * 10 ** simtime_exponent

def single_filter(by, attribute):

return len(by[attribute]) == 1

def attribute_filter_expression(by, attribute):

return sqa.and_(m.runattr.c.attrName == attribute,

m.runattr.c.attrValue.in_(_map_python_value(val) for val in by[attribute])

if by[attribute] else True)

# Normalize parameters TODO complete: variables -> list, by -> dict

by = normalize_by(by)

variable = normalize_variable(variable)

attribute_subqueries = {attribute: sqa.select([m.runattr.c.runId, m.runattr.c.dbId, m.runattr.c.attrValue])

.where(attribute_filter_expression(by, attribute))

.alias()

for attribute in by}

select = []

select.extend(query.c.attrValue.label(attribute)

for attribute, query in attribute_subqueries.items()

if not (len(by[attribute]) == 1 and not self_descriptive_result))

if time:

select.append(sqa.func.simtime(m.vectordata.c.simtimeRaw, m.run.c.simtimeExp).label('simtime'))

if module:

select.append(m.vector.c.moduleName)

if not (len(variable) == 1 and not self_descriptive_result):

select.append(m.vector.c.vectorName)

if aggregate is not None:

select.append(aggregate(m.vectordata.c.value))

else:

select.append(m.vectordata.c.value)

tables = (m.run

.join(m.vector)

.join(m.vectordata))

for query in attribute_subqueries.values():

tables = tables.join(query)

constraints = []

if filter_ is not None:

constraints.append(filter_)

constraints.append((m.vector.c.vectorName.in_(variable)))

stmt = sqa.select(select).select_from(tables).where(sqa.and_(*constraints))

if aggregate is not None:

stmt = stmt.group_by(*(query.c.attrValue

for attribute, query in attribute_subqueries.items()

if not single_filter(by, attribute)))

with engine.connect() as conn:

if time:

conn.connection.connection.create_function('simtime', 2, simtime)

df = pd.read_sql(stmt, conn)

for attr, vals in by.items(): # FIXME find out why .assign()-based approach does not work

df[attr] = df[attr].astype(pd.api.types.CategoricalDtype(vals, ordered=True)

if type(vals[0]) == str else type(vals[0]))