def get_result()

in svinfer/processor/matrix.py [0:0]

• 46 lines of code
• 11 McCabe index (conditional complexity)

def get_result(tags, query_runner=None):
    result = {}
    sample_size = None
    sql_columns = [sqlalchemy.func.count(1)]
    sql_parts = []
    for k, v in tags.items():
        if k == "sample_size":
            raise ValueError("'sample_size' is reserved")
        if isinstance(v, SqlMatrix):
            start = len(sql_columns)
            for j in range(v.ncol):
                sql_columns.append(sqlalchemy.func.avg(v.value[j]))
            stop = len(sql_columns)
            sql_parts.append({
                "key": k,
                "start": start,
                "stop": stop,
                "dim": v.dim
            })
        elif isinstance(v, NumpyMatrix):
            if sample_size is None:
                sample_size = v.value.shape[0]
            else:
                assert sample_size == v.value.shape[0]
            result[k] = np.array(v.value.mean(axis=0)).reshape(v.dim)
        else:
            raise TypeError("Unknown value type with key = " + k)
    if len(sql_columns) > 1:
        if query_runner is None:
            raise ValueError("'query_runner' is required")
        query_text = str(
            sqlalchemy.select(sql_columns).compile(
                compile_kwargs={"literal_binds": True}
            )
        )
        raw = query_runner(query_text)
        if sample_size is None:
            sample_size = raw[0]
        else:
            assert sample_size == raw[0]
        for part in sql_parts:
            result[part["key"]] = np.array(
                raw[part["start"]:part["stop"]]
            ).reshape(part["dim"])
    result["sample_size"] = sample_size
    return result