unitx_task_unit

demo.py

# %%
# Approach #1 - Use Proxy Pattern to control, the basic idea is to restrict and record 
# setter on python class

class RestrictedProtoMsg:
    """
    This class is a wrapper around a protobuf message, restricting the fields that can be modified.
    It also keeps track of whether certain fields have been updated.
    """
    def __init__(self, proto_msg, writable, should_update):
        """
        Initialize the RestrictedProtoMsg with the protobuf message, the fields that can be written to,
        and the fields that should be updated.
        """
        writable = set(writable + should_update)
        object.__setattr__(self, '_proto_msg', proto_msg)
        object.__setattr__(self, '_writable', writable)
        object.__setattr__(self, '_should_update', should_update)
        # Add an update counter
        from collections import defaultdict
        counter = defaultdict(lambda: 0)
        for f in should_update: counter[f] = 1
        object.__setattr__(self, '_should_update_counter', counter)

    def __setattr__(self, name, value):
        """
        Set the attribute of the protobuf message if it is writable.
        If the attribute is in the should_update list, decrement the counter.
        """
        if name in self.__dict__:  # Directly set the instance attribute
            object.__setattr__(self, name, value)
        elif name in self._writable:
            setattr(self._proto_msg, name, value)
            # When a field is being updated, decrement the counter
            if name in self._should_update_counter:
                self._should_update_counter[name] -= 1
        else:
            raise AttributeError(f"Attribute {name} is read-only")

    def __getattr__(self, name):
        """
        Get the attribute of the protobuf message.
        """
        if name in self.__dict__:
            return self.__dict__[name]
        else:
            return getattr(self._proto_msg, name)

    def check_updates(self):
        """
        Check if all fields in the should_update list have been updated.
        """
        for count in self._should_update_counter.values():
            if count > 0:
                return False
        return True

    def __repr__(self) -> str:
        """
        Return a string representation of the RestrictedProtoMsg.
        """
        return "RestrictedProtoMsg: " + repr(self._proto_msg)


def restricted_msg(proto_msg, writable=[], should_update=[]):
    """
    writable field: fields which allows modification
    should_udpate: keep track of change and 
    """
    return RestrictedProtoMsg(proto_msg, writable, should_update)


# Demo
class TaskContext: pass
class ImageMessage:
    def __setattr__(self, key, value):
        # Use the built-in object's __setattr__ for private attributes
        if key.startswith('_'):
            super().__setattr__(key, value)
        else:
            # Otherwise, store the attribute in the dictionary
            self.__dict__[key] = value

    def __getattr__(self, key):
        try:
            # If the attribute exists in the dictionary, return it
            return self.__dict__[key]
        except KeyError:
            raise AttributeError(f"'{type(self).__name__}' object has no attribute '{key}'")

img_msg = ImageMessage()
img_msg.foo = 'bar'
img_msg.year = 2044
img_msg.need_change = None

def process_msg(
    image_message: ImageMessage, context: TaskContext
) -> ImageMessage:

    image_message = restricted_msg(
        image_message, 
        writable=["foo"],
        should_update=['need_change'])

    image_message.foo = 'this is ok'
    try:
        image_message.year = 2023 # this raise an error
    except AttributeError as e:
        print("Error: ", e)

    # This could be abstracted into a decorator
    print("Has update happened: ", image_message.check_updates())
    image_message.need_change = 'something magical happened'
    print("Has update happened: ", image_message.check_updates())

    return image_message

# Approach #2 - Using type hints, this requires each task unit to be defined with separate types
# %% Demo of using python typing system at runtime against data
from typing import Optional, Union, Any, get_origin, get_args
from typing import List, Any, get_origin, get_args, Sequence

def is_instance_of_type(value, type_hint):
    origin = get_origin(type_hint)
    args = get_args(type_hint)
    
    # For generic types
    if origin:
        # handles, Union and Optional, Optional is Union(Any, None)
        if origin is Union:
            NoneType = type(None)
            if NoneType in args: 
                return value is None or any(is_instance_of_type(value, arg) for arg in args if arg is not NoneType)
            return any(is_instance_of_type(value, arg) for arg in args)
        # For other generic types
        elif isinstance(value, origin):
            # For parameterized generics, we need to check inner types too
            if args:
                # Example: for List[int], check inner int type
                if origin in [list, set]:
                    return all(is_instance_of_type(item, args[0]) for item in value)
                # For Dict[K, V], check key and value types
                elif origin is dict:
                    return all(is_instance_of_type(k, args[0]) and is_instance_of_type(v, args[1]) for k, v in value.items())
            return True
        return False
    # For simple types
    else:
        if type_hint is Any: # Support for Any
            return True
        return isinstance(value, type_hint)

# Testing the function
print(is_instance_of_type([1, 2, 3], list[int]))  # True
print(is_instance_of_type(5, Optional[int]))  # True
print(is_instance_of_type(None, Optional[int]))  # True
print(is_instance_of_type("hello", Union[int, str]))  # True
print(is_instance_of_type(10, Union[int, str]))  # True
print(is_instance_of_type([10, "hi"], list[Union[int, str]]))  # True
print(is_instance_of_type([10, "hi", 5.0], list[Union[int, str]]))  # False, due to 5.0 not being int or str
print(is_instance_of_type({1, 2, 3}, set[int]))  # True
print(is_instance_of_type({1, 2, 3}, set[str]))  # False, due to set containing ints
print(is_instance_of_type({1: "one", 2: "two"}, dict[int, str]))  # True
print(is_instance_of_type({1: "one", "2": "two"}, dict[int, str]))  # False, due to key "2" not being int
print(is_instance_of_type(5, Any))  # True
print(is_instance_of_type("hello", Any))  # True
print(is_instance_of_type(None, Any))  # True
print(is_instance_of_type([1, 2, 3], Any))  # True
print(is_instance_of_type({1, 2, 3}, Any))  # True
print(is_instance_of_type({1: "one", 2: "two"}, Any))  # True
print(is_instance_of_type(5.0, Any))  # True
print(is_instance_of_type("hi", Any))  # True
print(is_instance_of_type([10, "hi", 5.0], Any))  # True

# %% Demo of using python typing computations
from typing import List, Any, get_origin, get_args, Sequence
from collections.abc import Sequence as SequenceABC


def is_superset(type1, type2):
    origin1 = get_origin(type1) or type1  # If type1 is not parameterized, use type1 as origin
    origin2 = get_origin(type2) or type2  # If type2 is not parameterized, use type2 as origin

    args1 = get_args(type1)
    args2 = get_args(type2)

    # Check if List is a subtype of Sequence
    if origin1 in [Sequence, SequenceABC] and origin2 is list:
        if not args1 or args1[0] is Any:  # If args1 is empty, then type1 is just Sequence without type hints.
            return True
        return issubclass(args2[0], args1[0]) or args1[0] == args2[0]

    # Check if the origins are the same
    if origin1 is not origin2:
        return False

    if origin1 in [list, set]:
        return args1[0] is Any or args1[0] == args2[0]

    return False

# Testing with explanations
print(is_superset(List[Any], List[str]))       # True, because Any is a superset of all types, including str
print(is_superset(Sequence[Any], List[str]))   # True, because Any is a superset of all types, including str
print(is_superset(Sequence[str], List[str]))   # True, because str is a superset of itself print(is_superset(List[str], List[Any]))       # False, because str is not a superset of Any
print(is_superset(List[int], List[str]))       # False, because int is not a superset of str
print(is_superset(Sequence[int], List[str]))   # False, because int is not a superset of str