200 lines
7.8 KiB
Python
200 lines
7.8 KiB
Python
import json
|
|
import sys
|
|
|
|
import click
|
|
from pathlib import Path
|
|
import logging
|
|
import re
|
|
|
|
from iottb import definitions
|
|
from iottb.models.device_metadata import DeviceMetadata
|
|
from iottb.models.iottb_config import IottbConfig
|
|
from iottb.definitions import CFG_FILE_PATH, TB_ECHO_STYLES
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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, 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, default="")
|
|
@click.option('--db', '--database', type=str,
|
|
envvar='IOTTB_DB', show_envvar=True, default="",
|
|
help='Database in which to add this device. If not specified use default from config.')
|
|
@click.option('--guided', is_flag=True,
|
|
help='Add device interactively')
|
|
def add_device(device, db, guided):
|
|
"""Add a new device to a database
|
|
|
|
Device name must be supplied unless in an interactive setup.
|
|
Database is taken from config by default.
|
|
If this device name contains spaces or other special characters normalization is performed to derive a canonical name.
|
|
"""
|
|
logger.info('add-device invoked')
|
|
|
|
# Step 1: Load Config
|
|
# 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 != "":
|
|
database = db
|
|
path = config.db_path_dict[database]
|
|
logger.debug(f'Resolved (path, db) {path}, {database}')
|
|
else:
|
|
path = config.default_db_location
|
|
database = config.default_database
|
|
logger.debug(f'Default (path, db) {path}, {database}')
|
|
click.secho(f'Using database {database}')
|
|
full_db_path = Path(path) / database
|
|
if not full_db_path.is_dir():
|
|
logger.warning(f'No database at {database}')
|
|
click.echo(f'No database found at {full_db_path}', lvl='w')
|
|
click.echo(
|
|
f'You need to initialize the testbed before before you add devices!')
|
|
click.echo(
|
|
f'To initialize the testbed in the default location run "iottb init-db"')
|
|
click.echo('Exiting...')
|
|
sys.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=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 {device} already exists in the database.')
|
|
click.echo('Exiting...')
|
|
sys.exit()
|
|
try:
|
|
device_dir.mkdir()
|
|
except OSError as e:
|
|
logger.error(f'Error trying to create device {e}')
|
|
click.echo('Exiting...')
|
|
sys.exit()
|
|
|
|
# Step 4: Save metadata into device_dir
|
|
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 {device} to database')
|
|
logger.debug(f'Added device {device} to database {database}. Full path of metadata {metadata_path}')
|
|
logger.info(f'Metadata for {device} {device_metadata.print_attributes()}')
|