blob: cc71c7f9be016831b753c5e405785ade396184e6 [file] [log] [blame]
# 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()