a ijd @sXdZddlmZddlZddlZddlZddlmZmZmZddl m Z ddl m Z ddl m Z ddl mZdd l mZeZed Zed Zejd#d d d d d d d dddddddddddZejd d d d d d d d ddddddd dddZd$ddddddd ddZd d dddZd d dddZd dd ddd Zd d dd!d"Ze jZdS)%zDataclass utils.) annotationsN)AnyCallableTypeVar)epy) cast_utils)context) frozen_utils)helpers_ClsT_T.)kw_onlyreplacerepr auto_cast contextvarsallow_unfrozenNoneboolzCallable[[_ClsT], _ClsT])clsr rrrrrreturncCsdSNrr rrrrrrrW/home/ghrups/robot-bench/.venv/lib/python3.9/site-packages/etils/edc/dataclass_utils.py dataclass#s rcCsdSrrrrrrr1s FTcCs|durtjt|||||dS|r*t|}|r6t|}|rBt|}|rPt|}g}|rp|t j t j t j d|r|t j tjtjdt ||}|S)aAugment a dataclass with additional features. `auto_cast`: Auto-convert init assignements to the annotated class. ```python @edc.dataclass class A: path: edc.AutoCast[epath.Path] some_enum: edc.AutoCast[MyEnum] x: edc.AutoCast[str] a = A( path='/some/path', some_enum='A', x=123 ) # Fields annotated with `AutoCast` are automatically casted to their type assert a.path == epath.Path('/some/path') assert a.some_enum is MyEnum.A assert a.x == '123' ``` `allow_unfrozen`: allow nested dataclass to be updated. This add two methods: * `.unfrozen()`: Create a lazy deep-copy of the current dataclass. Updates to nested attributes will be propagated to the top-level dataclass. * `.frozen()`: Returns the frozen dataclass, after it was mutated. Example: ```python old_x = X(y=Y(z=123)) x = old_x.unfrozen() x.y.z = 456 x = x.frozen() assert x == X(y=Y(z=123)) # Only new x is mutated assert old_x == X(y=Y(z=456)) # Old x is not mutated ``` Note: * Only the last `.frozen()` call resolve the dataclass by calling `.replace` recursivelly. * Dataclass returned by `.unfrozen()` and nested attributes are not the original dataclass but proxy objects which track the mutations. As such, those object are not compatible with `isinstance()`, `jax.tree_map`,... * Only the top-level dataclass need to be `allow_unfrozen=True` * Avoid using `unfrozen` if 2 attributes of the dataclass point to the same nested dataclass. Updates on one attribute might not be reflected on the other. ```python y = Y(y=123) x = X(x0=y, x1=y) # Same instance assigned twice in `x0` and `x1` x = x.unfrozen() x.x0.y = 456 # Changes in `x0` not reflected in `x1` x = x.frozen() assert x == X(x0=Y(y=456), x1=Y(y=123)) ``` This is because only attributes which are accessed are tracked, so `etils` do not know the object exist somewhere else in the attribute tree. * After `.frozen()` has been called, any of the temporary sub-attribute become invalid: ```python a = a.unfrozen() y = a.y a = a.frozen() y.x # Raise error (created between the unfrozen/frozen call) a.y.x # Work ``` `contextvars`: Fields annotated as `edc.ContextVar` are wrapped in a `contextvars.ContextVar`. Afterward each thread / asyncio coroutine will have its own version of the fields (similarly to `threading.local`). The contextvars are lazily initialized at first usage. Example: ```python @edc.dataclass @dataclasses.dataclass class Context: thread_id: edc.ContextVar[int] = dataclasses.field( default_factory=threading.get_native_id ) stack: edc.ContextVar[list[str]] = dataclasses.field(default_factory=list) # Global context object context = Context(thread_id=0) def worker(): # Inside each thread, the worker use its own context assert context.thread_id != 0 context.stack.append(1) with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: for _ in range(10): executor.submit(worker) ``` Args: cls: The dataclass to decorate kw_only: If True, make the dataclass `__init__` keyword-only. replace: If `True`, add a `.replace(` alias of `dataclasses.replace`. repr: If `True`, the class `__repr__` will return a pretty-printed `str` (one attribute per line) auto_cast: If `True`, fields annotated as `x: edc.AutoCast[Cls]` will be converted to `x: Cls = edc.field(validator=Cls)`. contextvars: It `True`, fields annotated as `x: edc.AutoCast[T]` are converted to `contextvars`. This allow to have a `threading.local`-like API for contextvars. allow_unfrozen: If `True`, add `.frozen`, `.unfrozen` methods. Returns: Decorated class N)r rrrr) annotationZ descriptor_fn) functoolspartialr _make_kw_onlyadd_repr _add_replacer Z add_unfrozenappendr ZDescriptorInforZAutoCastZmake_auto_cast_descriptorrZ ContextVarZmake_contextvar_descriptorZwrap_new)rr rrrrrZdescriptor_fnsrrrr?sF   )rrcs4d|jvr|S|jtfdd}||_|S)z1Replace the `__init__` by a keyword-only version.__init__cs0|r t|jjdt|d|fi|S)Nz! contructor is keyword-only. Got z positional arguments.) TypeError __class____name__len)selfargskwargsZold_initrrr#s z_make_kw_only..__init__)__dict__r#rwraps)rr#rr+rrs rcCst|dst|_|S)z=Add a `.replace` method to the class, if not already present.r)hasattrrrrrrr! s r!r)r(r*rcKstj|fi|S)z!Similar to `dataclasses.replace`.) dataclassesr)r(r*rrrrsrcCs$d|jvr|Stj|r t|_|S)z>Add a `.__repr__` method to the class, if not already present.__repr__)r,rZ text_utilsZhas_default_reprr1r/rrrr s   r ).)N)__doc__ __future__rr0rtypingrrrZetilsrZ etils.edcrrr r Z_Clsr r overloadrrr!rr pretty_reprr1rrrrs\      " " 4