OpenLane/scripts/write_runtime.py

# Copyright 2021 Efabless Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import re
import sys
import yaml
import click
from typing import List, Optional


def timestamp_to_seconds(runtime: str) -> Optional[float]:
    pattern = re.compile(r"\s*([\d+]+)h([\d+]+)m([\d+]+)s(?:([\d+]+)+ms)?")
    m = pattern.match(runtime)
    if m is None:
        return None
    time = (
        int(m.group(1)) * 60 * 60
        + int(m.group(2)) * 60
        + int(m.group(3))
        + int(m.group(4) or 0) / 1000.0
    )
    return time


def seconds_to_timestamp(runtime: float) -> str:
    hours = int(runtime // 3600)
    minutes_and_seconds_and_milliseconds = runtime % 3600
    minutes = int(minutes_and_seconds_and_milliseconds // 60)
    seconds_and_milliseconds = minutes_and_seconds_and_milliseconds % 60
    seconds = int(seconds_and_milliseconds // 1)
    milliseconds = int((seconds_and_milliseconds % 1) * 1000)
    return f"{hours}h{minutes}m{seconds}s{milliseconds}ms"


runtime_file_path: str = os.path.join(os.environ["RUN_DIR"], "runtime.yaml")


def read_runtime_yaml():
    runtime_data_str: str = "[]"
    try:
        runtime_data_str = open(runtime_file_path).read()
    except FileNotFoundError:
        pass

    yaml_docs: List[List] = list(yaml.safe_load_all(runtime_data_str))

    return yaml_docs[0]


def write_runtime(status: str, time_in: float):
    runtime_data = read_runtime_yaml()

    obj = {}
    obj["status"] = f"{os.environ['CURRENT_INDEX']} - {status}"
    obj["runtime_s"] = round(time_in, 2)
    obj["runtime_ts"] = seconds_to_timestamp(time_in)

    runtime_data.append(obj)

    with open(runtime_file_path, "w") as f:
        f.write(yaml.safe_dump(runtime_data, sort_keys=False))


def conclude_run(status: str, time_in: float):
    runtime_data = read_runtime_yaml()

    final_runtime_data = []

    timer_start = os.getenv("timer_start")
    if timer_start is None:
        raise Exception(
            "Attempted to conclude a run without timer_start environment variable set"
        )
    timer_start = float(timer_start)

    timer_routed = os.getenv("timer_routed")
    routed_s = -1
    routed_ts = "-1"
    if timer_routed is not None:
        timer_routed = float(timer_routed)
        routed_s = timer_routed - timer_start
        routed_ts = seconds_to_timestamp(routed_s)

    obj = {}
    obj["status"] = "routed"
    obj["runtime_s"] = routed_s
    obj["runtime_ts"] = routed_ts
    final_runtime_data.append(obj)

    done_s = time_in - timer_start
    done_ts = seconds_to_timestamp(done_s)

    obj = {}
    obj["status"] = status
    obj["runtime_s"] = done_s
    obj["runtime_ts"] = done_ts
    final_runtime_data.append(obj)

    with open(runtime_file_path, "w") as f:
        f.write(yaml.safe_dump_all([runtime_data, final_runtime_data], sort_keys=False))


@click.command("write_runtime")
@click.option("--conclude/--no-conclude", default=False)
@click.option(
    "--seconds/--timestamp",
    "seconds",
    default=False,
    help="Process the input string either as a timestamp or a floating point second value",
)
@click.option(
    "--time-in",
    default=None,
    help="If this argument is not specified, time is read from stdin.",
)
@click.argument("status")
def cli(seconds, conclude, time_in, status):
    if time_in is None:
        time_in = sys.stdin.read().rstrip()

    if seconds:
        time_in = float(time_in)
    else:
        temp = timestamp_to_seconds(time_in)
        if temp is None:
            raise Exception(f"Invalid timestamp {time_in}")
        time_in = temp

    if conclude:
        conclude_run(status, time_in)
    else:
        write_runtime(status, time_in)


if __name__ == "__main__":
    cli()