108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
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):
|
|
pass
|
|
|
|
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
|