"""
SKA telescope model command line utility.
Usage:
ska-telmodel [-vULS<uris>] cat [<key>]
ska-telmodel [-vULS<uris>] cp [-R] <key> [<path>]
ska-telmodel [-vULS<uris>] ls [<prefix>]
ska-telmodel [-vULS<uris>] pin
ska-telmodel [-vULS<uris>] validate [-tR] <key/path>
ska-telmodel help [<command>]
Options:
-L, --local Equivalent to "--sources=file://."
-R, --recursive Copy / validate keys or files recursively
-S <uris>, --sources <uris>
Set telescope model data sources of truth
(','-separated list of URIs)
-t, --strict Strict validation mode
-U, --update Update source list
-v, --verbose Verbose mode
"""
import logging
import os
import pathlib
import shutil
import sys
from inspect import cleandoc
import docopt
from ska_telmodel.data import TMData
from ska_telmodel.data.frontend import TMDataBackend
from ska_telmodel.data.sources import DEFAULT_SOURCES
from ska_telmodel.schema import validate
logger = logging.getLogger(__name__)
class CustomFormatter(logging.Formatter):
# Adapted from https://stackoverflow.com/questions/
# 384076/how-can-i-color-python-logging-output
white = "\x1b[1m"
grey = "\x1b[38;20m"
yellow = "\x1b[33;20m"
red = "\x1b[31;20m"
bold_red = "\x1b[31;1m"
reset = "\x1b[0m"
format_normal = "%(message)s"
format_verbose = (
"%(asctime)s - %(name)s - %(levelname)s - "
"%(message)s (%(filename)s:%(lineno)d)"
)
def __init__(self, verbose):
format_str = self.format_verbose if verbose else self.format_normal
self.formatters = {
logging.DEBUG: logging.Formatter(
self.grey + format_str + self.reset
),
logging.INFO: logging.Formatter(
(self.white if verbose else self.grey)
+ format_str
+ self.reset
),
logging.WARNING: logging.Formatter(
self.yellow + format_str + self.reset
),
logging.ERROR: logging.Formatter(
self.red + format_str + self.reset
),
logging.CRITICAL: logging.Formatter(
self.bold_red + format_str + self.reset
),
}
def format(self, record):
return self.formatters[record.levelno].format(
record
) # pragma: no cover
def main():
# Parse command line
args = docopt.docopt(__doc__)
# Initialise logging (globally)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(CustomFormatter(args["--verbose"]))
logging.basicConfig(
level=(logging.DEBUG if args["--verbose"] else logging.INFO),
handlers=[ch],
)
logging.captureWarnings(True)
# Get sources from command line, then sources
srcs = args.get("--sources")
if not srcs:
if args.get("--local"):
srcs = "file://."
else:
srcs = os.getenv("SKA_TELMODEL_SOURCES")
if srcs:
srcs = srcs.split(",")
else:
srcs = DEFAULT_SOURCES
# help?
if args["help"]:
return cmd_help(args)
data = TMData(source_uris=srcs, update=args["--update"])
# list?
if args["ls"]:
return cmd_ls(args, data)
# pin?
if args["pin"]:
return cmd_pin(args, data)
# cat?
if args["cat"]:
return cmd_cat(args, data)
# copy?
if args["cp"]:
return cmd_cp(args, data)
# validate?
if args["validate"]:
return cmd_validate(args, data)
[docs]def cmd_ls(args, data):
"""
List telescope model keys with a particular prefix
Usage::
ska-telmodel [-vUs<uris>] ls [<prefix>]
"""
path = args.get("<prefix>")
if path:
data = data[path]
for entry in data:
print(entry)
[docs]def cmd_pin(args, data):
"""
Generates a "pinned" telescope model data source list, where all
URIs replaced such that they will uniquely identify the contents
of the telescope model data repository.
Usage::
ska-telmodel [-vUs<uris>] pin
After pinning, the source list precisely identifies the contents
of the all telescope model data. For instance, this will replace
GitLab URIs like ``gitlab://gitlab.com/grp/proj#path`` with
``gitlab://gitlab.com/grp/proj?[commit]#path``, therefore baking in
the exact commit referenced. You can set pinned sources in the
environment as follows::
$ export $(ska-telmodel pin -U)
$ export $(ska-telmodel pin -US [custom sources])
This will especially prevent the ska-telmodel tool from
infrequently (once a day) re-checking whether cached telescope
model data contents is still current. The ``-U`` flag forces
the cache refresh, which is generally a good idea before pinning.
"""
src_uris = data.get_sources(True)
for uri in src_uris:
logging.info("Using %s", uri)
print(f'SKA_TELMODEL_SOURCES={",".join(src_uris)}')
[docs]def cmd_cat(args, data):
"""
Retrieves and prints the telescope model data identifed by the
given key to stdout.
Usage::
ska-telmodel [-vUs<uris>] cat [<key>]
Use ``ska-telmodel ls`` to obtain a list of valid keys.
How exactly the object is retrieved depends on the backend and
the state of the cache. For a GitLab backend, the typical
behaviour is to download a tarball either from the SKAO
central artefact repository, or from GitLab directly. The
latter should be avoided and will generate a warning.
"""
with data[args["<key>"]].open() as f:
shutil.copyfileobj(f, sys.stdout.buffer)
[docs]def cmd_cp(args, srcs):
"""
Retrieves specified telescope model data, and copies it to
the given path.
Usage::
ska-telmodel [-vUs<uris>] cp [-R] <key> [<path>]
If ``-R`` is given, the key can be a key directory, in which all
keys that start with ``<path>/`` will be copied. Note that you
can especially give the empty string (``""``) as ``<key>``, in which
case all available telescope model data will be copied.
This is especially useful for serving telescope model data
either partially or completely from storage. For instance::
$ ska-telmodel cp -UR "" tmdata
$ export SKA_TELMODEL_SOURCES=file://$(pwd)/tmdata
Would completely mirror the telescope model to the given
location.
"""
data = TMData(source_uris=srcs)
src_path = args["<key>"] or ""
dest_path = pathlib.Path(args["<path>"] or ".")
# Individual file?
if not args["--recursive"] or TMDataBackend.valid_key(src_path):
if dest_path.is_dir():
dest_path = pathlib.Path(dest_path, pathlib.Path(src_path).name)
data[src_path].copy(dest_path)
print(src_path)
return
# Global?
if src_path:
src_data = data[src_path]
else:
src_data = data
# Copy files recursively
for key in src_data:
dest = pathlib.Path(dest_path, key)
dest.parent.mkdir(parents=True, exist_ok=True)
src_data[key].copy(dest)
print(pathlib.Path(src_path, key))
[docs]def cmd_validate(args, srcs):
"""
Validates given keys (or files) against applicable schemas from
the telescope model library
Usage::
ska-telmodel [-vUs<uris>] validate [-tlR] [<key/path>]*
If ``-R`` is given, the key can be a key directory, in which all
keys that start with ``<path>/`` will be copied. Note that you
can especially give the empty string (``""``) as ``<key>``, in which
case all available telescope model data will be copied.
This is especially useful for serving telescope model data
either partially or completely from storage. For instance::
$ ska-telmodel cp -R "" tmdata
$ export SKA_TELMODEL_SOURCES=file://$(pwd)/tmdata
Would completely mirror the telescope model to the given
location.
"""
data = TMData(source_uris=srcs)
src_path = args["<key/path>"] or ""
# Individual file?
if not args["--recursive"] or TMDataBackend.valid_key(src_path):
keys_to_check = [src_path]
else:
# Global?
if src_path:
keys_to_check = [f"{src_path}/{key}" for key in data[src_path]]
else:
keys_to_check = list(data)
# Start checking keys
success = True
for key in keys_to_check:
logger.info(f" * Checking {key}:")
# Attempt to deserialise
try:
dct = data[key].get_dict()
except ValueError as e:
logger.error(data[key].get())
logger.error(f" {e}")
success = False
continue
# Check whether we can validate
validated = False
if "interface" in dct:
try:
logger.info(dct["interface"])
validate(dct["interface"], dct, 1 + args["--strict"])
validated = True
except ValueError as e:
logger.error(f"{e}")
success = False
continue
# Fail if we do not find any method for validation
if not validated:
logger.error("No schema to use for validation!")
success = False
# Fail if appropriate
if not success:
exit(1)
def cmd_help(args):
subject = args["<command>"]
# Get docstrings
topics = {}
for cmd in [cmd_cat, cmd_cp, cmd_pin]:
topics[cmd.__name__[4:]] = cleandoc(cmd.__doc__)
# If subject known: Print docstring
if subject in topics:
for line in topics[subject].split("\n"):
print(line)
return
# If subject is unknwon, show general help
if subject:
logger.warning(f"Unknown help topic {subject}!")
help_topic = __doc__ + "\n"
help_topic += cleandoc(
f"""
Available help topics:
{', '.join(topics)}
"""
)
for line in help_topic.split("\n"):
print(line)
if __name__ == "__main__":
main() # pragma: no cover