import json import uuid from datetime import datetime from pathlib import Path from typing import Optional, List # iottb modules from iottb.definitions import ReturnCodes, DEVICE_METADATA_FILE from iottb.logger import logger # 3rd party libs IMMUTABLE_FIELDS = {'device_name', 'device_short_name', 'device_id', 'date_created'} class DeviceMetadata: # Required fields device_name: str device_short_name: str device_id: str date_created: str device_root_path: Path # Optional Fields aliases: Optional[List[str]] = None device_type: Optional[str] = None device_serial_number: Optional[str] = None device_firmware_version: Optional[str] = None date_updated: Optional[str] = None capture_files: Optional[List[str]] = [] def __init__(self, device_name: str, device_root_path: Path): self.device_name = device_name self.device_short_name = device_name.lower().replace(' ', '_') self.device_id = str(uuid.uuid4()) self.date_created = datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower() self.device_root_path = device_root_path if not self.device_root_path or not self.device_root_path.is_dir(): logger.error(f'Invalid device root path: {device_root_path}') raise ValueError(f'Invalid device root path: {device_root_path}') logger.debug(f'Device name: {device_name}') logger.debug(f'Device short_name: {self.device_short_name}') logger.debug(f'Device root dir: {device_root_path}') logger.info(f'Initialized DeviceMetadata model: {device_name}') @classmethod def load_from_json(cls, device_file_path: Path): logger.info(f'Loading DeviceMetadata from JSON file: {device_file_path}') assert device_file_path.is_file(), f'{device_file_path} is not a file' assert device_file_path.name == DEVICE_METADATA_FILE, f'{device_file_path} is not a {DEVICE_METADATA_FILE}' device_meta_filename = device_file_path with device_meta_filename.open('r') as file: metadata_json = json.load(file) metadata_model_obj = cls.from_json(metadata_json) return metadata_model_obj def save_to_json(self, file_path: Path): logger.info(f'Saving DeviceMetadata to JSON file: {file_path}') if file_path.is_file(): print(f'File {file_path} already exists, update instead.') return ReturnCodes.FILE_ALREADY_EXISTS metadata = self.to_json(indent=2) with file_path.open('w') as file: json.dump(metadata, file) return ReturnCodes.SUCCESS @classmethod def from_json(cls, metadata_json): if isinstance(metadata_json, dict): return DeviceMetadata(**metadata_json) def to_json(self, indent=2): # TODO: atm almost exact copy as in CaptureMetadata data = {} fields = { 'device_name': True, 'device_short_name': True, 'device_id': True, 'date_created': True, 'device_root_path': True, 'aliases': False, 'device_type': False, 'device_serial_number': False, 'device_firmware_version': False, 'date_updated': False, 'capture_files': False, } for field, is_mandatory in fields.items(): value = getattr(self, field, None) if value not in [None, ''] or is_mandatory: if value in [None, ''] and is_mandatory: logger.debug(f'Mandatory field {field}: {value}') raise ValueError(f'Field {field} is required and cannot be empty.') data[field] = str(value) if not isinstance(value, str) else value logger.debug(f'Device metadata: {data}') return json.dumps(data, indent=indent) def dir_contains_device_metadata(dir_path: Path): if not dir_path.is_dir(): return False else: meta_file_path = dir_path / DEVICE_METADATA_FILE print(f'Device metadata file path {str(meta_file_path)}') if not meta_file_path.is_file(): return False else: return True