import json import uuid from datetime import datetime from pathlib import Path from typing import Optional, List, Any # 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 = lambda: str(uuid.uuid4()) date_created: str = lambda: datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower() 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_dir: Path): self.device_name = device_name self.device_short_name = device_name.lower().replace(" ", "_") # assert dir_contains_device_metadata(device_root_dir), \ # f"Directory {device_root_dir} is missing a {DEVICE_METADATA_FILE} file" self.device_root_dir = device_root_dir 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_dir}") 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": True, "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: raise ValueError(f"Field {field} is required and cannot be empty.") data[field] = str(value) if not isinstance(value, Path) 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