Forum Discussion

Sidhant's avatar
Sidhant
Advocate V
1 year ago
Solved

Power BI Report creation through Script

Hi everyone, So recently I was experimenting that can we create reports through script apart from Power BI desktop and I found out that we can do it by using the '.pbip' file extension. And I did fo...
  • Sidhant's avatar
    1 year ago

    Hi everyone,
    A quick update so I am able to create reports in a programatic fashion it looks like:

    def create_model_bim(tables, output_folder):
        # Load a fresh copy of the model template
        model_bim = copy.deepcopy(model_template)
     
        # Replace {{query_order}} annotation
        for annotation in model_bim["model"].get("annotations", []):
            if annotation.get("value") == "{{query_order}}":
                annotation["value"] = str([tbl["name"] for tbl in tables])
     
        # Build table_column_map to validate relationships later
        table_column_map = {}
        
        with open("Static/data_type_mapping.json") as f:
            type_mapping = json.load(f)
    
        model_bim["model"]["tables"] = []
        for table in tables:
            column_names = [col["name"] for col in table["columns"]]
            table_dict = {
                "name": table["name"],
                "columns": [
                    {
                        "name": col["name"],
                        "dataType": type_mapping.get(col["type"].lower(), "string"),
                        "sourceColumn": col["name"]
                    } for col in table["columns"]
                ],
                "partitions": [
                    {
                        "name": f"{table['name']}_Partition",
                        "mode": "import",
                        "source": {
                            "type": "m",
                            "expression": (
                                f"let Source = Sql.Database(\"server_name\", \"database_name\") "
                                f"in Source{{[Schema=\"dbo\", Item=\"{table['name']}\"]}}[Data]"
                            )
                        }
                    }
                ]
            }
            
            
            model_bim["model"]["tables"].append(table_dict)
            # table_column_map[table["name"]] = [col["name"] for col in table["columns"]]
            table_column_map[table["name"]] = column_names
     
        #  Load relationships sheet and parse valid entries
        # metadata_path = os.path.join(os.path.dirname(output_folder), "Files", "MetaData.xlsx")
        metadata_path = os.path.abspath(os.path.join(output_folder, "..", "..", "Files", "MetaData.xlsx"))
        # relationships = extract_relationships_from_metadata(metadata_path)
        relationships = extract_relationships_from_metadata(metadata_path, table_column_map)
        if relationships:
            model_bim["model"]["relationships"] = relationships
     
        measures = extract_measures_from_metadata(metadata_path)
        if measures:
             measures_table = {
                 "name": "__Measures",
                 "columns": [
                    {
                        "name": "Dummy",
                        "dataType": "string"
                    }
                 ],
                 "measures": measures,
                 "partitions": [
                     {
                        "name": "__Measures_Partition",
                        "mode": "import",
                        "source": 
                            {
                                 "type": "m",
                                "expression": "let Source = #table({\"Dummy\"}, {}) in Source"
                            }
                    }
                 ]
             }
             model_bim["model"]["tables"].append(measures_table)
        
        # Write model.bim to file
        bim_path = os.path.join(output_folder, "model.bim")
        with open(bim_path, "w", encoding="utf-8") as file:
            json.dump(model_bim, file, indent=4)
     
        print(f" model.bim created at: {bim_path}")
     
     
    #-- Report.json file creation:
    def create_valid_report_json(report_folder_path, chart_types_df, chart_axes_df):
        base_config = copy.deepcopy(report_static_config["config"])
    
        def create_visual_config(visual_type, visual_index, axes):
            visual_id = str(uuid.uuid4())
            layout = {
                "id": 0,
                "position": {
                    "x": 100.0 + (visual_index % 2) * 450.0,
                    "y": 100.0 + (visual_index // 2) * 350.0,
                    "z": 0,
                    "width": 400.0,
                    "height": 300.0,
                    "tabOrder": 0
                }
            }
    
            projections = {}
            selects = []
            from_tables = set()
    
            for axis_type, axis_list in axes.items():
                projections[axis_type] = []
                for axis in axis_list:
                    table = axis['table_name']
                    column = axis['column']
                    full_ref = f"{table}.{column}"
                    from_tables.add(table)
    
                    if axis.get('aggregation'):
                        agg_func = 0  # sum
                        projections[axis_type].append({"queryRef": f"Sum({full_ref})"})
                        selects.append({
                            "Aggregation": {
                                "Expression": {
                                    "Column": {
                                        "Expression": {"SourceRef": {"Source": table}},
                                        "Property": column
                                    }
                                },
                                "Function": agg_func
                            },
                            "Name": f"Sum({full_ref})",
                            "NativeReferenceName": f"Sum of {column}"
                        })
                    else:
                        projections[axis_type].append({"queryRef": full_ref, "active": True})
                        selects.append({
                            "Column": {
                                "Expression": {"SourceRef": {"Source": table}},
                                "Property": column
                            },
                            "Name": full_ref,
                            "NativeReferenceName": column
                        })
    
            visual_config = {
                "name": visual_id,
                "layouts": [layout],
                "singleVisual": {
                    "visualType": visual_type,
                    "projections": projections,
                    "prototypeQuery": {
                        "Version": 2,
                        "From": [{"Name": t, "Entity": t, "Type": 0} for t in from_tables],
                        "Select": selects
                    },
                    "drillFilterOtherVisuals": True,
                    "hasDefaultSort": True,
                    "objects": {},
                    "vcObjects": {
                        "title": [
                            {
                                "properties": {
                                    "text": {
                                        "expr": {
                                            "Literal": {
                                                "Value": f"'{visual_type.title()} Visual {visual_index + 1}'"
                                            }
                                        }
                                    }
                                }
                            }
                        ]
                    }
                }
            }
            return visual_config
    
        visual_containers = []
    
        for idx, row in chart_types_df.iterrows():
            visual_type = row.get('plot_type')
            worksheet = row.get('worksheet')
    
            if not visual_type or not worksheet:
                continue  # Skip incomplete rows
    
            relevant_axes = chart_axes_df[
                (chart_axes_df['worksheet'] == worksheet) &
                (chart_axes_df['type'].isin(['rows', 'cols'])) &
                # (chart_axes_df['order_id'] == 0) &
                (chart_axes_df['table_name'].notna())
            ]
    
            axes_dict = {}
            for axis_type in ['rows', 'cols']:
                group = relevant_axes[relevant_axes['type'] == axis_type]
                if not group.empty:
                    axes_dict['Category' if axis_type == 'rows' else 'Y'] = group.apply(
                        lambda axis_row: {
                            'table_name': axis_row['table_name'],
                            'column': axis_row['column'],
                            'aggregation': str(axis_row.get('aggregation', '')).strip().lower() == 'sum'
                        },
                        axis=1
                    ).tolist()
    
            if axes_dict:
                config = create_visual_config(visual_type, idx, axes_dict)
                container = {
                    "config": json.dumps(config),
                    "filters": "[]",
                    "height": 300.0,
                    "width": 400.0,
                    "x": 100.0 + (idx % 2) * 450.0,
                    "y": 100.0 + (idx // 2) * 350.0,
                    "z": 0.0
                }
                visual_containers.append(container)
    
        report_json = {
            "config": json.dumps(base_config),
            "layoutOptimization": 0,
            "resourcePackages": [],
            "sections": [
                {
                    "config": "{}",
                    "displayName": "Auto Page",
                    "displayOption": 1,
                    "filters": "[]",
                    "height": 720.0,
                    "name": str(uuid.uuid4()),
                    "visualContainers": visual_containers,
                    "width": 1280.0
                }
            ]
        }
    
        output_path = os.path.join(report_folder_path, "report.json")
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(report_json, f, indent=4)
    
        print(f" report.json with {len(visual_containers)} visuals written at: {output_path}")