Block Code Example to generate Metrics

This sample Block code generates some grouped Bar Graph and publishes to the metric server so that it can be visualized in the Pipeline run results interface under ‘Metrics’ tab.

from pipelineblocksdk.api.Singleton import Singleton
from pipelineblocksdk.construct.base.BatchBlock import BatchBlock

# Following code is the definition for a batch block
class MyBlock(BatchBlock, Singleton):

    # This is the entrypoint for the block. This is a mandatory function.
    def run(self):
        # -------- Generating  Bar Graphs -------------
        # Metric default value
        self.metrics["keys"] = self.block_params
        self.metrics["name"] = "BAR_GRAPH_V1.0"
        self.logger.info("Generating graphs....")

        self.metrics["metricId"] = "Animals In Zoo"
        self.logger.info("Creating graph for : " + "Animals In Zoo")
        self.metrics["metrics"] = {
            str(int(1)): {
                "x": "Giraffes",
                "y": [20,12],
                "x_title": "Animals",
                "y_title": "No : of Animals",
                "y_label": ["SF Zoo", "LA Zoo"],
                "colors": ["blue", "rgba(255,0,0,.8)"],
                "chart_type": "group"
            }
                
        }
        self.logger.info("Sending data to metric server... ")
        self.metrics_handler.save_metrics_time(self.metrics)
        
        self.metrics["metrics"] = {
            str(int(2)): {
                "x": "Monkey",
                "y": [30,8],
                "x_title": "Animals",
                "y_title": "No : of Animals",
                "y_label": ["SF Zoo", "LA Zoo"],
                "colors": ["blue", "rgba(255,0,0,.8)"],
                "chart_type": "group"
            }
                
        }
        self.logger.info("Sending data to metric server... ")
        self.metrics_handler.save_metrics_time(self.metrics)
        
        self.metrics["metrics"] = {
            str(int(3)): {
                "x": "Deer",
                "y": [34,19],
                "x_title": "Animals",
                "y_title": "No : of Animals",
                "y_label": ["SF Zoo", "LA Zoo"],
                "colors": ["blue", "rgba(255,0,0,.8)"],
                "chart_type": "group"
            } 
        }
        self.logger.info("Sending data to metric server... ")
        self.metrics_handler.save_metrics_time(self.metrics)
                              
        # -------- Generating Line Graphs -------------
        # Metric default value
        self.metrics["keys"] = self.block_params
        self.metrics["name"] = "LINE_GRAPH_v1.1"
        self.logger.info("Generating graphs....")
        self.metrics["appId"] = "METRIC_LINE_GRAPH"
            
        self.metrics["metricId"] = "No : of Giraffes over 3 days in a Zoo"
        self.logger.info("Creating graph for : " + "No : of Giraffes over 3 days in a Zoo")
        self.metrics["metrics"] = {
            str(int(time.time()*1000000)): {
                "x": 1,
                "y": [20,12],
                "x_label": "Days",
                "y_label": ["SF Zoo", "LA Zoo"],
                "x_type": "INTEGER",
                "y_type": "INTEGER"
            }
        }
        self.logger.info("Sending data to metric server... ")
        self.metrics_handler.save_metrics_time(self.metrics)
        
        self.metrics["metrics"] = {
            str(int(time.time()*1000000)): {
                "x":2,
                "y": [30,8],
                "x_label": "Zoo",
                "y_label": ["SF Zoo", "LA Zoo"],
                "x_type": "INTEGER",
                "y_type": "INTEGER"
            }
        }
        self.logger.info("Sending data to metric server... ")
        self.metrics_handler.save_metrics_time(self.metrics)
        
        self.metrics["metrics"] = {
            str(int(time.time()*1000000)): {
                "x":3,
                "y": [35,18],
                "x_label": "Zoo",
                "y_label": ["SF Zoo", "LA Zoo"],
                "x_type": "INTEGER",
                "y_type": "INTEGER"
            }
        }
        self.logger.info("Sending data to metric server... ")
        self.metrics_handler.save_metrics_time(self.metrics)

        # Set the output parameter
        output_dict = dict()
      
        # Set the status of the block as completed
        self.block_status = "COMPLETED"

        return output_dict