# -*- coding: UTF-8 -*-
"""
Author: Jacek 'Szumak' Kotlarski --<szumak@virthost.pl>
Created: 29.10.2023
Purpose: DataProcessor class for processing dataset operations.
"""
from inspect import currentframe
from typing import List, Tuple, Optional, Union, Any
from abc import ABC, abstractmethod
from copy import copy
from ...attribtool import NoDynamicAttributes, ReadOnlyClass
from ...raisetool import Raise
from ...basetool import BData
class _Keys(object, metaclass=ReadOnlyClass):
"""Keys definition class.
For internal purpose only.
"""
DATA: str = "__data__"
DESC: str = "__desc__"
DESCRIPTION: str = "__description__"
MAIN: str = "__main__"
NAME: str = "__name__"
VALUE: str = "__value__"
VARIABLES: str = "__variables__"
[docs]
class IModel(ABC):
"""Model class interface."""
@property
@abstractmethod
def dump(self) -> Union[List[str], "VariableModel"]:
"""Dump data.
### Returns:
Union[List[str], VariableModel] - Dumped data as list of strings or model object.
"""
@property
@abstractmethod
def name(self) -> Optional[str]:
"""Get name property.
### Returns:
Optional[str] - The name property value, or None if not set.
"""
@name.setter
@abstractmethod
def name(self, name: str) -> None:
"""Set name property.
### Arguments:
* name: str - The name to set.
"""
[docs]
@abstractmethod
def parser(self, value: str) -> None:
"""Parser method."""
[docs]
@abstractmethod
def search(self, name: str) -> bool:
"""Search method."""
[docs]
class VariableModel(BData, IModel, NoDynamicAttributes):
"""Representation of a configuration variable entry.
### Purpose:
Stores name, value, and description for a single configuration variable.
"""
[docs]
def __init__(
self,
name: Optional[str] = None,
value: Optional[Union[str, int, float, bool, List]] = None,
desc: Optional[str] = None,
) -> None:
"""Initialise a variable record.
### Arguments:
* name: Optional[str] - Variable identifier.
* value: Optional[Union[str, int, float, bool, List]] - Payload to store.
* desc: Optional[str] - Optional human-friendly description.
"""
self._set_data(key=_Keys.NAME, value=None, set_default_type=Optional[str])
self._set_data(
key=_Keys.VALUE,
value=None,
set_default_type=Optional[Union[str, int, float, bool, List]],
)
self._set_data(key=_Keys.DESC, value=None, set_default_type=Optional[str])
self.name = name
self.value = value
self.desc = desc
[docs]
def __repr__(self) -> str:
"""Return representation class string.
### Returns:
[str] - Debug-friendly section representation.
"""
tmp: str = ""
tmp += f"name='{self.name}', " if self.name is not None else ""
if isinstance(self.value, (int, float, bool)):
tmp += f"value={self.value}, " if self.value is not None else ""
elif isinstance(self.value, List):
tmp += f"value=[{self.value}], " if self.value is not None else ""
else:
tmp += f"value='{self.value}', " if self.value is not None else ""
tmp += f"desc='{self.desc}'" if self.desc is not None else ""
return f"{self._c_name}({tmp})"
[docs]
def __str__(self) -> str:
"""Return formatted section header.
### Returns:
[str] - String in INI-style header format.
"""
tmp: str = ""
tmp += f"{self.name} = " if self.name is not None else ""
if isinstance(self.value, (int, float, bool)):
tmp += f"{self.value}" if self.value is not None else ""
elif isinstance(self.value, List):
tmp += f"{self.value}" if self.value is not None else ""
else:
tmp += (
'"{}"'.format(str(self.value).strip().strip("\"'"))
if self.value is not None
else ""
)
if tmp:
tmp += f" # {self.desc}" if self.desc is not None else ""
else:
tmp += f"# {self.desc}" if self.desc is not None else "#"
return tmp
@property
def desc(self) -> Optional[str]:
"""Get description property.
### Returns:
[Optional[str]] - Description string or None.
"""
return self._get_data(key=_Keys.DESC)
@desc.setter
def desc(self, desc: Optional[str]) -> None:
"""Set description property.
### Arguments:
* desc: Optional[str] - Description text.
"""
self._set_data(key=_Keys.DESC, value=desc, set_default_type=Optional[str])
@property
def dump(self) -> "VariableModel":
"""Dump data.
### Returns:
[List[Any]] - Shallow copy of section with variables.
"""
return self # type: ignore
@property
def name(self) -> Optional[str]:
"""Get name property.
### Returns:
[Optional[str]] - Section name.
"""
return self._get_data(key=_Keys.NAME)
@name.setter
def name(self, name: Optional[str]) -> None:
"""Set name property.
### Arguments:
* name: Optional[str] - New variable name.
### Raises:
* ValueError: Raised when the trimmed name becomes empty.
"""
if name is None:
self._set_data(key=_Keys.NAME, value=None)
else:
if not isinstance(name, str):
raise Raise.error(
f"Expected 'str' type, received: '{type(name)}'",
TypeError,
self._c_name,
currentframe(),
)
cleaned = name.strip()
if not cleaned:
raise Raise.error(
"Variable name cannot be empty.",
ValueError,
self._c_name,
currentframe(),
)
self._set_data(key=_Keys.NAME, value=cleaned)
[docs]
def parser(self, value: str) -> None:
"""Parse raw string input ensuring valid state.
### Arguments:
* value: str - Raw value string.
### Raises:
* ValueError: Raised when the parsed value is empty after trimming.
"""
if not isinstance(value, str):
raise Raise.error(
f"Expected 'str' type, received: '{type(value)}'",
TypeError,
self._c_name,
currentframe(),
)
if not value.strip():
raise Raise.error(
"Variable value cannot be empty.",
ValueError,
self._c_name,
currentframe(),
)
self.value = value
[docs]
def search(self, name: str) -> bool:
"""Search method.
### Arguments:
* name: str - Section name to match.
### Returns:
[bool] - True when names coincide.
"""
return self.name == name
@property
def value(self) -> Optional[Union[str, int, float, bool, List]]:
"""Get value property.
### Returns:
[Optional[Union[str, int, float, bool, List]]] - Stored value.
"""
return self._get_data(key=_Keys.VALUE)
@value.setter
def value(self, value: Optional[Union[str, int, float, bool, List]]) -> None:
"""Set value property.
### Arguments:
* value: Optional[Union[str, int, float, bool, List]] - Payload.
"""
self._set_data(key=_Keys.VALUE, value=value)
[docs]
class SectionModel(BData, IModel, NoDynamicAttributes):
"""Representation of a configuration section.
### Purpose:
Aggregates variables and descriptions under a named section.
"""
[docs]
def __init__(self, name: Optional[str] = None) -> None:
"""Initialise section state.
### Arguments:
* name: Optional[str] - Optional section header value.
"""
self._set_data(key=_Keys.NAME, value=None, set_default_type=Optional[str])
self._set_data(
key=_Keys.VARIABLES,
value=[],
set_default_type=List[VariableModel],
)
self.name = name
[docs]
def __repr__(self) -> str:
"""Return representation class string.
### Returns:
[str] - Debug-friendly section representation.
"""
return f"{self._c_name}(name='{self.name}')"
[docs]
def __str__(self) -> str:
"""Return formatted section header.
### Returns:
[str] - String in INI-style header format.
"""
return f"[{self.name}]"
@property
def dump(self) -> List[Any]:
"""Dump data.
### Returns:
[List[Any]] - Shallow copy of section with variables.
"""
tmp: List = []
tmp.append(self)
for item in self.variables:
tmp.append(item.dump)
return copy(tmp)
[docs]
def parser(self, value: Optional[str]) -> None:
"""Parse and validate section name.
### Arguments:
* value: Optional[str] - Raw header text.
### Raises:
* ValueError: Raised when the name resolves to an empty string.
"""
if value is None:
return
if not isinstance(value, str):
raise Raise.error(
f"Expected 'str' type, received: '{type(value)}'",
TypeError,
self._c_name,
currentframe(),
)
tmp: str = f"{value}".strip("[] \n")
if tmp:
self._set_data(key=_Keys.NAME, value=tmp)
else:
raise Raise.error(
f"Expected String name, received: '{tmp}'.",
ValueError,
self._c_name,
currentframe(),
)
[docs]
def search(self, name: str) -> bool:
"""Search method.
### Arguments:
* name: str - Section name to match.
### Returns:
[bool] - True when names coincide.
"""
return self.name == name
@property
def name(self) -> Optional[str]:
"""Get name property.
### Returns:
[Optional[str]] - Section name.
"""
return self._get_data(key=_Keys.NAME)
@name.setter
def name(self, name: Optional[str]) -> None:
"""Set name property.
### Arguments:
* name: Optional[str] - New section name.
"""
if name is None:
self._set_data(key=_Keys.NAME, value=None)
return
self.parser(name)
[docs]
def get_variable(self, name: str) -> Optional[VariableModel]:
"""Search and return VariableModel if exists.
### Arguments:
* name: str - Variable name to locate.
### Returns:
[Optional[VariableModel]] - Matching variable instance or None.
"""
name = str(name).strip()
for item in self.variables:
if item.name == name:
return item
return None
[docs]
def set_variable(
self,
name: Optional[str] = None,
value: Optional[Any] = None,
desc: Optional[str] = None,
) -> None:
"""Add or update VariableModel.
### Arguments:
* name: Optional[str] - Variable name to set or update.
* value: Optional[Any] - Value payload.
* desc: Optional[str] - Description text.
### Raises:
* ValueError: Raised when both name and value are None.
"""
if name is not None:
key = name.strip() if isinstance(name, str) else name
tmp: Optional[VariableModel] = (
self.get_variable(key) if key is not None else None
)
if tmp is not None:
item: VariableModel = tmp
if value is not None or (value is None and desc is None):
item.value = value
if desc is not None or (desc is None and value is None):
item.desc = desc
return
# add new VariableModel
if name is None and value is not None:
raise Raise.error(
"Variable name is required when providing a value.",
ValueError,
self._c_name,
currentframe(),
)
normalized_name: Optional[str] = name.strip() if isinstance(name, str) else name
items: List[VariableModel] = copy(self.variables)
items.append(VariableModel(normalized_name, value, desc))
self._set_data(key=_Keys.VARIABLES, value=items)
@property
def variables(self) -> List[VariableModel]:
"""Return list of VariableModel.
### Returns:
[List[VariableModel]] - Mutable list of section variables.
"""
items: Optional[List[VariableModel]] = self._get_data(key=_Keys.VARIABLES)
if items is None:
self._set_data(key=_Keys.VARIABLES, value=[])
items = self._get_data(key=_Keys.VARIABLES)
return items # type: ignore[return-value]
[docs]
class DataProcessor(BData, NoDynamicAttributes):
"""Manage configuration data composed of sections and variables.
### Purpose:
Provides helpers for reading, setting, and dumping configuration structures.
"""
[docs]
def __init__(self) -> None:
"""Initialise DataProcessor state.
### Purpose:
Ensures the internal container for sections exists.
"""
self._set_data(
key=_Keys.DATA,
value=[],
set_default_type=List[SectionModel],
)
self._set_data(key=_Keys.MAIN, value=None, set_default_type=Optional[str])
@property
def main_section(self) -> Optional[str]:
"""Return main section name.
### Returns:
[Optional[str]] - Name of the primary section.
"""
return self._get_data(key=_Keys.MAIN)
@main_section.setter
def main_section(self, name: str) -> None:
"""Set main section name.
### Arguments:
* name: str - Target section name.
"""
if not isinstance(name, str):
name = str(name)
self._set_data(key=_Keys.MAIN, value=name)
self.add_section(name)
@property
def sections(self) -> Tuple:
"""Return sections keys tuple.
### Returns:
[Tuple] - Tuple containing names of tracked sections.
"""
out = []
items: Optional[List[SectionModel]] = self._get_data(key=_Keys.DATA)
for item in items or []:
out.append(item.name)
return tuple(sorted(out))
[docs]
def add_section(self, name: str) -> str:
"""Add section object to dataset.
### Arguments:
* name: str - Raw section name.
### Returns:
[str] - Sanitised section name actually stored.
"""
sm = SectionModel(str(name))
if sm.name not in self.sections:
items: List[SectionModel] = copy(self._get_data(key=_Keys.DATA) or [])
items.append(sm)
self._set_data(key=_Keys.DATA, value=items)
return f"{sm.name}"
[docs]
def get_section(self, name: str) -> Optional[SectionModel]:
"""Get section object if exists.
### Arguments:
* name: str - Section name.
### Returns:
[Optional[SectionModel]] - Matching section or None.
"""
sm = SectionModel(name)
items: Optional[List[SectionModel]] = self._get_data(key=_Keys.DATA)
for item in items or []:
if item.name == sm.name:
return item
return None
[docs]
def set(
self,
section: str,
varname: Optional[str] = None,
value: Optional[Any] = None,
desc: Optional[str] = None,
) -> None:
"""Set data to [SectionModel]->[VariableModel].
### Arguments:
* section: str - Section name.
* varname: Optional[str] - Variable name.
* value: Optional[Any] - Value payload.
* desc: Optional[str] - Description text.
"""
section_name: str = self.add_section(section)
tmp: Optional[SectionModel] = self.get_section(section_name)
if tmp is not None:
found_section: SectionModel = tmp
found_section.set_variable(varname, value, desc)
[docs]
def get(
self, section: str, varname: Optional[str] = None, desc: bool = False
) -> Optional[Any]:
"""Return value.
### Arguments:
* section: str - Section name.
* varname: Optional[str] - Variable name to fetch.
* desc: bool - When True returns description.
### Returns:
[Optional[Any]] - Value or description based on `desc`.
### Raises:
* KeyError: Section not present.
"""
sm = SectionModel(section)
if sm.name in self.sections:
tmp: Optional[SectionModel] = self.get_section(section)
if tmp is not None:
found_section: SectionModel = tmp
if varname is not None:
found_var: Optional[VariableModel] = found_section.get_variable(
varname
)
if found_var is not None:
if desc:
# Return description for varname
return found_var.desc
else:
# Return value for varname
return found_var.value
else:
return None
else:
# Return list od description for section
out: List[str] = []
for item in found_section.variables:
if item.name is None and item.desc is not None:
out.append(item.desc)
return out
else:
raise Raise.error(
f"Given section name: '{section}' not found.",
KeyError,
self._c_name,
currentframe(),
)
def __dump(self, section: str) -> str:
"""Return formatted configuration data for section name.
### Arguments:
* section: str - Section name to dump.
### Returns:
[str] - Section content formatted for output.
### Raises:
* KeyError: Section not present.
"""
out: str = ""
if section in self.sections:
tmp: Optional[SectionModel] = self.get_section(section)
if tmp is not None:
found_section: SectionModel = tmp
out += f"{found_section}\n"
for item in found_section.variables:
out += f"{item}\n"
out += f"# -----<end of section: '{found_section.name}'>-----\n"
else:
raise Raise.error(
f"Section name: '{section}' not found.",
KeyError,
self._c_name,
currentframe(),
)
return out
@property
def dump(self) -> str:
"""Return formatted configuration data string.
### Returns:
[str] - Full configuration payload.
### Raises:
* KeyError: Main section not set.
"""
out: str = ""
# first section is a main section
if self.main_section is None:
raise Raise.error(
"Main section is not set.",
KeyError,
self._c_name,
currentframe(),
)
out = self.__dump(self.main_section)
# other sections
for section in sorted(tuple(set(self.sections) ^ set([self.main_section]))):
out += self.__dump(section)
return out
# #[EOF]#######################################################################