def dump_timeline()

in smallpond/session.py [0:0]

• 23 lines of code
• 5 McCabe index (conditional complexity)

    def dump_timeline(self, path: Optional[str] = None):
        """
        Dump the task timeline to a file.
        """
        path = path or os.path.join(self.runtime_ctx.log_root, "timeline")
        # the default timeline is grouped by worker
        exec_path = f"{path}_exec"
        ray.timeline(exec_path)
        logger.debug(f"dumped timeline to {exec_path}")

        # generate another timeline grouped by node
        with open(exec_path) as f:
            records = json.load(f)
        new_records = []
        for record in records:
            # swap record name and pid-tid
            name = record["name"]
            try:
                node_id = name.split(",")[-1]
                task_id = name.split("-")[1].split(".")[0]
                task_name = name.split("-")[0]
                record["pid"] = f"{node_id}-{task_name}"
                record["tid"] = f"task {task_id}"
                new_records.append(record)
            except Exception:
                # filter out other records
                pass
        node_path = f"{path}_plan"
        with open(node_path, "w") as f:
            json.dump(new_records, f)
        logger.debug(f"dumped timeline to {node_path}")