#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
This command line tool creates all the files for a chosen device
that would normally be either pushed to that device or created locally
in an output directory specified by the command line parameters.
By default, the tool will use the unmodified default device
configuration, but this can optionally be modified by supplying a
valid import database source. Initially only XML import with optional
Kajiki_ template statements is supported.
example: ``-x my.xml``
Properties can be defined on the command line to support passing text
definitions to Kajiki_ and similar templating engines.
example: ``-p variant=soundbar colour=blue``
.. _Kajiki: https://kajiki.readthedocs.io/en/latest/xml-templates.html
"""
#
# Copyright Qualcomm Technologies Inc, 2019.
# All Rights Reserved
#
# Python imports
from pathlib import Path
import sys
import logging
import logging.config
import json
from argparse import ArgumentParser, Action, ArgumentTypeError
from argparse import RawDescriptionHelpFormatter
from shutil import copy
# Thirdparty imports
from zope.component import getUtility
import transaction
# Imports from local packages
from qct_interfaces import load_components
from qct_interfaces.database import IDatabaseConverter
from configurationmodel import DeviceDatabase, ConfigAttributeBase
from pyconfigtool.utils import locate_resource, get_device_properties, uri_to_dev
from pyconfigtool import __version__
DEVICE_DBNAME = 'device_db'
#### Command line argument processing ####
[docs]class ReadableFile(Action):
"""Argparse action to accept a readable file path.
:raises: ArgumentTypeError
"""
def __call__(self, parser, namespace, values, option_string=None):
pth = Path(values)
if pth.exists():
setattr(namespace, self.dest, pth)
else:
raise ArgumentTypeError("{VALUES} is not readable".format(VALUES=values))
[docs]class WriteableDir(Action):
"""Argparse action to accept a writeable directory."""
def __call__(self, parser, namespace, values, option_string=None):
pth = Path(values)
if not pth.is_dir():
pth.mkdir(parents=True)
setattr(namespace, self.dest, pth)
[docs]class Properties(Action):
"""Argparse action to accept a readable file.
:raises: ArgumentTypeError
"""
def __call__(self, parser, namespace, values, option_string=None):
try:
# pylint: disable=consider-using-dict-comprehension
setattr(namespace, self.dest, dict([_.split("=") for _ in values]))
except ValueError:
raise ArgumentTypeError("-p {VALUES} not understood".format(VALUES=values))
[docs]def arg_parser() -> None:
"""Parse command line arguments. Use --help to see usage at the
command line to see help.
"""
parser = ArgumentParser(
description=__doc__, add_help=True, formatter_class=RawDescriptionHelpFormatter)
parser.set_defaults(loglevel=logging.INFO)
# Device Selection
device_group = parser.add_argument_group("Device selection")
# Use an existing device database
choose_device_group = device_group.add_mutually_exclusive_group()
choose_device_group.add_argument(
'-f', '--database-file', action=ReadableFile,
help="Open a device using its Device Database (.fs) file")
# Create a device database
choose_device_group.add_argument(
'-u', '--uri', default='device://test/testdev3',
help="Create a device using its Unique Resource Identifier\nDefault: %(default)s")
# Use an import file or template to create the device database
import_group = parser.add_argument_group("Import configuration set")
import_group.add_argument(
'-x', '--xml-file', action=ReadableFile,
help="Use configuration from an XML file or template to create the device database.")
# Add extra properties to templated XML parser
import_group.add_argument(
'-p', '--properties', default=dict(), nargs="+", action=Properties,
help="""\
One or more properties to extend or replace the device
properties in a templated import file.
Example: -p device_type=test device_name=testdev1""")
# Log verbosity
log_group = parser.add_mutually_exclusive_group()
log_group.add_argument(
'-v', '--verbose', help='Verbose (debug) logging', action='store_const',
const=logging.DEBUG, dest='loglevel')
log_group.add_argument(
'-q', '--quiet', help='Show warnings and errors only', action='store_const',
const=logging.WARN, dest='loglevel')
# Common parameters
parser.add_argument(
'-o', '--output-dir', action=WriteableDir,
default=Path("./configcmd_output"),
help="""\
The output directory where all the device files will be
sent. Device files will be named as their remotepath or
localpath. An exported XML file ({DEVICE_DBNAME}.xml)
and created device database ({DEVICE_DBNAME}.fs)
will also be placed here.""".format(DEVICE_DBNAME=DEVICE_DBNAME))
parser.add_argument(
'--version', action='version', version='%(prog)s {__VERSION__}'.format(
__VERSION__=__version__))
return parser
#### Main Program ####
[docs]def main() -> int:
"""Generates all the files that would normally be pushed to a device or
created locally in the output directory specified on the command line.
Also exports the database in supported export formats.
:return: Success code
"""
# Use the GUI tools logging module configuation
log_ini = Path(locate_resource('assets/logging.ini'))
if log_ini.is_file():
logging.config.fileConfig(str(log_ini))
else:
logging.basicConfig(format='%(asctime)-15s %(message)s')
# Read the command line arguments
args = arg_parser().parse_args()
# Update the logging level
logging.getLogger().setLevel(args.loglevel)
# Load plug-ins
load_components()
def _replace_metadata_writers_callback(config_domain_item):
"""The location(s) a particular data item are emitted to is defined in the
``writers`` portion of the metadata for ``ConfigAttributeBase`` instances.
This function alters that metadata to emit the item to the output_directory
defined at the command line.
:param config_domain_item: A config domain item in the database.
:return: None
"""
if not isinstance(config_domain_item, ConfigAttributeBase):
return
metadata = json.loads(config_domain_item.attr_meta_data or '{}')
writer_list = metadata.get("writers", [])
for writers in writer_list:
for writer in writers:
try:
pth = writers[writer].get("remotepath") or writers[writer].get("localpath")
filename = Path(pth).parts[-1]
writers[writer].clear()
writers[writer].update(localpath=str(args.output_dir / filename))
except TypeError:
logging.debug("No writer for %s", config_domain_item.attr_key)
config_domain_item.attr_meta_data = json.dumps(metadata)
# Get Converter utility plug-ins
supported_converters = ['XML', 'YAML']
converters = {_.lower():getUtility(IDatabaseConverter, name=_) for _ in supported_converters}
# Get the device plug-in and populate device properties
device = uri_to_dev(args.uri)
properties = get_device_properties(args.uri)
properties.update(args.properties)
logging.debug("Properties: %s", properties)
if not args.output_dir.is_dir():
args.output_dir.mkdir(parents=True)
if args.database_file:
# Open the specified device database
dbsrc = args.database_file
dbpath = args.output_dir / dbsrc.name
logging.info("Opening device database: %s", str(dbpath))
copy(str(dbsrc), str(dbpath))
else:
# Create the default device database in the output directory
dbpath = args.output_dir / '{DEVICE_DBNAME}.fs'.format(DEVICE_DBNAME=DEVICE_DBNAME)
if dbpath.exists():
dbpath.unlink()
logging.info("Creating device database: %s", str(dbpath))
device.create_db(dbpath, properties)
database = DeviceDatabase(dbpath)
# Create the output files
with database as root:
# Import the database from an alternative XML template or file if required
if args.xml_file:
is_successful = converters["xml"].import_to_database(args.xml_file, root, properties)
if not is_successful:
logging.debug("Failed to import to database xml file")
return 1
# Export the database file now before we change anything
for converter in converters:
converters[converter].export_from_database(
root, args.output_dir / '{DEVICE_DBNAME}.{CONVERTER}'.format(
DEVICE_DBNAME=DEVICE_DBNAME, CONVERTER=converter))
# Save the created database
savepoint = transaction.savepoint()
# Change the config item metadata writer methods to send their
# output to the output directory
database.walk_database(_replace_metadata_writers_callback)
# Push the config items to the local output drectory
device.scatter(database)
# And restore the original metadata
savepoint.rollback()
# Sucess
return 0
###########################
if __name__ == '__main__':
# Run the tool
sys.exit(main())