129 lines
4.6 KiB
Python
129 lines
4.6 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
|
|
# 3rd party libs
|
|
from pydantic import BaseModel, Field
|
|
|
|
IMMUTABLE_FIELDS = {"device_name", "device_short_name", "device_id", "date_created"}
|
|
|
|
|
|
class DeviceMetadata(BaseModel):
|
|
# Required fields
|
|
device_name: str
|
|
device_short_name: str
|
|
device_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
|
|
date_created: str = Field(default_factory=lambda: datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower())
|
|
|
|
device_root_path: Path
|
|
# Optional Fields
|
|
aliases: List[str] = Field(default_factory=lambda: [])
|
|
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, /, **data: Any):
|
|
super().__init__(**data)
|
|
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
|
|
|
|
def get_device_id(self) -> str:
|
|
return self.device_id
|
|
|
|
def get_device_name(self) -> str:
|
|
return self.device_name
|
|
|
|
def get_device_short_name(self) -> str:
|
|
return self.device_short_name
|
|
|
|
def get_device_type(self) -> str:
|
|
return self.device_type
|
|
|
|
def get_device_serial_number(self) -> str:
|
|
return self.device_serial_number
|
|
|
|
def get_device_firmware_version(self) -> str:
|
|
return self.device_firmware_version
|
|
|
|
def get_date_updated(self) -> str:
|
|
return self.date_updated
|
|
|
|
def get_capture_files(self) -> List[str]:
|
|
return self.capture_files
|
|
|
|
def get_aliases(self) -> List[str]:
|
|
return self.aliases
|
|
|
|
def set_device_type(self, device_type: str) -> None:
|
|
self.device_type = device_type
|
|
self.date_updated = datetime.now().strftime('%d-%m-%YT%H:%M:%S')
|
|
|
|
def set_device_serial_number(self, device_serial_number: str) -> None:
|
|
self.device_serial_number = device_serial_number
|
|
self.date_updated = datetime.now().strftime('%d-%m-%YT%H:%M:%S')
|
|
|
|
def set_device_firmware_version(self, device_firmware_version: str) -> None:
|
|
self.device_firmware_version = device_firmware_version
|
|
self.date_updated = datetime.now().strftime('%d-%m-%YT%H:%M:%S')
|
|
|
|
def set_device_name(self, device_name: str) -> None:
|
|
self.device_name = device_name
|
|
self.device_short_name = device_name.lower().replace(" ", "_")
|
|
self.date_updated = datetime.now().strftime('%d-%m-%YT%H:%M:%S')
|
|
|
|
@classmethod
|
|
def load_from_json(cls, device_file_path: 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.model_validate_json(metadata_json)
|
|
return metadata_model_obj
|
|
|
|
def save_to_json(self, file_path: Path):
|
|
if file_path.is_file():
|
|
print(f"File {file_path} already exists, update instead.")
|
|
return ReturnCodes.FILE_ALREADY_EXISTS
|
|
metadata = self.model_dump_json(indent=2)
|
|
with file_path.open('w') as file:
|
|
json.dump(metadata, file)
|
|
return ReturnCodes.SUCCESS
|
|
|
|
@classmethod
|
|
def update_metadata_in_json(cls, file_path: Path, **kwargs):
|
|
# TODO Maybe not needed at all.
|
|
assert file_path.is_file()
|
|
for field in IMMUTABLE_FIELDS:
|
|
if field in kwargs:
|
|
print(f"Field {field} is immutable")
|
|
return ReturnCodes.IMMUTABLE
|
|
metadata = cls.load_from_json(file_path)
|
|
for field, value in kwargs.items():
|
|
if field in metadata.model_fields_set:
|
|
setattr(metadata, field, value)
|
|
metadata.date_updated = datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower()
|
|
pass
|
|
|
|
|
|
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
|