Skip to content

server

server

Defines database server for DataProductCollection from analyses

get_data_products 

get_data_products(
    analysis: str = "workdir.products", tag: str = "trace_plots", product_type: str = "DataProduct"
)
Traces

parse-json | project "elements" | extend "label"="pen.label" | mv-expand "points" | extend "x"="points.x", "y"="points.y" | project "label", "x", "y" | pivot sum("y"), "x", "label" | project "label", "x", "y"

Source code in src/trendify/server.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@app.route("/data_products/<analysis>/<tag>")
@app.route("/data_products/<analysis>/<tag>/")
@app.route("/data_products/<analysis>/<tag>/<product_type>")
@app.route("/data_products/<analysis>/<tag>/<product_type>/")
def get_data_products(
    analysis: str = "workdir.products",
    tag: str = "trace_plots",
    product_type: str = "DataProduct",
):
    """
    Example: Traces
        parse-json
        | project "elements"
        | extend "label"="pen.label"
        | mv-expand "points"
        | extend "x"="points.x", "y"="points.y"
        | project "label", "x", "y"
        | pivot sum("y"), "x", "label"
        | project "label", "x", "y"
    """
    FAILED_RETURN_VALUE = None
    query_return = FAILED_RETURN_VALUE
    product_type = str(product_type)

    match product_type:
        case DataProduct.__name__:
            filter_type = DataProduct
        case XYData.__name__:
            filter_type = XYData
        case Trace2D.__name__:
            filter_type = Trace2D
        case Point2D.__name__:
            filter_type = Point2D
        case TableEntry.__name__:
            filter_type = TableEntry
        case HistogramEntry.__name__:
            filter_type = HistogramEntry
        case _:
            query_return = f"{product_type = } should be in {valid_types_names_list}"
            return query_return

    try:
        analysis = str(analysis)
        analysis_path_components = (
            analysis.split(".") if "." in analysis else [analysis]
        )
        tag = str(tag)
        tag_path_components = tag.split(".") if "." in tag else [tag]
        collection_path_components = analysis_path_components + tag_path_components
        data_dir = DATABASE_ROOT.joinpath(*tuple(analysis_path_components))
        collection_dir = data_dir.joinpath(*tuple(tag_path_components))
        assert not any(("." in x) for x in collection_path_components)
        assert collection_dir.is_relative_to(data_dir)
    except AssertionError:
        query_return = f"Do not try to access stuff outside of {data_dir = }"
        print(f"Do not try to access stuff outside of {data_dir = }")
        return query_return

    data: DataProductCollection = DataProductCollection.collect_from_all_jsons(
        collection_dir
    )
    if data is None:
        return f"Did not find data product jsons in {collection_dir}"
    formatted_tag = (
        tag_path_components[0]
        if len(tag_path_components) == 1
        else tuple(tag_path_components)
    )
    filtered_data = data.get_products(
        tag=formatted_tag,
        object_type=filter_type,
    )
    query_return = filtered_data.model_dump_json()
    return query_return