"""
filepath.py
File and filename path utilities.
"""
import datetime
import errno
import json
import io
import os
import subprocess
from typing import Any, Optional, Generator
import wholecell
TIMEOUT = 60 # seconds
# The wcEcoli/ project root path which contains wholecell/.
ROOT_PATH = os.path.dirname(os.path.dirname(os.path.realpath(wholecell.__file__)))
OUT_DIR = os.path.join(ROOT_PATH, "out")
DEBUG_OUT_DIR = os.path.join(OUT_DIR, "debug")
MATPLOTLIBRC_FILE = os.path.join(ROOT_PATH, "matplotlibrc")
# Regex for current and previous timestamp() formats: 'YYYYMMDD.HHMMSS[.uuuuuu]'.
TIMESTAMP_PATTERN = r"\d{8}\.\d{6}(?:\.\d{6})?"
[docs]
def makedirs(path: str, *paths: str) -> str:
"""Join one or more path components, make that directory path (using the
default mode 0o0777), and return the full path.
Raise FileExistsError if there's a file (not a directory) with that path.
No exception if the directory already exists.
"""
full_path = os.path.join(path, *paths)
if full_path:
os.makedirs(full_path, exist_ok=True)
return full_path
[docs]
def timestamp(dt: Optional[datetime.datetime] = None) -> str:
"""Construct a datetime-timestamp from `dt` [default = `now()`], such as
we use to timestamp a simulation output directory.
"""
if not dt:
dt = datetime.datetime.now()
return dt.strftime("%Y%m%d.%H%M%S")
[docs]
def verify_file_exists(file_path: str, message: str = ""):
"""Raise an IOError if file_path isn't an existing file."""
if not os.path.isfile(file_path):
raise IOError(errno.ENOENT, 'Missing file "{}". {}'.format(file_path, message))
[docs]
def verify_dir_exists(dir_path: str, message: str = ""):
"""Raise an IOError if dir_path isn't an existing directory."""
if not os.path.isdir(dir_path):
raise IOError(errno.ENOENT, 'Missing dir "{}". {}'.format(dir_path, message))
[docs]
def run_cmd2(
tokens: list[str],
trim: bool = True,
timeout: int = TIMEOUT,
env: Optional[dict] = None,
input_: Optional[str] = None,
) -> tuple[str, str]:
"""Run a shell command-line (in token list form) and return a tuple
containing its (stdout, stderr).
This does not expand filename patterns or environment variables or do other
shell processing steps.
Args:
tokens: The command line as a list of string tokens.
trim: Whether to trim off trailing whitespace. This is useful
because the outputs usually end with a newline.
timeout: timeout in seconds; None for no timeout.
env: optional environment variables for the new process to use instead
of inheriting the current process' environment.
input_: input for any prompts that may appear (passed to the subprocess' stdin)
Returns:
The command's stdout and stderr strings.
Raises:
OSError (e.g. FileNotFoundError [Python 3] or PermissionError),
subprocess.SubprocessError (TimeoutExpired or CalledProcessError)
"""
out = subprocess.run(
tokens,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
env=env,
encoding="utf-8",
timeout=timeout,
input=input_,
)
if trim:
return out.stdout.rstrip(), out.stderr.rstrip()
return out.stdout, out.stderr
[docs]
def run_cmd(
tokens: list[str],
trim: bool = True,
timeout: int = TIMEOUT,
env: Optional[dict] = None,
input_: Optional[str] = None,
) -> str:
"""Run a shell command-line (in token list form) and return its stdout.
See run_cmd2().
"""
return run_cmd2(tokens, trim=trim, timeout=timeout, env=env, input_=input_)[0]
[docs]
def run_cmdline(
line: str,
trim: bool = True,
timeout: int = TIMEOUT,
input_: Optional[str] = None,
fallback: Optional[str] = None,
) -> Optional[str]:
"""Run a shell command-line string then return its output or fallback if it
failed. This does not expand filename patterns or environment variables or
do other shell processing steps like quoting.
Args:
line: The command line as a string to split.
trim: Whether to trim off trailing whitespace. This is useful
because the subprocess output usually ends with a newline.
timeout: timeout in seconds; None for no timeout.
input_: input for any prompts that may appear (passed to the subprocess' stdin)
fallback: Return this if the subprocess fails, e.g. trying to run git
in a Docker Image that has no git repo.
Returns:
The command's output string, or None if it couldn't even run.
"""
try:
return run_cmd(tokens=line.split(), trim=trim, input_=input_, timeout=timeout)
except (OSError, subprocess.SubprocessError) as e:
if fallback is None:
print("failed to run command line {}: {}".format(line, e))
return fallback
[docs]
def git_hash():
"""Return the source code git hash, or the environment variable
$IMAGE_GIT_HASH if there's no git repo (in a Docker Image), or else '--'.
"""
return run_cmdline(
"git rev-parse HEAD", fallback=os.environ.get("IMAGE_GIT_HASH", "--")
)
[docs]
def git_branch():
"""Return the source code git branch name, or the environment variable
$IMAGE_GIT_BRANCH if there's no git repo (in a Docker Image), or '--'.
"""
return run_cmdline(
"git symbolic-ref --short HEAD",
fallback=os.environ.get("IMAGE_GIT_BRANCH", "--"),
)
[docs]
def write_file(filename: str, content: str):
"""Write text string `content` as a utf-8 text file."""
with io.open(filename, "w", encoding="utf-8") as f:
f.write(str(content))
[docs]
def write_json_file(filename: str, obj: Any, indent: int = 4):
"""Write `obj` to a file in a pretty JSON format. This supports Unicode."""
# Indentation puts a newline after each ',' so suppress the space there.
message = (
json.dumps(
obj,
ensure_ascii=False,
indent=indent,
separators=(",", ": "),
sort_keys=True,
)
+ "\n"
)
write_file(filename, message)
[docs]
def read_json_file(filename: str) -> Any:
"""Read and parse JSON file. This supports Unicode."""
with io.open(filename, encoding="utf-8") as f:
return json.load(f)
[docs]
def iter_variants(
variant_type: str, first_index: int, last_index: int
) -> Generator[tuple[int, str], None, None]:
"""Generate Variant subdirs (index, name) over [first .. last] inclusive."""
# TODO(jerry): Return a list instead of generating items?
for i in range(first_index, last_index + 1):
yield i, os.path.join("{}_{:06d}".format(variant_type, i))