Add guided device add functionality.
This commit is contained in:
Sebastian Lenzlinger
parent
1b5c324c50
commit
c5b8ea42e7
@ -13,18 +13,117 @@ from iottb.definitions import CFG_FILE_PATH, TB_ECHO_STYLES
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def add_device_guided(ctx, cn, db):
|
||||
click.echo('TODO: Implement')
|
||||
def prompt_for_device_details():
|
||||
device_details = {}
|
||||
aliases = []
|
||||
while True:
|
||||
click.echo("\nEnter the details for the new device:")
|
||||
click.echo("1. Device Name")
|
||||
click.echo("2. Description")
|
||||
click.echo("3. Model")
|
||||
click.echo("4. Manufacturer")
|
||||
click.echo("5. Current Firmware Version")
|
||||
click.echo("6. Device Type")
|
||||
click.echo("7. Supported Interfaces")
|
||||
click.echo("8. Companion Applications")
|
||||
click.echo("9. Add Alias")
|
||||
click.echo("10. Finish and Save")
|
||||
|
||||
choice = click.prompt("Choose an option", type=int)
|
||||
|
||||
if choice == 1:
|
||||
device_details['device_name'] = click.prompt("Enter the device name")
|
||||
elif choice == 2:
|
||||
device_details['description'] = click.prompt("Enter the description")
|
||||
elif choice == 3:
|
||||
device_details['model'] = click.prompt("Enter the model")
|
||||
elif choice == 4:
|
||||
device_details['manufacturer'] = click.prompt("Enter the manufacturer")
|
||||
elif choice == 5:
|
||||
device_details['firmware_version'] = click.prompt("Enter the current firmware version")
|
||||
elif choice == 6:
|
||||
device_details['device_type'] = click.prompt("Enter the device type")
|
||||
elif choice == 7:
|
||||
device_details['supported_interfaces'] = click.prompt("Enter the supported interfaces")
|
||||
elif choice == 8:
|
||||
device_details['companion_applications'] = click.prompt("Enter the companion applications")
|
||||
elif choice == 9:
|
||||
alias = click.prompt("Enter an alias")
|
||||
aliases.append(alias)
|
||||
elif choice == 10:
|
||||
break
|
||||
else:
|
||||
click.echo("Invalid choice. Please try again.")
|
||||
|
||||
device_details['aliases'] = aliases
|
||||
return device_details
|
||||
|
||||
|
||||
def confirm_and_add_device(device_details, db_path):
|
||||
click.echo("\nDevice metadata:")
|
||||
for key, value in device_details.items():
|
||||
click.echo(f"{key.replace('_', ' ').title()}: {value}")
|
||||
|
||||
confirm = click.confirm("Do you want to add this device with above metadata?")
|
||||
if confirm:
|
||||
device_name = device_details.get('device_name')
|
||||
if not device_name:
|
||||
click.echo("Device name is required. Exiting...")
|
||||
return
|
||||
|
||||
device_metadata = DeviceMetadata(**device_details)
|
||||
device_dir = db_path / device_metadata.canonical_name
|
||||
|
||||
if device_dir.exists():
|
||||
click.echo(f"Device {device_name} already exists in the database.")
|
||||
click.echo("Exiting...")
|
||||
return
|
||||
|
||||
try:
|
||||
device_dir.mkdir(parents=True, exist_ok=True)
|
||||
metadata_path = device_dir / definitions.DEVICE_METADATA_FILE_NAME
|
||||
device_metadata.save_metadata_to_file(metadata_path)
|
||||
click.echo(f"Successfully added device {device_name} to database.")
|
||||
except OSError as e:
|
||||
click.echo(f"Error trying to create device directory: {e}")
|
||||
click.echo("Exiting...")
|
||||
else:
|
||||
click.echo("Operation cancelled. Exiting...")
|
||||
|
||||
|
||||
def add_device_guided(cfg, device, db):
|
||||
logger.info('Adding device interactively')
|
||||
# logger.debug(f'Parameters: {params}. value: {value}')
|
||||
databases = cfg.db_path_dict
|
||||
if not databases:
|
||||
click.echo('No databases found in config file.')
|
||||
return
|
||||
click.echo('Available Databases:')
|
||||
last = 0
|
||||
for i, db_name in enumerate(databases.keys(), start=1):
|
||||
click.echo(f'[{i}] {db_name}')
|
||||
last = i if last < i else last
|
||||
db_choice = click.prompt('Select the database to add the new device to (1 - {last}, 0 to quit)',
|
||||
type=int, default=1)
|
||||
if 1 <= db_choice <= last:
|
||||
selected_db = list(databases.keys())[db_choice - 1]
|
||||
click.confirm(f'Use {selected_db}?', abort=True)
|
||||
db_path = Path(databases[selected_db]) / selected_db
|
||||
logger.debug(f'DB Path {str(db_path)}')
|
||||
device_details = prompt_for_device_details()
|
||||
confirm_and_add_device(device_details, db_path)
|
||||
elif db_choice == 0:
|
||||
click.echo(f'Quitting...')
|
||||
else:
|
||||
click.echo(f'{db_choice} is not a valid choice. Please rerun command and select a valid database.')
|
||||
|
||||
|
||||
@click.command('add-device', help='Add a device to a database')
|
||||
@click.argument('device', type=str, required=True)
|
||||
@click.argument('device', type=str, default="")
|
||||
@click.option('--db', '--database', type=click.Path(exists=True, file_okay=False, dir_okay=True),
|
||||
envvar='IOTTB_DB', show_envvar=True,
|
||||
help='Database in which to add this device. If not specified use default from config.')
|
||||
@click.option('--guided', is_flag=True, default=False, show_default=True, envvar='IOTTB_GUIDED_ADD', show_envvar=True,
|
||||
@click.option('--guided', is_flag=True,
|
||||
help='Add device interactively')
|
||||
def add_device(device, db, guided):
|
||||
"""Add a new device to a database
|
||||
@ -39,10 +138,16 @@ def add_device(device, db, guided):
|
||||
# Dependency: Config file must exist
|
||||
config = IottbConfig(Path(CFG_FILE_PATH))
|
||||
logger.debug(f'Config loaded: {config}')
|
||||
# If guided flag set, continue with guided add and leave
|
||||
if guided:
|
||||
click.echo('Guided option set. Continuing with guided add.')
|
||||
add_device_guided(config, device, db)
|
||||
logger.info('Finished guided device add.')
|
||||
return
|
||||
|
||||
# Step 2: Load database
|
||||
# dependency: Database folder must exist
|
||||
if db:
|
||||
if db != "":
|
||||
database = db
|
||||
path = config.db_path_dict[database]
|
||||
logger.debug(f'Resolved (path, db) {path}, {database}')
|
||||
@ -61,16 +166,20 @@ def add_device(device, db, guided):
|
||||
f'To initialize the testbed in the default location run "iottb init-db"')
|
||||
click.echo('Exiting...')
|
||||
exit()
|
||||
# Ensure a device name was passed as argument
|
||||
if device == "":
|
||||
click.echo("Device name cannot be an empty string. Exiting...", lvl='w')
|
||||
return
|
||||
|
||||
# Step 3: Check if device already exists in database
|
||||
# dependency: DeviceMetadata object
|
||||
device_metadata = DeviceMetadata(device_name=dev)
|
||||
device_metadata = DeviceMetadata(device_name=device)
|
||||
device_dir = full_db_path / device_metadata.canonical_name
|
||||
|
||||
# Check if device is already registered
|
||||
if device_dir.exists():
|
||||
logger.warning(f'Device directory {device_dir} already exists.')
|
||||
click.echo(f'Device {dev} already exists in the database.')
|
||||
click.echo(f'Device {device} already exists in the database.')
|
||||
click.echo('Exiting...')
|
||||
exit()
|
||||
try:
|
||||
@ -84,7 +193,7 @@ def add_device(device, db, guided):
|
||||
metadata_path = device_dir / definitions.DEVICE_METADATA_FILE_NAME
|
||||
with metadata_path.open('w') as metadata_file:
|
||||
json.dump(device_metadata.__dict__, metadata_file, indent=4)
|
||||
click.echo(f'Successfully added device {dev} to database')
|
||||
logger.debug(f'Added device {dev} to database {
|
||||
click.echo(f'Successfully added device {device} to database')
|
||||
logger.debug(f'Added device {device} to database {
|
||||
database}. Full path of metadata {metadata_path}')
|
||||
logger.info(f'Metadata for {dev} {device_metadata.print_attributes()}')
|
||||
logger.info(f'Metadata for {device} {device_metadata.print_attributes()}')
|
||||
|
||||
@ -10,9 +10,12 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option('-d', '--dest', type=click.Path(), help='Location to put (new) iottb database')
|
||||
@click.option('-n', '--name', default=DB_NAME, type=str, help='Name of new database.')
|
||||
@click.option('--update-default/--no-update-default', default=True, help='If new db should be set as the new default')
|
||||
@click.option('-d', '--dest', type=click.Path(exists=True, file_okay=False, dir_okay=True),
|
||||
help='Location to put (new) iottb database')
|
||||
@click.option('-n', '--name', default=DB_NAME, type=str,
|
||||
help='Name of new database.')
|
||||
@click.option('--update-default/--no-update-default', default=True,
|
||||
help='If new db should be set as the new default')
|
||||
@click.pass_context
|
||||
def init_db(ctx, dest, name, update_default):
|
||||
logger.info('init-db invoked')
|
||||
@ -31,7 +34,7 @@ def init_db(ctx, dest, name, update_default):
|
||||
logger.debug(f'DB name {name} registered but does not exist.')
|
||||
if not dest:
|
||||
logger.info('No dest set, choosing default destination.')
|
||||
dest = Path(config.default_db_location).parent
|
||||
dest = Path(config.default_db_location)
|
||||
|
||||
db_path = Path(dest).joinpath(name)
|
||||
logger.debug(f'Full path for db {str(db_path)}')
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
@ -11,12 +12,12 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
class DeviceMetadata:
|
||||
def __init__(self, device_name, description="", model="", manufacturer="", firmware_version="", device_type="",
|
||||
supported_interfaces="", companion_applications="", save_to_file=None):
|
||||
supported_interfaces="", companion_applications="", save_to_file=None, aliases=None):
|
||||
self.device_id = str(uuid.uuid4())
|
||||
self.device_name = device_name
|
||||
cn, aliases = make_canonical_name(device_name)
|
||||
logger.debug(f'cn, aliases = {cn}, {str(aliases)}')
|
||||
self.aliases = aliases
|
||||
cn, default_aliases = make_canonical_name(device_name)
|
||||
logger.debug(f'cn, default aliases = {cn}, {str(default_aliases)}')
|
||||
self.aliases = default_aliases if aliases is None else default_aliases + aliases
|
||||
self.canonical_name = cn
|
||||
self.date_added = datetime.now().isoformat()
|
||||
self.description = description
|
||||
@ -42,3 +43,8 @@ class DeviceMetadata:
|
||||
print(f'Printing attribute value pairs in {__name__}')
|
||||
for attr, value in self.__dict__.items():
|
||||
print(f'{attr}: {value}')
|
||||
|
||||
def save_metadata_to_file(self, metadata_path):
|
||||
with open(metadata_path, 'w') as metadata_file:
|
||||
json.dump(self.__dict__, metadata_file, indent=4)
|
||||
click.echo(f'Metadata saved to {metadata_path}')
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user