67 lines
2.3 KiB
Python
67 lines
2.3 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
|
|
|
|
# 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())
|
|
|
|
# Optional Fields
|
|
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, /, **data: Any):
|
|
super().__init__(**data)
|
|
self.device_name = device_name
|
|
self.device_short_name = device_name.lower().replace(" ", "_")
|
|
|
|
@classmethod
|
|
def load_from_json(cls, file_path: Path):
|
|
assert file_path.is_file()
|
|
with file_path.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
|