tm1py-1.11-read-performance.py

tm1py-1.11-read-performance.py

from datetime import datetime

from TM1py.Objects.Axis import ViewAxisSelection
from TM1py.Objects.Cube import Cube
from TM1py.Objects.Dimension import Dimension
from TM1py.Objects.Hierarchy import Hierarchy
from TM1py.Objects.NativeView import NativeView
from TM1py.Objects.Subset import AnonymousSubset
from TM1py.Services.TM1Service import TM1Service

sdata_params = {
    "address": "",
    "port": 8010,
    "ssl": True,
    "user": "Admin",
    "password": "apple"
}

CUBE = "TM1py Cube"
SAMPLE_SIZE = 5

MDX = """
SELECT
NON EMPTY {TM1SubsetAll([TM1py Dimension 1])} ON ROWS,
NON EMPTY {TM1SubsetAll([TM1py Dimension 2])} ON COLUMNS
FROM [TM1py Cube]
"""


def setup(tm1: TM1Service):
    hierarchy = Hierarchy("TM1py Dimension 1", "TM1py Dimension 1")
    hierarchy.remove_all_elements()
    for e in range(1_000_000):
        element = str(e).zfill(7)
        hierarchy.add_element(element, "Numeric")
    dimension = Dimension(name="TM1py Dimension 1", hierarchies=[hierarchy])
    tm1.dimensions.update_or_create(dimension)

    hierarchy = Hierarchy("TM1py Dimension 2", "TM1py Dimension 2")
    hierarchy.add_element("Measure", "Numeric")
    dimension = Dimension(name="TM1py Dimension 2", hierarchies=[hierarchy])
    tm1.dimensions.update_or_create(dimension)

    cube = Cube(name="TM1py Cube", dimensions=["TM1py Dimension 1", "TM1py Dimension 2"])
    tm1.cubes.update_or_create(cube)
    print("Created Cube")

    cells = dict()
    element2 = "Measure"
    for e in range(1_000_000):
        element1 = str(e).zfill(7)
        value = e
        cells[element1, element2] = value
    tm1.cells.write(CUBE, cells, use_blob=True)

    view = NativeView(
        cube_name=CUBE,
        view_name="Default",
        rows=[ViewAxisSelection(
            dimension_name="TM1py Dimension 1",
            subset=AnonymousSubset(
                dimension_name="TM1pyDimension1",
                expression="{Tm1SubsetAll([TM1pyDimension1])}"))],
        columns=[ViewAxisSelection(
            dimension_name="TM1py Dimension 2",
            subset=AnonymousSubset(
                dimension_name="TM1pyDimension2",
                expression="{Tm1SubsetAll([TM1pyDimension2])}"))],
        suppress_empty_columns=True,
        suppress_empty_rows=True)
    tm1.views.update_or_create(view)


with TM1Service(**sdata_params) as tm1:
    setup(tm1)

    before = datetime.now()
    for _ in range(SAMPLE_SIZE):
        df = tm1.cells.execute_mdx_dataframe(MDX, use_blob=True)
    elapsed_time = datetime.now() - before
    print(f"execute_mdx_dataframe with use_blob=True: {elapsed_time / SAMPLE_SIZE}")

    before = datetime.now()
    for _ in range(SAMPLE_SIZE):
        df = tm1.cells.execute_mdx_dataframe(MDX, use_blob=False)
    elapsed_time = datetime.now() - before
    print(f"execute_mdx_dataframe with use_blob=False: {elapsed_time / SAMPLE_SIZE}")

    before = datetime.now()
    for _ in range(SAMPLE_SIZE):
        df = tm1.cells.execute_view_dataframe(cube_name=CUBE, view_name="Default", use_blob=True)
    elapsed_time = datetime.now() - before
    print(f"execute_view_dataframe with use_blob=True: {elapsed_time / SAMPLE_SIZE}")

    before = datetime.now()
    for _ in range(SAMPLE_SIZE):
        df = tm1.cells.execute_view_dataframe(cube_name=CUBE, view_name="Default", use_blob=False)
    elapsed_time = datetime.now() - before
    print(f"execute_view_dataframe with use_blob=False: {elapsed_time / SAMPLE_SIZE}")