Source code for openfactory.assets.utils
""" OpenFactory Assets helper functions. """
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Union, Literal
[docs]
def openfactory_timestamp(timestamp: datetime) -> str:
"""
Convert a datetime to OpenFactory timestamp format.
The format is ISO 8601 with milliseconds precision and a 'Z' to indicate UTC time,
e.g., '2025-05-04T12:34:56.789Z'.
Args:
timestamp (datetime.datetime): A datetime object (UTC recommended).
Returns:
str: Timestamp formatted in OpenFactory style.
"""
return timestamp.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
[docs]
def current_timestamp() -> str:
"""
Returns the current timestamp in OpenFactory format.
The format is ISO 8601 with milliseconds precision and a 'Z' to indicate UTC time,
e.g., '2025-05-04T12:34:56.789Z'.
Returns:
str: The current UTC timestamp formatted in OpenFactory style.
"""
return openfactory_timestamp(datetime.now(timezone.utc))
[docs]
@dataclass
class AssetAttribute:
"""
Represents a single attribute of an asset, including its value, type, tag, and timestamp.
Attributes:
value (Union[str, float]): The actual value of the attribute. Can be a string or float.
type (Literal['Samples', 'Condition', 'Events', 'Method', 'OpenFactory', 'UNAVAILABLE']):
The category/type of the attribute, must be one of the allowed literal strings.
tag (str): The tag or identifier associated with this attribute.
timestamp (str): Timestamp when the attribute was recorded, in OpenFactory format.
Defaults to the current timestamp if not provided.
"""
value: Union[str, float]
type: Literal['Samples', 'Condition', 'Events', 'Method', 'OpenFactory', 'UNAVAILABLE']
tag: str
timestamp: str = field(default_factory=current_timestamp)
def __post_init__(self) -> None:
"""
Validates the type of the attribute after initialization.
Raises:
ValueError: If the type is not one of the allowed literal strings.
"""
ALLOWED_TYPES = {'Samples', 'Condition', 'Events', 'Method', 'OpenFactory', 'UNAVAILABLE'}
if self.type not in ALLOWED_TYPES:
raise ValueError(f"Invalid type '{self.type}'. Allowed values are: {', '.join(ALLOWED_TYPES)}")