pygx.hyper

Hyper objects: representing template-based object space.

hyper

Hyper objects: representing template-based object space.

In PyGX, an object space is represented by a hyper object, which is a symbolic object with parts placeheld by hyper primitives (pg.hyper.HyperPrimitive). Through hyper objects, object templates (pg.hyper.ObjectTemplate) can be obtained to generate objects based on program genomes (pg.DNA).

classDiagram
  HyperValue <|-- ObjectTemplate
  HyperValue <|-- HyperPrimitive
  HyperPrimitive <|-- Choices
  Choices <|-- OneOf
  Choices <|-- ManyOf
  HyperPrimitive <|-- Float
  HyperPrimitive <|-- CustomHyper
  CustomHyper <|-- Evolvable

Hyper values map 1:1 to genotypes as the following:

Hyper class	Genotype class
pg.hyper.HyperValue	pg.DNASpec
pg.hyper.ObjectTemplate	pg.geno.Space
pg.hyper.HyperPrimitive	pg.geno.DecisionPoint
pg.hyper.Choices	pg.geno.Choices
pg.hyper.Float	pg.geno.Float
pg.hyper.CustomHyper	pg.geno.CustomDecisionPoint

HyperPrimitive

HyperPrimitive(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: Object, HyperValue

Base class for hyper primitives.

A hyper primitive is a pure symbolic object which represents an object generation rule. It corresponds to a decision point (pygx.geno.DecisionPoint) in the algorithm's view.

Child classes:

• pygx.hyper.Choices

  • pygx.hyper.OneOf
  • pygx.hyper.ManyOf
  • pygx.hyper.Float
  • pygx.hyper.CustomHyper

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

HyperValue

HyperValue()

Bases: NonDeterministic

Base class for a hyper value.

Hyper value represents a space of objects, which is essential for programmatically generating objects. It can encode a concrete object into a DNA, or decode a DNA into a concrete object.

DNA is a nestable numeric interface we use to generate objects (see pg.geno). Each position in the DNA represents either the index of a choice, or a value itself is numeric. There could be multiple choices standing side-by-side, representing knobs on different parts of an object, or choices being chained, forming conditional choice spaces, which can be described by a tree structure.

Hyper values form a tree as the following:

classDiagram
  ObjectTemplate --> HyperPrimitive : elements (1..*)
  HyperPrimitive <|-- OneOf_ManyOf
  HyperPrimitive <|-- Float
  HyperPrimitive <|-- CustomHyper
  OneOf_ManyOf --> ObjectTemplate : candidates (1..*)

Source code in pygx/hyper/_base.py

def __init__(self):
    # DNA and decoded value are states for __call__.
    # Though `decode` and `encode` methods are stateless.
    self._dna = None
    self._decoded_value = None

dna property

dna: DNA | None

Returns the DNA that is being used by this hyper value.

set_dna

set_dna(dna: DNA) -> None

Use this DNA to generate value.

NOTE(daiyip): self.dna is only used in __call_. Thus 'set_dna' can be called multiple times to generate different values.

Parameters:

Name	Type	Description	Default
dna	DNA	DNA to use to decode the value.	required

Source code in pygx/hyper/_base.py

def set_dna(self, dna: geno.DNA) -> None:
    """Use this DNA to generate value.

    NOTE(daiyip): self._dna is only used in __call__.
    Thus 'set_dna' can be called multiple times to generate different values.

    Args:
      dna: DNA to use to decode the value.
    """
    self._dna = dna
    # Invalidate decoded value when DNA is refreshed.
    self._decoded_value = None

decode

decode(dna: DNA) -> Any

Decode a value from a DNA.

Source code in pygx/hyper/_base.py

def decode(self, dna: geno.DNA) -> Any:
    """Decode a value from a DNA."""
    self.set_dna(dna)
    return _detach(self._decode())

encode abstractmethod

encode(value: Any) -> DNA

Encode a value into a DNA.

Parameters:

Name	Type	Description	Default
value	Any	A value that conforms to the hyper value definition.	required

Returns:

Type	Description
DNA	DNA for the value.

Source code in pygx/hyper/_base.py

@abc.abstractmethod
def encode(self, value: Any) -> geno.DNA:
    """Encode a value into a DNA.

    Args:
      value: A value that conforms to the hyper value definition.

    Returns:
      DNA for the value.
    """

dna_spec abstractmethod

dna_spec(location: KeyPath | None = None) -> DNASpec

Get DNA spec of DNA that is decodable/encodable by this hyper value.

Source code in pygx/hyper/_base.py

@abc.abstractmethod
def dna_spec(
    self, location: topology.KeyPath | None = None
) -> geno.DNASpec:
    """Get DNA spec of DNA that is decodable/encodable by this hyper value."""

Choices

Choices(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: HyperPrimitive

Categorical choices from a list of candidates.

Example:

# A single categorical choice:
v = pg.oneof([1, 2, 3])

# A multiple categorical choice as a list:
vs = pg.manyof(2, [1, 2, 3])

# A hierarchical categorical choice:
v2 = pg.oneof([
    'foo',
    'bar',
    pg.manyof(2, [1, 2, 3])
])

See also:

• pygx.hyper.OneOf
• pygx.hyper.ManyOf
• pygx.oneof
• pygx.manyof
• pygx.permutate

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

candidate_templates property

candidate_templates

Returns candidate templates.

is_leaf property

is_leaf: bool

Returns whether this is a leaf node.

on_sym_bound

on_sym_bound()

On members are bound.

Source code in pygx/hyper/_categorical.py

def on_sym_bound(self):
    """On members are bound."""
    super().on_sym_bound()
    if (
        self.num_choices is not None
        and self.num_choices > len(self.candidates)
        and self.choices_distinct
    ):
        raise ValueError(
            f'{len(self.candidates)} candidates cannot produce '
            f'{self.num_choices} distinct choices.'
        )
    self._candidate_templates = [
        object_template.ObjectTemplate(c, where=self.where)
        for c in self.candidates
    ]
    self._constant_candidate_index = self._build_constant_candidate_index()
    # ValueSpec for candidate.
    self._value_spec = None

dna_spec

dna_spec(location: KeyPath | None = None) -> Choices

Returns corresponding DNASpec.

Source code in pygx/hyper/_categorical.py

def dna_spec(
    self, location: topology.KeyPath | None = None
) -> geno.Choices:
    """Returns corresponding DNASpec."""
    return geno.Choices(
        num_choices=self.num_choices,  # pyright: ignore[reportArgumentType]
        candidates=[ct.dna_spec() for ct in self._candidate_templates],
        literal_values=[
            self._literal_value(c) for i, c in enumerate(self.candidates)
        ],
        distinct=self.choices_distinct,
        sorted=self.choices_sorted,
        hints=self.hints,
        name=self.name,
        location=location or topology.KeyPath(),
    )

encode

encode(value: list[Any]) -> DNA

Encode a list of values into DNA.

Each value in the list is matched against the candidates from left to right; the DNA records the index of the first matching candidate, plus child decisions when the candidate contains nested hyper values.

Example:

# A multiple choice encodes to the list of chosen indices.
c = pg.manyof(2, ['a', 'b', 'c'])
assert c.encode(['a', 'c']) == pg.DNA([0, 2])

# A nested hyper value contributes child decisions.
c = pg.oneof(['foo', pg.Dict(x=pg.oneof([1, 2]))])
assert c.encode('foo') == pg.DNA(0)
assert c.encode(pg.Dict(x=2)) == pg.DNA((1, 1))

Parameters:

Name	Type	Description	Default
value	list[Any]	A list of values that can match choice candidates.	required

Returns:

Type	Description
DNA	Encoded DNA.

Raises:

Type	Description
ValueError	If value cannot be encoded.

Source code in pygx/hyper/_categorical.py

def encode(self, value: list[Any]) -> geno.DNA:
    """Encode a list of values into DNA.

    Each value in the list is matched against the candidates from left
    to right; the DNA records the index of the first matching candidate,
    plus child decisions when the candidate contains nested hyper
    values.

    **Example:**

    ```python
    # A multiple choice encodes to the list of chosen indices.
    c = pg.manyof(2, ['a', 'b', 'c'])
    assert c.encode(['a', 'c']) == pg.DNA([0, 2])

    # A nested hyper value contributes child decisions.
    c = pg.oneof(['foo', pg.Dict(x=pg.oneof([1, 2]))])
    assert c.encode('foo') == pg.DNA(0)
    assert c.encode(pg.Dict(x=2)) == pg.DNA((1, 1))
    ```

    Args:
      value: A list of values that can match choice candidates.

    Returns:
      Encoded DNA.

    Raises:
      ValueError: If `value` cannot be encoded.
    """
    if not isinstance(value, list):
        raise ValueError(
            topology.message_on_path(
                'Cannot encode value: value should be a list type. '
                f'Encountered: {value!r}.',
                self.topo_path,
            )
        )
    choices = []
    if self.num_choices is not None and len(value) != self.num_choices:
        raise ValueError(
            topology.message_on_path(
                'Length of input list is different from the number of choices '
                f'({self.num_choices}). Encountered: {value}.',
                self.topo_path,
            )
        )
    candidate_templates = self._candidate_templates
    constant_index = self._constant_candidate_index
    for v in value:
        # Fast path: every candidate is a primitive constant, so match by
        # dict lookup instead of probing each candidate via `try_encode`.
        if (
            constant_index is not None
            and v.__class__ in _HASHABLE_PRIMITIVE_TYPES
        ):
            idx = constant_index.get(v)
            if idx is not None:
                choices.append(geno.DNA(idx))
                continue
        choice_id = None
        child_dna = None
        for i, b in enumerate(candidate_templates):
            succeeded, child_dna = b.try_encode(v)
            if succeeded:
                choice_id = i
                break
        if child_dna is None:
            raise ValueError(
                topology.message_on_path(
                    'Cannot encode value: no candidates matches with '
                    f'the value. Value: {v!r}, Candidates: {self.candidates}.',
                    self.topo_path,
                )
            )
        # `DNA(choice_id, [DNA(None, [])])` collapses to `DNA(choice_id)`,
        # so skip the wrapper allocation when `child_dna` is value-less
        # and empty (typical for constant candidates).
        if child_dna.value is None and not child_dna.children:
            choices.append(geno.DNA(choice_id))
        else:
            choices.append(geno.DNA(choice_id, [child_dna]))
    return geno.DNA(None, choices)

ManyOf

ManyOf(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: Choices

N Choose K.

Example:

# Chooses 2 distinct candidates.
v = pg.manyof(2, [1, 2, 3])

# Chooses 2 non-distinct candidates.
v = pg.manyof(2, [1, 2, 3], distinct=False)

# Chooses 2 distinct candidates sorted by their indices.
v = pg.manyof(2, [1, 2, 3], sorted=True)

# Permutates the candidates.
v = pg.permutate([1, 2, 3])

# A hierarchical categorical choice:
v2 = pg.manyof(2, [
    'foo',
    'bar',
    pg.oneof([1, 2, 3])
])

See also:

• pygx.manyof
• pygx.permutate
• pygx.hyper.Choices
• pygx.hyper.OneOf
• pygx.hyper.Float
• pygx.hyper.CustomHyper

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

custom_apply

custom_apply(
    path: KeyPath,
    value_spec: ValueSpec,
    allow_partial: bool,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
) -> tuple[bool, Choices]

Validate candidates during value_spec binding time.

Source code in pygx/hyper/_categorical.py

def custom_apply(
    self,
    path: topology.KeyPath,
    value_spec: pg_typing.ValueSpec,
    allow_partial: bool,
    child_transform: None
    | (Callable[[topology.KeyPath, pg_typing.Field, Any], Any]) = None,
) -> tuple[bool, 'Choices']:
    """Validate candidates during value_spec binding time."""
    # Check if value_spec directly accepts `self`.
    if value_spec.value_type and isinstance(self, value_spec.value_type):
        return (False, self)

    if self._value_spec:
        src_spec = self._value_spec
        dest_spec = value_spec
        if not dest_spec.is_compatible(src_spec):
            raise TypeError(
                topology.message_on_path(
                    f'Cannot bind an incompatible value spec {dest_spec!r} '
                    f'to {self.__class__.__name__} with bound spec {src_spec!r}.',
                    path,
                )
            )
        return (False, self)

    list_spec = pg_typing.ensure_value_spec(
        value_spec, pg_typing.List(pg_typing.Any()), path
    )
    if list_spec is None:
        return (False, self)
    list_spec = cast(pg_typing.List, list_spec)
    for i, c in enumerate(self.candidates):
        list_spec.element.value.apply(
            c, self._allow_partial, root_path=path + f'candidates[{i}]'
        )
    self._value_spec = list_spec
    return (False, self)

OneOf

OneOf(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: Choices

N Choose 1.

Example:

# A single categorical choice:
v = pg.oneof([1, 2, 3])

# A hierarchical categorical choice:
v2 = pg.oneof([
    'foo',
    'bar',
    pg.oneof([1, 2, 3])
])

See also:

• pygx.oneof
• pygx.hyper.Choices
• pygx.hyper.ManyOf
• pygx.hyper.Float
• pygx.hyper.CustomHyper

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

on_sym_bound

on_sym_bound()

Event triggered when members are bound.

Source code in pygx/hyper/_categorical.py

def on_sym_bound(self):
    """Event triggered when members are bound."""
    super().on_sym_bound()
    assert self.num_choices == 1

encode

encode(value: Any) -> DNA

Encode a value into a DNA.

Source code in pygx/hyper/_categorical.py

def encode(self, value: Any) -> geno.DNA:
    """Encode a value into a DNA."""
    # NOTE(daiyip): Single choice DNA will automatically be pulled
    # up from children to current node. Thus we simply returns
    # encoded DNA from parent node.
    return super().encode([value])

custom_apply

custom_apply(
    path: KeyPath,
    value_spec: ValueSpec,
    allow_partial: bool,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
) -> tuple[bool, OneOf]

Validate candidates during value_spec binding time.

Source code in pygx/hyper/_categorical.py

def custom_apply(
    self,
    path: topology.KeyPath,
    value_spec: pg_typing.ValueSpec,
    allow_partial: bool,
    child_transform: None
    | (Callable[[topology.KeyPath, pg_typing.Field, Any], Any]) = None,
) -> tuple[bool, 'OneOf']:
    """Validate candidates during value_spec binding time."""
    # Check if value_spec directly accepts `self`.
    if value_spec.value_type and isinstance(self, value_spec.value_type):
        return (False, self)

    if self._value_spec:
        if not value_spec.is_compatible(self._value_spec):
            raise TypeError(
                topology.message_on_path(
                    f'Cannot bind an incompatible value spec {value_spec!r} '
                    f'to {self.__class__.__name__} with bound '
                    f'spec {self._value_spec!r}.',
                    path,
                )
            )
        return (False, self)

    for i, c in enumerate(self.candidates):
        value_spec.apply(
            c, self._allow_partial, root_path=path + f'candidates[{i}]'
        )
    self._value_spec = value_spec
    return (False, self)

CustomHyper

CustomHyper(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: HyperPrimitive

User-defined hyper primitive.

User-defined hyper primitive is useful when users want to have full control on the semantics and genome encoding of the search space. For example, the decision points are of variable length, which is not yet supported by built-in hyper primitives.

Using a user-defined hyper primitive is simple. The user should:

1) Subclass CustomHyper and implement the custom_decode method. It's optional to implement the custom_encode method, which is only necessary when the user wants to encode a material object into a DNA. 2) Introduce a DNAGenerator that can generate DNA for the pygx.geno.CustomDecisionPoint.

For example, the following code tries to find an optimal sub-sequence of an integer sequence by their sums:

import random

class IntSequence(pg.hyper.CustomHyper):

  def custom_decode(self, dna):
    return [int(v) for v in dna.value.split(',') if v != '']

class SubSequence(pg.algo.evolution.Mutator):

  def mutate(self, dna):
    genome = dna.value
    items = genome.split(',')
    start = random.randint(0, len(items))
    end = random.randint(start, len(items))
    new_genome = ','.join(items[start:end])
    return pg.DNA(new_genome, spec=dna.spec)

@pg.geno.dna_generator
def initial_population(dna_spec):
  yield pg.DNA('12,-34,56,-2,100,98', spec=dna_spec)

algo = pg.algo.evolution.Evolution(
    (pg.algo.evolution.selectors.Random(10)
     >> pg.algo.evolution.selectors.Top(1)
     >> SubSequence()),
    population_init=initial_population(),
    population_update=pg.algo.evolution.selectors.Last(20))

best_reward, best_example = None, None
for int_seq, feedback in pg.sample(IntSequence(), algo, num_examples=100):
  reward = sum(int_seq)
  if best_reward is None or best_reward < reward:
    best_reward, best_example = reward, int_seq
  feedback(reward)

print(best_reward, best_example)

Please note that user-defined hyper value can be used together with PyGX's built-in hyper primitives, for example:

pg.oneof([IntSequence(), None])

Therefore it's also a mechanism to extend PyGX's search space definitions.

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

custom_decode abstractmethod

custom_decode(dna: DNA) -> Any

Decode a DNA whose value is a string of user-defined genome.

Source code in pygx/hyper/_custom.py

@abc.abstractmethod
def custom_decode(self, dna: geno.DNA) -> Any:
    """Decode a DNA whose value is a string of user-defined genome."""

encode

encode(value: Any) -> DNA

Encode value into DNA with user-defined genome.

Source code in pygx/hyper/_custom.py

def encode(self, value: Any) -> geno.DNA:
    """Encode value into DNA with user-defined genome."""
    return self.custom_encode(value)

custom_encode

custom_encode(value: Any) -> DNA

Encode value to user defined genome.

Source code in pygx/hyper/_custom.py

def custom_encode(self, value: Any) -> geno.DNA:
    """Encode value to user defined genome."""
    raise NotImplementedError(
        f"'custom_encode' is not supported by {self.__class__.__name__!r}."
    )

dna_spec

dna_spec(location: KeyPath | None = None) -> DNASpec

Always returns CustomDecisionPoint for CustomHyper.

Source code in pygx/hyper/_custom.py

def dna_spec(
    self, location: topology.KeyPath | None = None
) -> geno.DNASpec:
    """Always returns CustomDecisionPoint for CustomHyper."""
    return geno.CustomDecisionPoint(
        hyper_type=self.__class__.__name__,
        next_dna_fn=self.next_dna,
        random_dna_fn=self.random_dna,
        hints=self.hints,
        name=self.name,
        location=location,
    )

first_dna

first_dna() -> DNA

Returns the first DNA of current sub-space.

Returns:

Type	Description
DNA	A string-valued DNA.

Source code in pygx/hyper/_custom.py

def first_dna(self) -> geno.DNA:
    """Returns the first DNA of current sub-space.

    Returns:
      A string-valued DNA.
    """
    if self.next_dna.__code__ is CustomHyper.next_dna.__code__:
        raise NotImplementedError(
            f'{self.__class__!r} must implement method `next_dna` to be used in '
            f'dynamic evaluation mode.'
        )
    dna = self.next_dna(None)
    assert dna is not None
    return dna

next_dna

next_dna(dna: DNA | None = None) -> DNA | None

Override to support sweeping via pg.geno.Sweeping.

Source code in pygx/hyper/_custom.py

def next_dna(self, dna: geno.DNA | None = None) -> geno.DNA | None:
    """Override to support sweeping via ``pg.geno.Sweeping``."""
    raise NotImplementedError(
        f'`next_dna` is not implemented in {self.__class__!r}'
    )

random_dna

random_dna(
    random_generator: ModuleType | Random | None = None,
    previous_dna: DNA | None = None,
) -> DNA

Override to support random sampling via pg.random_dna.

Source code in pygx/hyper/_custom.py

def random_dna(
    self,
    random_generator: types.ModuleType | random.Random | None = None,
    previous_dna: geno.DNA | None = None,
) -> geno.DNA:
    """Override to support random sampling via ``pg.random_dna``."""
    raise NotImplementedError(
        f'`random_dna` is not implemented in {self.__class__!r}'
    )

custom_apply

custom_apply(
    path: KeyPath,
    value_spec: ValueSpec,
    allow_partial: bool,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
) -> tuple[bool, CustomHyper]

Validate candidates during value_spec binding time.

Source code in pygx/hyper/_custom.py

def custom_apply(
    self,
    path: topology.KeyPath,
    value_spec: pg_typing.ValueSpec,
    allow_partial: bool,
    child_transform: None
    | (Callable[[topology.KeyPath, pg_typing.Field, Any], Any]) = None,
) -> tuple[bool, 'CustomHyper']:
    """Validate candidates during value_spec binding time."""
    del path, value_spec, allow_partial, child_transform
    # Allow custom hyper to be assigned to any type.
    return (False, self)

DerivedValue

DerivedValue(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: Object, CustomTyping

Base class for values that reference other values in an object tree.

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

derive abstractmethod

derive(*args: Any) -> Any

Derive the value from referenced values.

Source code in pygx/hyper/_derived.py

@abc.abstractmethod
def derive(self, *args: Any) -> Any:
    """Derive the value from referenced values."""

resolve

resolve(
    reference_path_or_paths: str | list[str] | None = None,
) -> tuple[Symbolic, KeyPath] | list[tuple[Symbolic, KeyPath]]

Resolve reference paths based on the location of this node.

Parameters:

Name	Type	Description	Default
reference_path_or_paths	str | list[str] | None	(Optional) a string or KeyPath as a reference path or a list of strings or KeyPath objects as a list of reference paths. If this argument is not provided, prebound reference paths of this object will be used.	None

Returns:

Type	Description
tuple[Symbolic, KeyPath] | list[tuple[Symbolic, KeyPath]]	A tuple (or list of tuple) of (resolved parent, resolved full path)

Source code in pygx/hyper/_derived.py

def resolve(
    self, reference_path_or_paths: str | list[str] | None = None
) -> (
    tuple[symbolic.Symbolic, topology.KeyPath]
    | list[tuple[symbolic.Symbolic, topology.KeyPath]]
):
    """Resolve reference paths based on the location of this node.

    Args:
      reference_path_or_paths: (Optional) a string or KeyPath as a reference
        path or a list of strings or KeyPath objects as a list of
        reference paths.
        If this argument is not provided, prebound reference paths of this
        object will be used.

    Returns:
      A tuple (or list of tuple) of (resolved parent, resolved full path)
    """
    single_input = False
    if reference_path_or_paths is None:
        reference_paths = self.reference_paths
    elif isinstance(reference_path_or_paths, str):
        reference_paths = [topology.KeyPath.parse(reference_path_or_paths)]
        single_input = True
    elif isinstance(reference_path_or_paths, topology.KeyPath):
        reference_paths = [reference_path_or_paths]
        single_input = True
    elif isinstance(reference_path_or_paths, list):
        paths = []
        for path in reference_path_or_paths:
            if isinstance(path, str):
                path = topology.KeyPath.parse(path)
            elif not isinstance(path, topology.KeyPath):
                raise ValueError(
                    "Argument 'reference_path_or_paths' must be None, "
                    'a string, KeyPath object, a list of strings, or a '
                    'list of KeyPath objects.'
                )
            paths.append(path)
        reference_paths = paths
    else:
        raise ValueError(
            "Argument 'reference_path_or_paths' must be None, "
            'a string, KeyPath object, a list of strings, or a '
            'list of KeyPath objects.'
        )

    resolved_paths = []
    for reference_path in reference_paths:
        parent = self.topo_parent
        while parent is not None and not reference_path.exists(parent):
            parent = getattr(parent, 'topo_parent', None)
        if parent is None:
            raise ValueError(
                f"Cannot resolve '{reference_path}': parent not found."
            )
        resolved_paths.append((parent, parent.topo_path + reference_path))
    return resolved_paths if not single_input else resolved_paths[0]

ValueReference

ValueReference(reference_paths: list[str], **kwargs)

Bases: DerivedValue

Class that represents a value referencing another value.

Source code in pygx/hyper/_derived.py

def __init__(self, reference_paths: list[str], **kwargs):
    # The `reference_paths` value spec accepts strings and transforms
    # them to `KeyPath`s during `apply`, but the annotation form
    # surfaces the post-transform type — silence the resulting
    # pyright mismatch at this boundary.
    super().__init__(reference_paths=reference_paths, **kwargs)  # pyright: ignore[reportArgumentType]

on_sym_bound

on_sym_bound()

Custom init.

Source code in pygx/hyper/_derived.py

def on_sym_bound(self):
    """Custom init."""
    super().on_sym_bound()
    if len(self.reference_paths) != 1:
        raise ValueError(
            f"Argument 'reference_paths' should have exact 1 "
            f'item. Encountered: {self.reference_paths}'
        )

derive

derive(referenced_value: Any) -> Any

Derive value by return a copy of the referenced value.

Source code in pygx/hyper/_derived.py

@override
# pylint: disable-next=arguments-differ
def derive(self, referenced_value: Any) -> Any:  # pyright: ignore[reportIncompatibleMethodOverride]
    """Derive value by return a copy of the referenced value."""
    return copy.copy(referenced_value)

custom_apply

custom_apply(
    path: KeyPath,
    value_spec: ValueSpec,
    allow_partial: bool,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
) -> tuple[bool, DerivedValue]

Implement pg_typing.CustomTyping interface.

Source code in pygx/hyper/_derived.py

def custom_apply(
    self,
    path: topology.KeyPath,
    value_spec: pg_typing.ValueSpec,
    allow_partial: bool,
    child_transform: None
    | (Callable[[topology.KeyPath, pg_typing.Field, Any], Any]) = None,
) -> tuple[bool, 'DerivedValue']:
    """Implement pg_typing.CustomTyping interface."""
    # TODO(daiyip): perform possible static analysis on referenced paths.
    del path, value_spec, allow_partial, child_transform
    return (False, self)

DynamicEvaluationContext

DynamicEvaluationContext(
    where: Callable[[HyperPrimitive], bool] | None = None,
    require_hyper_name: bool = False,
    per_thread: bool = True,
    dna_spec: DNASpec | None = None,
)

Context for dynamic evaluation of hyper primitives.

Example:

# Define a function that implicitly declares a search space.
def foo():
  return pg.oneof(range(-10, 10)) ** 2 + pg.oneof(range(-10, 10)) ** 2

# Define the search space by running `foo` once.
search_space = pg.hyper.DynamicEvaluationContext()
with search_space.collect():
  _ = foo()

# Create a search algorithm.
search_algorithm = pg.algo.evolution.regularized_evolution(
    pg.algo.evolution.mutators.Uniform(),
    population_size=32, tournament_size=16)

# Define the feedback loop.
best_foo, best_reward = None, None
for example, feedback in pg.sample(
    search_space, search_algorithm, num_examples=100):
  # Calling `example` returns a context manager under which `foo` is
  # connected with current search algorithm decisions.
  with example():
    reward = foo()
  feedback(reward)
  if best_reward is None or best_reward < reward:
    best_foo, best_reward = example, reward

Create a dynamic evaluation context.

Parameters:

Name	Type	Description	Default
where	Callable[[HyperPrimitive], bool] | None	A callable object that decide whether a hyper primitive should be included when being instantiated under collect. If None, all hyper primitives under collect will be included.	None
require_hyper_name	bool	If True, all hyper primitives (e.g. pg.oneof) must come with a name. This option helps to eliminate errors when a function that contains hyper primitive definition may be called multiple times. Since hyper primitives sharing the same name will be registered to the same decision point, repeated call to the hyper primitive definition will not matter.	False
per_thread	bool	If True, the context manager will be applied to current thread only. Otherwise, it will be applied on current process.	True
dna_spec	DNASpec | None	External provided search space. If None, the dynamic evaluation context can be used to create new search space via the collect context manager. Otherwise, current context will use the provided DNASpec to apply decisions.	None

Source code in pygx/hyper/_dynamic_evaluation.py

def __init__(
    self,
    where: Callable[[base.HyperPrimitive], bool] | None = None,
    require_hyper_name: bool = False,
    per_thread: bool = True,
    dna_spec: geno.DNASpec | None = None,
) -> None:  # pylint: disable=redefined-outer-name
    """Create a dynamic evaluation context.

    Args:
      where: A callable object that decide whether a hyper primitive should be
        included when being instantiated under `collect`.
        If None, all hyper primitives under `collect` will be
        included.
      require_hyper_name: If True, all hyper primitives (e.g. pg.oneof) must
        come with a `name`. This option helps to eliminate errors when a
        function that contains hyper primitive definition may be called multiple
        times. Since hyper primitives sharing the same name will be registered
        to the same decision point, repeated call to the hyper primitive
        definition will not matter.
      per_thread: If True, the context manager will be applied to current thread
        only. Otherwise, it will be applied on current process.
      dna_spec: External provided search space. If None, the dynamic evaluation
        context can be used to create new search space via the `collect`
        context manager. Otherwise, current context will use the provided
        DNASpec to apply decisions.
    """
    self._where = where
    self._require_hyper_name: bool = require_hyper_name
    self._name_to_hyper: dict[str, base.HyperPrimitive] = dict()
    self._annoymous_hyper_name_accumulator = (
        DynamicEvaluationContext._AnnoymousHyperNameAccumulator()
    )
    self._hyper_dict = symbolic.Dict() if dna_spec is None else None
    self._dna_spec: geno.DNASpec | None = dna_spec
    self._per_thread = per_thread
    self._decision_getter = None
    # Guards the registration tail of `add_decision_point` (#747).
    self._registration_lock = threading.Lock()

per_thread property

per_thread: bool

Returns True if current context collects/applies decisions per thread.

dna_spec property

dna_spec: DNASpec

Returns the DNASpec of the search space defined so far.

is_external property

is_external: bool

Returns True if the search space is defined by an external DNASpec.

hyper_dict property

hyper_dict: Dict | None

Returns collected hyper primitives as a dict.

None if current context is controlled by an external DNASpec.

collect

collect() -> Iterator[Dict]

A context manager for collecting hyper primitives within this context.

Within the scope, each hyper primitive is registered as a decision point in the search space and evaluates to its first value.

Example:

context = pg.hyper.DynamicEvaluationContext()
with context.collect():
  x = pg.oneof([1, 2, 3]) + pg.oneof([4, 5, 6])

# Prints 5 (1 + 4): each `pg.oneof` evaluated to its first
# candidate. Meanwhile 2 decision points were registered in the
# search space represented by the context.
print(x)

Yields:

Type	Description
Dict	The hyper dict representing the search space.

Source code in pygx/hyper/_dynamic_evaluation.py

@contextlib.contextmanager
def collect(self) -> Iterator[symbolic.Dict]:
    """A context manager for collecting hyper primitives within this context.

    Within the scope, each hyper primitive is registered as a decision
    point in the search space and evaluates to its first value.

    **Example:**

    ```python
    context = pg.hyper.DynamicEvaluationContext()
    with context.collect():
      x = pg.oneof([1, 2, 3]) + pg.oneof([4, 5, 6])

    # Prints 5 (1 + 4): each `pg.oneof` evaluated to its first
    # candidate. Meanwhile 2 decision points were registered in the
    # search space represented by the context.
    print(x)
    ```

    Yields:
      The hyper dict representing the search space.
    """
    if self.is_external:
        raise ValueError(
            f'`collect` cannot be called on a dynamic evaluation context that is '
            f'using an external DNASpec: {self._dna_spec}.'
        )

    # Ensure per-thread dynamic evaluation context will not be used
    # together with process-level dynamic evaluation context.
    _dynamic_evaluation_stack.ensure_thread_safety(self)

    self._hyper_dict = symbolic.Dict()
    with dynamic_evaluate(
        self.add_decision_point,  # pyright: ignore[reportArgumentType]
        per_thread=self._per_thread,
    ):
        try:
            # Push current context to dynamic evaluatoin stack so nested context
            # can defer unresolved hyper primitive to current context.
            _dynamic_evaluation_stack.push(self)
            yield self._hyper_dict

        finally:
            # Invalidate DNASpec.
            self._dna_spec = None

            # Pop current context from dynamic evaluatoin stack.
            _dynamic_evaluation_stack.pop(self)

add_decision_point

add_decision_point(hyper_primitive: HyperPrimitive)

Registers a parameter with current context and return its first value.

Source code in pygx/hyper/_dynamic_evaluation.py

def add_decision_point(self, hyper_primitive: base.HyperPrimitive):
    """Registers a parameter with current context and return its first value."""

    def _add_child_decision_point(c):
        if isinstance(c, types.LambdaType):
            s = pg_typing.signature(c, auto_typing=False, auto_doc=False)
            if not s.args and not s.has_wildcard_args:
                sub_context = DynamicEvaluationContext(
                    where=self._where, per_thread=self._per_thread
                )
                sub_context._annoymous_hyper_name_accumulator = (  # pylint: disable=protected-access
                    self._annoymous_hyper_name_accumulator
                )
                with sub_context.collect() as hyper_dict:
                    v = c()
                return (v, hyper_dict)
        return (c, c)

    if self._where and not self._where(hyper_primitive):
        # Delegate the resolution of hyper primitives that do not pass
        # the `where` predicate to its parent context.
        parent_context = _dynamic_evaluation_stack.get_parent(self)
        if parent_context is not None:
            return parent_context.add_decision_point(hyper_primitive)
        return hyper_primitive

    if isinstance(hyper_primitive, object_template.ObjectTemplate):
        return hyper_primitive.value

    assert isinstance(hyper_primitive, base.HyperPrimitive), hyper_primitive
    name = self._decision_name(hyper_primitive)
    if isinstance(hyper_primitive, categorical.Choices):
        # Dynamic evaluation always materializes a concrete bound choice,
        # so the open-ended `num_choices=None` form (any-count) is not
        # reachable here. Pin the narrowed type for the body below.
        assert hyper_primitive.num_choices is not None
        num_choices = hyper_primitive.num_choices
        candidate_values, candidates = zip(
            *[
                _add_child_decision_point(c)
                for c in hyper_primitive.candidates
            ]
        )
        if hyper_primitive.choices_distinct:
            assert num_choices <= len(hyper_primitive.candidates)
            v = [candidate_values[i] for i in range(num_choices)]
        else:
            v = [candidate_values[0]] * num_choices
        hyper_primitive = cast(
            base.HyperPrimitive,
            hyper_primitive.sym_clone(
                deep=True, override={'candidates': list(candidates)}
            ),
        )
        first_value = (
            v[0] if isinstance(hyper_primitive, categorical.OneOf) else v
        )
    elif isinstance(hyper_primitive, numerical.Float):
        first_value = hyper_primitive.min_value
    else:
        assert isinstance(hyper_primitive, custom.CustomHyper), (
            hyper_primitive
        )
        first_value = hyper_primitive.decode(hyper_primitive.first_dna())

    # The conflict check and the two stores are one atom (#747): without
    # the lock, two threads registering DIFFERENT primitives under the
    # same explicit name could both slip past the check and silently
    # overwrite instead of raising. Only the tail is locked — the
    # candidate walk above may recurse into child contexts and allocate
    # more names, so wrapping the whole method would self-deadlock.
    with self._registration_lock:
        if (
            name in self._name_to_hyper
            and hyper_primitive != self._name_to_hyper[name]
        ):
            raise ValueError(
                'Found different hyper primitives under the same name '
                f'{name!r}: '
                f'Instance1={self._name_to_hyper[name]!r}, '
                f'Instance2={hyper_primitive!r}.'
            )
        assert self._hyper_dict is not None
        self._hyper_dict[name] = hyper_primitive
        self._name_to_hyper[name] = hyper_primitive
    return first_value

apply

apply(decisions: DNA | list[int | float | str]) -> Iterator[None]

Context manager for applying decisions.

Within the scope, each hyper primitive evaluates to the value chosen by the corresponding decision, instead of returning a symbolic object.

Example:

def fun():
  return pg.oneof([1, 2, 3]) + pg.oneof([4, 5, 6])

context = pg.hyper.DynamicEvaluationContext()
with context.collect():
  fun()

with context.apply([0, 1]):
  # Prints 6 (1 + 5).
  print(fun())

Parameters:

Name	Type	Description	Default
decisions	DNA | list[int | float | str]	A DNA or a list of numbers or strings as decisions for current search space.	required

Yields:

Type	Description
None	None

Source code in pygx/hyper/_dynamic_evaluation.py

@contextlib.contextmanager
def apply(
    self, decisions: geno.DNA | list[int | float | str]
) -> Iterator[None]:
    """Context manager for applying decisions.

    Within the scope, each hyper primitive evaluates to the value chosen
    by the corresponding decision, instead of returning a symbolic
    object.

    **Example:**

    ```python
    def fun():
      return pg.oneof([1, 2, 3]) + pg.oneof([4, 5, 6])

    context = pg.hyper.DynamicEvaluationContext()
    with context.collect():
      fun()

    with context.apply([0, 1]):
      # Prints 6 (1 + 5).
      print(fun())
    ```

    Args:
      decisions: A DNA or a list of numbers or strings as decisions for
        current search space.

    Yields:
      None
    """
    if not isinstance(decisions, (geno.DNA, list)):
        raise ValueError(
            '`decisions` should be a DNA or a list of numbers.'
        )

    # Ensure per-thread dynamic evaluation context will not be used
    # together with process-level dynamic evaluation context.
    _dynamic_evaluation_stack.ensure_thread_safety(self)

    get_current_decision, evaluation_finalizer = (
        self._decision_getter_and_evaluation_finalizer(decisions)
    )

    has_errors = False
    with dynamic_evaluate(self.evaluate, per_thread=self._per_thread):  # pyright: ignore[reportArgumentType]
        try:
            # Set decision getter for current decision.
            self._decision_getter = get_current_decision

            # Push current context to dynamic evaluation stack so nested context
            # can delegate evaluate to current context.
            _dynamic_evaluation_stack.push(self)

            yield
        except Exception:
            has_errors = True
            raise
        finally:
            # Pop current context from dynamic evaluatoin stack.
            _dynamic_evaluation_stack.pop(self)

            # Reset decisions.
            self._decision_getter = None

            # Call evaluation finalizer to make sure all decisions are used.
            if not has_errors:
                evaluation_finalizer()

evaluate

evaluate(hyper_primitive: HyperPrimitive)

Evaluates a hyper primitive based on current decisions.

Source code in pygx/hyper/_dynamic_evaluation.py

def evaluate(self, hyper_primitive: base.HyperPrimitive):
    """Evaluates a hyper primitive based on current decisions."""
    if self._decision_getter is None:
        raise ValueError(
            '`evaluate` needs to be called under the `apply` context.'
        )

    get_current_decision = self._decision_getter

    def _apply_child(c):
        if isinstance(c, types.LambdaType):
            s = pg_typing.signature(c, auto_typing=False, auto_doc=False)
            if not s.args and not s.has_wildcard_args:
                return c()
        return c

    if self._where and not self._where(hyper_primitive):
        # Delegate the resolution of hyper primitives that do not pass
        # the `where` predicate to its parent context.
        parent_context = _dynamic_evaluation_stack.get_parent(self)
        if parent_context is not None:
            return parent_context.evaluate(hyper_primitive)
        return hyper_primitive

    if isinstance(hyper_primitive, numerical.Float):
        return get_current_decision(hyper_primitive)

    if isinstance(hyper_primitive, custom.CustomHyper):
        return hyper_primitive.decode(
            geno.DNA(get_current_decision(hyper_primitive))
        )

    assert isinstance(hyper_primitive, categorical.Choices), hyper_primitive
    assert hyper_primitive.num_choices is not None
    value = symbolic.List()
    for i in range(hyper_primitive.num_choices):
        # NOTE(daiyip): during registering the hyper primitives when
        # constructing the search space, we will need to evaluate every
        # candidate in order to pick up sub search spaces correctly, which is
        # not necessary for `pg.DynamicEvaluationContext.apply`.
        # `get_current_decision` returns `int | float | str`; for the
        # `Choices` branch the value is always the chosen-candidate
        # index (an int). Narrow at the call site so `candidates[...]`
        # type-checks cleanly.
        decision = get_current_decision(hyper_primitive, i)
        assert isinstance(decision, int), decision
        value.append(_apply_child(hyper_primitive.candidates[decision]))
    if isinstance(hyper_primitive, categorical.OneOf):
        assert len(value) == 1
        value = value[0]
    return value

Evolvable

Evolvable(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: CustomHyper

Hyper primitive for evolving an arbitrary symbolic object.

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

mutation_points_and_weights

mutation_points_and_weights(
    value: Symbolic,
) -> tuple[list[MutationPoint], list[float]]

Returns mutation points with weights for a symbolic tree.

Source code in pygx/hyper/_evolvable.py

def mutation_points_and_weights(
    self, value: symbolic.Symbolic
) -> tuple[list[MutationPoint], list[float]]:
    """Returns mutation points with weights for a symbolic tree."""
    mutation_points: list[MutationPoint] = []
    mutation_weights: list[float] = []

    def _choose_mutation_point(
        k: topology.KeyPath, v: Any, p: symbolic.Symbolic | None
    ):
        """Visiting function for a symbolic node."""

        def _add_point(mt: MutationType, k=k, v=v, p=p):
            mutation_points.append(
                MutationPoint(
                    mutation_type=mt, location=k, old_value=v, parent=p
                )
            )
            mutation_weights.append(self._weights(mt, k, v, p))

        if p is not None:
            # Stopping mutating current branch if metadata said so.
            f = p.sym_attr_field(k.key)
            if f and f.metadata and 'no_mutation' in f.metadata:
                return symbolic.TraverseAction.CONTINUE

        _add_point(MutationType.REPLACE)

        # Special handle list traversal to add insertion and deletion.
        if isinstance(v, symbolic.List):
            if v.value_spec:
                spec = v.value_spec
                reached_max_size = spec.max_size and len(v) == spec.max_size
                reached_min_size = spec.min_size and len(v) == spec.min_size
            else:
                reached_max_size = False
                reached_min_size = False

            for i, cv in enumerate(v):
                ck = topology.KeyPath(i, parent=k)
                if not reached_max_size:
                    _add_point(
                        MutationType.INSERT,
                        k=ck,
                        v=topology.MISSING_VALUE,
                        p=v,
                    )

                if not reached_min_size:
                    _add_point(MutationType.DELETE, k=ck, v=cv, p=v)

                # Replace type and value will be added in traverse.
                symbolic.traverse(
                    cv, _choose_mutation_point, root_path=ck, parent=v
                )
                if not reached_max_size and i == len(v) - 1:
                    _add_point(
                        MutationType.INSERT,
                        k=topology.KeyPath(i + 1, parent=k),
                        v=topology.MISSING_VALUE,
                        p=v,
                    )
            return symbolic.TraverseAction.CONTINUE
        return symbolic.TraverseAction.ENTER

    # First-order traverse the symbolic tree to compute
    # the mutation points and weights.
    symbolic.traverse(value, _choose_mutation_point)
    return mutation_points, mutation_weights

first_dna

first_dna() -> DNA

Returns the first DNA of current sub-space.

Source code in pygx/hyper/_evolvable.py

def first_dna(self) -> geno.DNA:
    """Returns the first DNA of current sub-space."""
    return self.custom_encode(self.initial_value)

random_dna

random_dna(
    random_generator: ModuleType | Random | None = None,
    previous_dna: DNA | None = None,
) -> DNA

Generates a random DNA.

Source code in pygx/hyper/_evolvable.py

def random_dna(
    self,
    random_generator: types.ModuleType | random.Random | None = None,
    previous_dna: geno.DNA | None = None,
) -> geno.DNA:
    """Generates a random DNA."""
    random_generator = random_generator or random
    if previous_dna is None:
        return self.first_dna()
    return self.custom_encode(
        self.mutate(self.custom_decode(previous_dna), random_generator)
    )

mutate

mutate(
    value: Symbolic, random_generator: ModuleType | Random | None = None
) -> Symbolic

Returns the next value for a symbolic value.

Source code in pygx/hyper/_evolvable.py

def mutate(
    self,
    value: symbolic.Symbolic,
    random_generator: types.ModuleType | random.Random | None = None,
) -> symbolic.Symbolic:
    """Returns the next value for a symbolic value."""
    r = random_generator or random
    points, weights = self.mutation_points_and_weights(value)
    [point] = r.choices(points, weights, k=1)

    # Mutating value.
    if point.mutation_type == MutationType.REPLACE:
        if point.location:
            value.sym_rebind(
                {
                    str(point.location): self.node_transform(
                        point.location, point.old_value, point.parent
                    )
                }
            )
        else:
            value = self.node_transform(
                point.location, point.old_value, point.parent
            )
    elif point.mutation_type == MutationType.INSERT:
        assert isinstance(point.parent, symbolic.List), point
        assert point.old_value == topology.MISSING_VALUE, point
        assert isinstance(point.location.key, int), point
        with symbolic.allow_writable_accessors():
            point.parent.insert(
                point.location.key,
                self.node_transform(
                    point.location, point.old_value, point.parent
                ),
            )
    else:
        assert point.mutation_type == MutationType.DELETE, point
        assert isinstance(point.parent, symbolic.List), point
        assert isinstance(point.location.key, int), point
        with symbolic.allow_writable_accessors():
            del point.parent[point.location.key]
    return value

MutationPoint dataclass

MutationPoint(
    mutation_type: MutationType,
    location: KeyPath,
    old_value: Any,
    parent: Symbolic | None,
)

Information about a candidate mutation point in a symbolic tree.

Attributes:

Name	Type	Description
mutation_type	MutationType	The type of the mutation.
location	KeyPath	The location where the mutation will take place.
old_value	Any	The value of the mutation point before mutation.
parent	Symbolic | None	The parent node of the mutation point.

MutationType

Bases: str, Enum

Mutation type.

Float

Float(
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any
)

Bases: HyperPrimitive

A continuous value within a range.

Example:

# A float value between 0.0 and 1.0.
v = pg.floatv(0.0, 1.0)

See also:

• pygx.floatv
• pygx.hyper.Choices
• pygx.hyper.OneOf
• pygx.hyper.ManyOf
• pygx.hyper.CustomHyper

Source code in pygx/symbolic/_object.py

@_filter_typing_traceback_on_raise
def __init__(
    self,
    *,
    allow_partial: bool = False,
    sealed: bool | None = None,
    root_path: topology.KeyPath | None = None,
    explicit_init: bool = False,
    **kwargs: Any,
):
    """Create an Object instance.

    ``pg.Object`` synthesizes a keyword-only ``__init__``. Subclasses
    that need positional arguments must override ``__init__``
    explicitly and translate to keyword arguments before forwarding
    to ``super().__init__(**kwargs)``.

    Args:
      allow_partial: If True, the object can be partial.
      sealed: If True, seal the object from future modification (unless under
        a `pg.as_sealed(False)` context manager). If False, treat the object as
        unsealed. If None, it's determined by the class's ``frozen`` option.
      root_path: The symbolic path for current object. By default it's None,
        which indicates that newly constructed object does not have a parent.
      explicit_init: Should set to `True` when `__init__` is called via
        `pg.Object.__init__` instead of `super().__init__`.
      **kwargs: key/value arguments that align with the schema. All required
        keys in the schema must be specified, and values should be acceptable
        according to their value spec.

    Raises:
      TypeError: When a required field is missing, a value has an
        unacceptable type, or an unknown keyword argument is given.
      ValueError: When a value violates its value spec's constraints
        (e.g. out of range).
    """
    # Placeholder for Google-internal usage instrumentation.

    cls = self.__class__
    # The ever-instantiated latch `apply_schema`'s post-instance guard
    # reads (#432: `update_schema` is only legal before instances exist).
    # Own-`__dict__`, latched on the construction ATTEMPT — mirroring the
    # native core, whose `__native_ctor_meta__` (the equivalent marker) is
    # created on the first instance op. One dict read per construct.
    meta = cls.__sym_meta__
    if not meta.instantiated:
        meta.instantiated = True
    if sealed is None:
        sealed = cls.__sym_options__.frozen

    if not isinstance(allow_partial, bool):
        raise TypeError(
            f"Expect bool type for argument 'allow_partial' in "
            f'symbolic.Object.__init__ but encountered {allow_partial}.'
        )

    # Fast path: const-keyed schema.
    if cls._sym_init_fields is not None and self._init_fast_path(
        kwargs,
        allow_partial,
        sealed,
        root_path,
        explicit_init,
    ):
        return

    # Slow path: handles varargs and schemas with non-const keys.

    # Model before-hook (#516) — the fast path was not taken
    # (`_sym_init_fields is None`), so the hook has not run yet.
    if cls._sym_preinit_overridden:
        kwargs = _apply_preinit(cls, kwargs)

    # We delay the seal attempt until members are all set.
    super().__init__(
        allow_partial=allow_partial,
        accessor_writable=cls.__sym_options__.attr_write,
        sealed=sealed,
        root_path=root_path,
        init_super=not explicit_init,
    )

    # Fill field_args and init_args from **kwargs.
    _, unmatched_keys = cls.__schema__.resolve(list(kwargs.keys()))
    # Fields opted out of `__init__` via `enable_init=False` resolve
    # against the schema but must still be rejected as kwargs.
    get_field = cls.__schema__.get_field
    unmatched_keys = list(unmatched_keys) + [
        k
        for k in kwargs
        if (f := get_field(k)) is not None and not f.enable_init
    ]
    if unmatched_keys:
        _screen_unexpected_init_keys(cls, kwargs, unmatched_keys)

    # Declared aliases are RESERVED wire spellings (#517, owner ruling):
    # an open (extra) schema must not accept an extra kwarg equal to a
    # field's alias — the wire would become ambiguous. (Const-keyed
    # schemas never reach here with an alias-spelled kwarg; their
    # unknown-key screen already raised — or, under `extra='ignore'`,
    # raised for alias keys before dropping the rest.)
    if cls._sym_alias_maps is not None:
        alias_to_name = cls._sym_alias_maps[0]
        for k in kwargs:
            if k in alias_to_name:
                raise TypeError(
                    f'{cls.__name__}.__init__() got keyword argument '
                    f'{k!r}, which is reserved as the alias of field '
                    f'{alias_to_name[k]!r}.'
                )

    # All field values come from kwargs — `pg.Object.__init__` is
    # keyword-only.
    field_args = dict(kwargs)

    # Check missing arguments when partial binding is disallowed.
    if not base.accepts_partial(self):
        missing_args = []
        for field in self.__class__.__schema__.fields.values():
            if (
                not field.value.has_default
                and isinstance(field.key, pg_typing.ConstStrKey)
                and field.key not in field_args
            ):
                missing_args.append(str(field.key))
        if missing_args:
            arg_phrase = formatting.auto_plural(
                len(missing_args), 'argument'
            )
            keys_str = formatting.comma_delimited_str(missing_args)
            raise TypeError(
                f'{self.__class__.__name__}.__init__() missing {len(missing_args)} '
                f'required {arg_phrase}: {keys_str}.'
            )

    # NOTE(daiyip): Accessor writable is honored by `Object.__setattr__` so
    # we can always make `_sym_attributes` accessor writable. This prevents
    # a child object's attribute access from being changed when it's
    # attached to a parent whose symbolic attributes are not writable.
    # Per-field validation is gated by `field.enable_validation` inside
    # `Schema.apply` / `Dict._formalized_value` — so a class-wide
    # `validate=False` (resolved into each field) and per-field overrides
    # both flow through naturally here.
    vars(self)['_sym_attributes'] = pg_dict.Dict(
        field_args,  # pyright: ignore[reportArgumentType]
        value_spec=self.__class__.sym_fields,
        allow_partial=allow_partial,
        sealed=sealed,
        accessor_writable=True,
        root_path=root_path,
        as_object_attributes_container=True,
        under_topo=cls.__sym_options__.topo,
    )
    # The write-time `needs_topo` refusal (see `_init_fast_path` Phase 5):
    # the slow path formalizes inside the container, so scan its items.
    if not cls.__sym_options__.topo:
        for k, v in self._sym_attributes.sym_items():
            base.refuse_tree_walking_inferential(cls, k, v)
    self._sym_attributes.topo_setparent(self)
    # Seed `init=False` fields from their defaults before `on_sym_post_init`
    # runs so `on_sym_ready`/`on_sym_bound` overrides start from a known baseline.
    _seed_non_symbolic_fields(self, cls._non_symbolic_fields)
    # Model after-validator (#516): fires before any lifecycle hook can
    # observe the object, gated like `on_sym_ready` (concrete only).
    if cls._sym_validate_overridden:
        abstract = (
            self._allow_partial and self.sym_partial
        ) or self.sym_puresymbolic
        if not abstract:
            self.on_sym_validate()
    self.on_sym_post_init()
    self.sym_seal(sealed)

on_sym_bound

on_sym_bound()

Constructor.

Source code in pygx/hyper/_numerical.py

def on_sym_bound(self):
    """Constructor."""
    super().on_sym_bound()
    if self.min_value > self.max_value:
        raise ValueError(
            f"'min_value' ({self.min_value}) is greater than 'max_value' "
            f'({self.max_value}).'
        )
    if self.scale in ['log', 'rlog'] and self.min_value <= 0:
        raise ValueError(
            f"'min_value' must be positive when `scale` is {self.scale!r}. "
            f'encountered: {self.min_value}.'
        )

dna_spec

dna_spec(location: KeyPath | None = None) -> Float

Returns corresponding DNASpec.

Source code in pygx/hyper/_numerical.py

def dna_spec(self, location: topology.KeyPath | None = None) -> geno.Float:
    """Returns corresponding DNASpec."""
    return geno.Float(
        min_value=self.min_value,
        max_value=self.max_value,
        scale=self.scale,
        hints=self.hints,
        name=self.name,
        location=location or topology.KeyPath(),
    )

encode

encode(value: float) -> DNA

Encode a float value into a DNA.

Source code in pygx/hyper/_numerical.py

def encode(self, value: float) -> geno.DNA:
    """Encode a float value into a DNA."""
    if not isinstance(value, float):
        raise ValueError(
            topology.message_on_path(
                f'Value should be float to be encoded for {self!r}. '
                f'Encountered {value}.',
                self.topo_path,
            )
        )
    if value < self.min_value:
        raise ValueError(
            topology.message_on_path(
                f'Value should be no less than {self.min_value}. '
                f'Encountered {value}.',
                self.topo_path,
            )
        )
    if value > self.max_value:
        raise ValueError(
            topology.message_on_path(
                f'Value should be no greater than {self.max_value}. '
                f'Encountered {value}.',
                self.topo_path,
            )
        )
    return geno.DNA(value)

custom_apply

custom_apply(
    path: KeyPath,
    value_spec: ValueSpec,
    allow_partial: bool = False,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
) -> tuple[bool, Float]

Validate candidates during value_spec binding time.

Source code in pygx/hyper/_numerical.py

def custom_apply(
    self,
    path: topology.KeyPath,
    value_spec: pg_typing.ValueSpec,
    allow_partial: bool = False,
    child_transform: None
    | (Callable[[topology.KeyPath, pg_typing.Field, Any], Any]) = None,
) -> tuple[bool, 'Float']:
    """Validate candidates during value_spec binding time."""
    del allow_partial
    del child_transform
    # Check if value_spec directly accepts `self`.
    if value_spec.value_type and isinstance(self, value_spec.value_type):
        return (False, self)

    float_spec = pg_typing.ensure_value_spec(
        value_spec, pg_typing.Float(), path
    )
    if float_spec is None:
        return (False, self)
    float_spec = cast(pg_typing.Float, float_spec)
    if (
        float_spec.min_value is not None
        and self.min_value < float_spec.min_value
    ):
        raise ValueError(
            topology.message_on_path(
                f'Float.min_value ({self.min_value}) should be no less than '
                f'the min value ({float_spec.min_value}) of value spec: '
                f'{float_spec}.',
                path,
            )
        )
    if (
        float_spec.max_value is not None
        and self.max_value > float_spec.max_value
    ):
        raise ValueError(
            topology.message_on_path(
                f'Float.max_value ({self.max_value}) should be no greater than '
                f'the max value ({float_spec.max_value}) of value spec: '
                f'{float_spec}.',
                path,
            )
        )
    return (False, self)

is_leaf

is_leaf() -> bool

Returns whether this is a leaf node.

Source code in pygx/hyper/_numerical.py

def is_leaf(self) -> bool:
    """Returns whether this is a leaf node."""
    return True

ObjectTemplate

ObjectTemplate(
    value: Any,
    compute_derived: bool = False,
    where: Callable[[HyperPrimitive], bool] | None = None,
)

Bases: HyperValue, Formattable

Object template that encodes and decodes symbolic values.

An object template can be created from a hyper value, which is a symbolic value with some parts placeheld by hyper primitives.

Example:

x = pg.Dict(
    a=0,
    b=pg.oneof(['foo', 'bar']),
    c=pg.manyof(2, [1, 2, 3]),
    d=pg.floatv(0.1, 0.5),
    e=pg.oneof([
        pg.Dict(f=pg.oneof([True, False])),
        pg.Dict(g=pg.manyof(2, [4, 5, 6], distinct=False)),
    ]),
)
t = pg.template(x)

In this example, the root template has 4 child hyper primitives associated with keys 'b', 'c', 'd' and 'e', while the hyper primitive 'e' has child decision points under keys 'f' and 'g' - one per candidate - creating a conditional search space.

The DNA shape is determined by the definition of the template, described by its geno.DNASpec (accessible via t.dna_spec()), which in this case renders as:

Space({
  0 = 'b': Choices(num_choices=1, [
    (0): 'foo'
    (1): 'bar'
  ])
  1 = 'c': Choices(num_choices=2, [
    (0): 1
    (1): 2
    (2): 3
  ])
  2 = 'd': Float(min_value=0.1, max_value=0.5)
  3 = 'e': Choices(num_choices=1, [
    (0): Space({
      0 = 'f': Choices(num_choices=1, [
        (0): True
        (1): False
      ])
    })
    (1): Space({
      0 = 'g': Choices(num_choices=2, [
        (0): 4
        (1): 5
        (2): 6
      ], distinct=False)
    })
  ])
})

A DNA of this spec can then be decoded into a concrete value:

assert t.decode(pg.DNA([0, [0, 2], 0.1, (0, 0)])) == pg.Dict(
    a=0, b='foo', c=[1, 3], d=0.1, e=pg.Dict(f=True))

A template can also work on only a subset of the hyper primitives from the input value through the where function. This is useful to partition a search space into parts for separate optimization. Note that filtering by path requires the enclosing object to track topology (pg.Dict and pg.List do; pg.Object subclasses need topo=True):

class A(pg.Object, topo=True):
  a: int
  b: int

t = pg.hyper.ObjectTemplate(
    A(a=pg.oneof([1, 2]), b=pg.oneof([3, 4])),
    where=lambda e: e.topo_path == 'a')
assert len(t.dna_spec().decision_points) == 1

# 'b' stays a hyper primitive in the decoded value.
assert t.decode(pg.DNA(0)) == A(a=1, b=pg.oneof([3, 4]))

Constructor.

Parameters:

Name	Type	Description	Default
value	Any	Value (maybe) annotated with generators to use as template.	required
compute_derived	bool	Whether to compute derived value at this level. We only want to compute derived value at root level since reference path may go out of scope of a non-root ObjectTemplate.	False
where	Callable[[HyperPrimitive], bool] | None	Function to filter hyper primitives. If None, all hyper primitives from value will be included in the encoding/decoding process. Otherwise only the hyper primitives on which 'where' returns True will be included. where can be useful to partition a search space into separate optimization processes. Please see 'ObjectTemplate' docstr for details.	None

Source code in pygx/hyper/_object_template.py

def __init__(
    self,
    value: Any,
    compute_derived: bool = False,
    where: Callable[[base.HyperPrimitive], bool] | None = None,
):
    """Constructor.

    Args:
      value: Value (maybe) annotated with generators to use as template.
      compute_derived: Whether to compute derived value at this level.
        We only want to compute derived value at root level since reference path
        may go out of scope of a non-root ObjectTemplate.
      where: Function to filter hyper primitives. If None, all hyper primitives
        from `value` will be included in the encoding/decoding process.
        Otherwise only the hyper primitives on which 'where' returns True will
        be included. `where` can be useful to partition a search space into
        separate optimization processes.
        Please see 'ObjectTemplate' docstr for details.
    """
    super().__init__()
    self._value = value
    self._root_path = topology.KeyPath()
    self._compute_derived = compute_derived
    self._where = where
    self._parse_generators()

root_path property writable

root_path: KeyPath

Returns root path.

value property

value: Any

Returns templated value.

hyper_primitives property

hyper_primitives: list[tuple[str, HyperValue]]

Returns hyper primitives in tuple (relative path, hyper primitive).

is_constant property

is_constant: bool

Returns whether current template is constant value.

dna_spec

dna_spec(location: KeyPath | None = None) -> Space

Return DNA spec (geno.Space) from this template.

Source code in pygx/hyper/_object_template.py

def dna_spec(self, location: topology.KeyPath | None = None) -> geno.Space:
    """Return DNA spec (geno.Space) from this template."""
    return geno.Space(
        elements=[
            primitive.dna_spec(primitive_location)
            for primitive_location, primitive in self._hyper_primitives
        ],
        location=location or topology.KeyPath(),
    )

encode

encode(value: Any) -> DNA

Encode a value into a DNA.

Example:

# DNA of a constant template:
template = pg.hyper.ObjectTemplate({'a': 0})
assert template.encode({'a': 0}) == pg.DNA(None)
# Raises: Unmatched value between template and input.
template.encode({'a': 1})

# DNA of a template containing only one pg.oneof.
template = pg.hyper.ObjectTemplate({'a': pg.oneof([1, 2])})
assert template.encode({'a': 1}) == pg.DNA(0)

# DNA of a template containing only one pg.floatv.
template = pg.hyper.ObjectTemplate({'a': pg.floatv(0.1, 1.0)})
assert template.encode({'a': 0.5}) == pg.DNA(0.5)

Parameters:

Name	Type	Description	Default
value	Any	Value to encode.	required

Returns:

Type	Description
DNA	Encoded DNA.

Raises:

Type	Description
ValueError	If value cannot be encoded by this template.

Source code in pygx/hyper/_object_template.py

def encode(self, value: Any) -> geno.DNA:
    """Encode a value into a DNA.

    **Example:**

    ```python
    # DNA of a constant template:
    template = pg.hyper.ObjectTemplate({'a': 0})
    assert template.encode({'a': 0}) == pg.DNA(None)
    # Raises: Unmatched value between template and input.
    template.encode({'a': 1})

    # DNA of a template containing only one pg.oneof.
    template = pg.hyper.ObjectTemplate({'a': pg.oneof([1, 2])})
    assert template.encode({'a': 1}) == pg.DNA(0)

    # DNA of a template containing only one pg.floatv.
    template = pg.hyper.ObjectTemplate({'a': pg.floatv(0.1, 1.0)})
    assert template.encode({'a': 0.5}) == pg.DNA(0.5)
    ```

    Args:
      value: Value to encode.

    Returns:
      Encoded DNA.

    Raises:
      ValueError: If `value` cannot be encoded by this template.
    """
    children = []
    # Mirror `_parse_generators`' identity de-duplication (#821): a
    # primitive reused at several locations contributes ONE child to the
    # DNA, or encode/decode would disagree on the child count.
    encoded_primitives: set[int] = set()

    def _encode(
        path: topology.KeyPath, template_value: Any, input_value: Any
    ) -> Any:
        """Encode input value according to template value."""
        if (
            pg_typing.MISSING_VALUE == input_value
            and pg_typing.MISSING_VALUE != template_value
        ):
            raise ValueError(f"Value is missing from input. Path='{path}'.")
        # Mirror `_parse_generators`: a `pg.Ref` around a hyper primitive is
        # that primitive (#821). Without this the template would see a `Ref`
        # where the decoded input holds a plain value.
        if isinstance(template_value, symbolic.Ref) and isinstance(
            template_value.value, base.HyperValue
        ):
            template_value = template_value.value
        if isinstance(template_value, base.HyperValue) and (
            not self._where or self._where(template_value)  # pyright: ignore[reportArgumentType]
        ):
            if id(template_value) in encoded_primitives:
                # An alias of an already-encoded primitive. Its input value
                # still has to agree with the one the primitive decided, but
                # it adds no new decision.
                return template_value
            encoded_primitives.add(id(template_value))
            children.append(template_value.encode(input_value))
        elif isinstance(template_value, derived.DerivedValue):
            if self._compute_derived:
                resolved = cast(
                    'list[tuple[symbolic.Symbolic, topology.KeyPath]]',
                    template_value.resolve(),
                )
                referenced_values = [
                    reference_path.query(value)
                    for _, reference_path in resolved
                ]
                derived_value = template_value.derive(*referenced_values)
                if derived_value != input_value:
                    raise ValueError(
                        f'Unmatched derived value between template and input. '
                        f"(Path='{path}', Template={template_value!r}, "
                        f'ComputedValue={derived_value!r}, Input={input_value!r})'
                    )
            # For template that doesn't compute derived value, it get passed over
            # to parent template who may be able to handle.
        elif isinstance(template_value, symbolic.Object):
            if type(input_value) is not type(template_value):
                raise ValueError(
                    f'Unmatched Object type between template and input: '
                    f"(Path='{path}', Template={template_value!r}, "
                    f'Input={input_value!r})'
                )
            template_keys = set(template_value.sym_keys())
            value_keys = set(input_value.sym_keys())
            if template_keys != value_keys:
                raise ValueError(
                    f'Unmatched Object keys between template value and input '
                    f"value. (Path='{path}', "
                    f'TemplateOnlyKeys={template_keys - value_keys}, '
                    f'InputOnlyKeys={value_keys - template_keys})'
                )
            for key in template_value.sym_keys():
                topology.merge_tree(
                    template_value.sym_getattr(key),
                    input_value.sym_getattr(key),
                    _encode,
                    root_path=topology.KeyPath(key, path),
                )
        elif isinstance(template_value, symbolic.Dict):
            # `merge_tree` recurses through matching dict pairs, so `_encode` is
            # only called for a `symbolic.Dict` template when `input_value` is not
            # also a dict.
            assert not isinstance(input_value, dict), (
                template_value,
                input_value,
            )
            raise ValueError(
                f'Unmatched dict between template value and input '
                f"value. (Path='{path}', Template={template_value!r}, "
                f'Input={input_value!r})'
            )
        elif isinstance(template_value, symbolic.List):
            if not isinstance(input_value, list) or len(input_value) != len(
                template_value
            ):
                raise ValueError(
                    f'Unmatched list between template value and input '
                    f"value. (Path='{path}', Template={template_value!r}, "
                    f'Input={input_value!r})'
                )
            for i, template_item in enumerate(template_value):
                topology.merge_tree(
                    template_item,
                    input_value[i],
                    _encode,
                    root_path=topology.KeyPath(i, path),
                )
        else:
            if template_value != input_value:
                raise ValueError(
                    'Unmatched value between template and input. '
                    f"(Path='{path}', "
                    f'Template={formatting.quote_if_str(template_value)}, '
                    f'Input={formatting.quote_if_str(input_value)})'
                )
        return template_value

    topology.merge_tree(
        self._value, value, _encode, root_path=self._root_path
    )
    return geno.DNA(None, children)

try_encode

try_encode(value: Any) -> tuple[bool, DNA | None]

Try to encode a value without raising an exception.

Returns:

Type	Description
tuple[bool, DNA | None]	A tuple of (success, DNA). DNA is None when encoding fails.

Source code in pygx/hyper/_object_template.py

def try_encode(self, value: Any) -> tuple[bool, geno.DNA | None]:
    """Try to encode a value without raising an exception.

    Returns:
      A tuple of (success, DNA). DNA is None when encoding fails.
    """
    try:
        dna = self.encode(value)
        return (True, dna)
    except ValueError:
        return (False, None)
    except KeyError:
        return (False, None)

format

format(
    compact: bool = False, verbose: bool = True, root_indent: int = 0, **kwargs
) -> str

Format this object.

Source code in pygx/hyper/_object_template.py

def format(
    self,
    compact: bool = False,
    verbose: bool = True,
    root_indent: int = 0,
    **kwargs,
) -> str:
    """Format this object."""
    details = formatting.format(
        self._value, compact, verbose, root_indent, **kwargs
    )
    return f'{self.__class__.__name__}(value={details})'

custom_apply

custom_apply(
    path: KeyPath,
    value_spec: ValueSpec,
    allow_partial: bool,
    child_transform: None | Callable[[KeyPath, Field, Any], Any] = None,
) -> tuple[bool, ObjectTemplate]

Validate candidates during value_spec binding time.

Source code in pygx/hyper/_object_template.py

def custom_apply(
    self,
    path: topology.KeyPath,
    value_spec: pg_typing.ValueSpec,
    allow_partial: bool,
    child_transform: None
    | (Callable[[topology.KeyPath, pg_typing.Field, Any], Any]) = None,
) -> tuple[bool, 'ObjectTemplate']:
    """Validate candidates during value_spec binding time."""
    # Check if value_spec directly accepts `self`.
    if not value_spec.value_type or not isinstance(
        self, value_spec.value_type
    ):
        value_spec.apply(
            self._value, allow_partial, root_path=self.root_path
        )
    return (False, self)

manyof

manyof(
    k: int,
    candidates: Iterable[Any],
    distinct: bool = True,
    sorted: bool = False,
    *,
    name: str | None = None,
    hints: Any | None = None,
    **kwargs: Any
) -> Any

N choose K.

Example:

class A(pg.Object):
  x: int

# Chooses 2 distinct candidates.
v = pg.manyof(2, [1, 2, 3])

# Chooses 2 non-distinct candidates.
v = pg.manyof(2, [1, 2, 3], distinct=False)

# Chooses 2 distinct candidates sorted by their indices.
v = pg.manyof(2, [1, 2, 3], sorted=True)

# A complex type as candidate.
v1 = pg.manyof(2, ['a', {'x': 1}, A(x=1)])

# A hierarchical categorical choice:
v2 = pg.manyof(2, [
    'foo',
    'bar',
    A(x=pg.oneof([1, 2, 3]))
])

Note

Under symbolic mode (the default), pg.manyof returns a pg.hyper.ManyOf object. Under dynamic evaluation mode, i.e. within the context of pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply, it evaluates to a concrete list of candidate values.

To use conditional search space in dynamic evaluation mode, the candidate should be wrapped with a lambda function, which is not necessary under symbolic mode. For example:

pg.manyof(2, [
   lambda: pg.oneof([0, 1], name='sub_a'),
   lambda: pg.floatv(0.0, 1.0, name='sub_b'),
   lambda: pg.manyof(2, ['a', 'b', 'c'], name='sub_c')
], name='root')

See also:

• pygx.hyper.ManyOf
• pygx.manyof
• pygx.floatv
• pygx.permutate
• pygx.evolve

Parameters:

Name	Type	Description	Default
k	int	Number of choices to make. Should be no larger than the length of candidates unless distinct is set to False.	required
candidates	Iterable[Any]	Candidates to select from. Items of candidate can be any type, therefore it can have nested hyper primitives, which forms a hierarchical search space.	required
distinct	bool	If True, each choice needs to be unique.	True
sorted	bool	If True, choices are sorted by their indices in the candidates.	False
name	str | None	A name that identifies the decision point in the search space. Instances created with the same name refer to the same decision point (see pg.oneof) - this is needed when the code that instantiates the hyper primitive may be called multiple times under a pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply context. In symbolic mode, named decision points must be unique within a search space.	None
hints	Any | None	An optional value which acts as a hint for the controller.	None
**kwargs	Any	Keyword arguments for backward compatibility. choices_distinct: Old name for distinct. choices_sorted: Old name for sorted.	{}

Returns:

Type	Description
Any	In symbolic mode, this function returns a ManyOf object.
Any	In dynamic evaluation mode, this function returns a list of items in
Any	candidates:
Any	under a pg.hyper.DynamicEvaluationContext.apply scope, it returns
Any	the list of selected candidates;
Any	under a pg.hyper.DynamicEvaluationContext.collect scope, it returns
Any	the first valid combination from the candidates. For example:
Any	```python
Any	Evaluates to [0, 1, 2] under collect.
Any	pg.manyof(3, range(5))
Any	Evaluates to [0, 0, 0] under collect.
Any	pg.manyof(3, range(5), distinct=False)
Any	```

Source code in pygx/hyper/_categorical.py

def manyof(
    k: int,
    candidates: Iterable[Any],
    distinct: bool = True,
    sorted: bool = False,  # pylint: disable=redefined-builtin
    *,
    name: str | None = None,
    hints: Any | None = None,
    **kwargs: Any,
) -> Any:
    """N choose K.

    **Example:**

    ```python
    class A(pg.Object):
      x: int

    # Chooses 2 distinct candidates.
    v = pg.manyof(2, [1, 2, 3])

    # Chooses 2 non-distinct candidates.
    v = pg.manyof(2, [1, 2, 3], distinct=False)

    # Chooses 2 distinct candidates sorted by their indices.
    v = pg.manyof(2, [1, 2, 3], sorted=True)

    # A complex type as candidate.
    v1 = pg.manyof(2, ['a', {'x': 1}, A(x=1)])

    # A hierarchical categorical choice:
    v2 = pg.manyof(2, [
        'foo',
        'bar',
        A(x=pg.oneof([1, 2, 3]))
    ])
    ```

    !!! note

        Under symbolic mode (the default), `pg.manyof` returns a
        [`pg.hyper.ManyOf`][pygx.hyper.ManyOf] object. Under dynamic
        evaluation mode, i.e. within the context of
        ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply``, it evaluates to a
        concrete list of candidate values.

        To use conditional search space in dynamic evaluation mode, the
        candidate should be wrapped with a `lambda` function, which is not
        necessary under symbolic mode. For example:

        ```python
        pg.manyof(2, [
           lambda: pg.oneof([0, 1], name='sub_a'),
           lambda: pg.floatv(0.0, 1.0, name='sub_b'),
           lambda: pg.manyof(2, ['a', 'b', 'c'], name='sub_c')
        ], name='root')
        ```

    See also:

      * [`pygx.hyper.ManyOf`][pygx.hyper.ManyOf]
      * [`pygx.manyof`][pygx.hyper.manyof]
      * [`pygx.floatv`][pygx.hyper.floatv]
      * [`pygx.permutate`][pygx.hyper.permutate]
      * [`pygx.evolve`][pygx.hyper.evolve]

    Args:
      k: Number of choices to make. Should be no larger than the length of
        `candidates` unless `distinct` is set to False.
      candidates: Candidates to select from. Items of candidate can be any
        type, therefore it can have nested hyper primitives, which forms a
        hierarchical search space.
      distinct: If True, each choice needs to be unique.
      sorted: If True, choices are sorted by their indices in the
        candidates.
      name: A name that identifies the decision point in the search space.
        Instances created with the same name refer to the same decision point
        (see [`pg.oneof`][pygx.hyper.oneof]) - this is needed when the code
        that instantiates the hyper primitive may be called multiple times
        under a ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply`` context. In symbolic
        mode, named decision points must be unique within a search space.
      hints: An optional value which acts as a hint for the controller.
      **kwargs: Keyword arguments for backward compatibility.
        `choices_distinct`: Old name for `distinct`.
        `choices_sorted`: Old name for `sorted`.

    Returns:
      In symbolic mode, this function returns a `ManyOf` object.
      In dynamic evaluation mode, this function returns a list of items in
      `candidates`:
      under a ``pg.hyper.DynamicEvaluationContext.apply`` scope, it returns
      the list of selected candidates;
      under a ``pg.hyper.DynamicEvaluationContext.collect`` scope, it returns
      the first valid combination from the `candidates`. For example:

      ```python
      # Evaluates to [0, 1, 2] under `collect`.
      pg.manyof(3, range(5))

      # Evaluates to [0, 0, 0] under `collect`.
      pg.manyof(3, range(5), distinct=False)
      ```

    """
    choices_distinct = kwargs.pop('choices_distinct', distinct)
    choices_sorted = kwargs.pop('choices_sorted', sorted)
    return ManyOf(
        num_choices=k,
        candidates=list(candidates),
        choices_distinct=choices_distinct,
        choices_sorted=choices_sorted,
        name=name,
        hints=hints,
    )

oneof

oneof(
    candidates: Iterable[Any],
    *,
    name: str | None = None,
    hints: Any | None = None
) -> Any

N choose 1.

Example:

class A(pg.Object):
  x: int

# A single categorical choice:
v = pg.oneof([1, 2, 3])

# A complex type as candidate.
v1 = pg.oneof(['a', {'x': 1}, A(x=1)])

# A hierarchical categorical choice:
v2 = pg.oneof([
    'foo',
    'bar',
    A(x=pg.oneof([1, 2, 3]))
])

Decision points at different call sites are distinct by default. Passing the same name merges them into ONE decision point: the search space has a single decision under that name, and every call evaluates to the same value. Primitives sharing a name must have identical definitions, otherwise an error is raised.

def fn():
  # Both calls carry the name 'x', so they refer to the same decision
  # point in the search space.
  return pg.oneof([1, 2, 3], name='x') + pg.oneof([1, 2, 3], name='x')

context = pg.hyper.trace(fn)
assert len(context.dna_spec.decision_points) == 1
with context.apply([2]):
  assert fn() == 3 + 3

See also:

• pygx.hyper.OneOf
• pygx.manyof
• pygx.floatv
• pygx.permutate
• pygx.evolve

Note

Under symbolic mode (the default), pg.oneof returns a pg.hyper.OneOf object. Under dynamic evaluation mode, i.e. within the context of pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply, it evaluates to a concrete candidate value.

To use conditional search space in dynamic evaluation mode, the candidate should be wrapped with a lambda function, which is not necessary under symbolic mode. For example:

pg.oneof([lambda: pg.oneof([0, 1], name='sub'), 2], name='root')

Parameters:

Name	Type	Description	Default
candidates	Iterable[Any]	Candidates to select from. Items of candidate can be any type, therefore it can have nested hyper primitives, which forms a hierarchical search space.	required
name	str | None	A name that identifies the decision point in the search space. Instances created with the same name refer to the same decision point (see above) - this is needed when the code that instantiates the hyper primitive may be called multiple times under a pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply context. In symbolic mode, named decision points must be unique within a search space.	None
hints	Any | None	An optional value which acts as a hint for the controller.	None

Returns:

Type	Description
Any	In symbolic mode, this function returns a OneOf object.
Any	In dynamic evaluation mode, this function returns one of the items in
Any	candidates:
Any	under a pg.hyper.DynamicEvaluationContext.apply scope, it returns
Any	the selected candidate;
Any	under a pg.hyper.DynamicEvaluationContext.collect scope, it returns
Any	the first candidate.

Source code in pygx/hyper/_categorical.py

def oneof(
    candidates: Iterable[Any],
    *,
    name: str | None = None,
    hints: Any | None = None,
) -> Any:
    """N choose 1.

    **Example:**

    ```python
    class A(pg.Object):
      x: int

    # A single categorical choice:
    v = pg.oneof([1, 2, 3])

    # A complex type as candidate.
    v1 = pg.oneof(['a', {'x': 1}, A(x=1)])

    # A hierarchical categorical choice:
    v2 = pg.oneof([
        'foo',
        'bar',
        A(x=pg.oneof([1, 2, 3]))
    ])
    ```

    Decision points at different call sites are distinct by default. Passing
    the same `name` merges them into ONE decision point: the search space has
    a single decision under that name, and every call evaluates to the same
    value. Primitives sharing a name must have identical definitions,
    otherwise an error is raised.

    ```python
    def fn():
      # Both calls carry the name 'x', so they refer to the same decision
      # point in the search space.
      return pg.oneof([1, 2, 3], name='x') + pg.oneof([1, 2, 3], name='x')

    context = pg.hyper.trace(fn)
    assert len(context.dna_spec.decision_points) == 1
    with context.apply([2]):
      assert fn() == 3 + 3
    ```

    See also:

      * [`pygx.hyper.OneOf`][pygx.hyper.OneOf]
      * [`pygx.manyof`][pygx.hyper.manyof]
      * [`pygx.floatv`][pygx.hyper.floatv]
      * [`pygx.permutate`][pygx.hyper.permutate]
      * [`pygx.evolve`][pygx.hyper.evolve]

    !!! note

        Under symbolic mode (the default), `pg.oneof` returns a
        [`pg.hyper.OneOf`][pygx.hyper.OneOf] object. Under dynamic evaluation
        mode, i.e. within the context of
        ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply``, it evaluates to a
        concrete candidate value.

        To use conditional search space in dynamic evaluation mode, the
        candidate should be wrapped with a `lambda` function, which is not
        necessary under symbolic mode. For example:

        ```python
        pg.oneof([lambda: pg.oneof([0, 1], name='sub'), 2], name='root')
        ```

    Args:
      candidates: Candidates to select from. Items of candidate can be any
        type, therefore it can have nested hyper primitives, which forms a
        hierarchical search space.
      name: A name that identifies the decision point in the search space.
        Instances created with the same name refer to the same decision point
        (see above) - this is needed when the code that instantiates the
        hyper primitive may be called multiple times under a
        ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply`` context. In symbolic
        mode, named decision points must be unique within a search space.
      hints: An optional value which acts as a hint for the controller.

    Returns:
      In symbolic mode, this function returns a `OneOf` object.
      In dynamic evaluation mode, this function returns one of the items in
      `candidates`:
      under a ``pg.hyper.DynamicEvaluationContext.apply`` scope, it returns
      the selected candidate;
      under a ``pg.hyper.DynamicEvaluationContext.collect`` scope, it returns
      the first candidate.
    """
    return OneOf(candidates=list(candidates), name=name, hints=hints)

permutate

permutate(
    candidates: Iterable[Any], name: str | None = None, hints: Any | None = None
) -> Any

Permutation of candidates.

Example:

class A(pg.Object):
  x: int

# Permutates the candidates.
v = pg.permutate([1, 2, 3])

# A complex type as candidate.
v1 = pg.permutate(['a', {'x': 1}, A(x=1)])

# A hierarchical categorical choice:
v2 = pg.permutate([
    'foo',
    'bar',
    A(x=pg.oneof([1, 2, 3]))
])

Note

Under symbolic mode (the default), pg.permutate returns a pg.hyper.ManyOf object. Under dynamic evaluation mode, i.e. within the context of pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply, it evaluates to a concrete permutation of the candidate values.

To use conditional search space in dynamic evaluation mode, the candidate should be wrapped with a lambda function, which is not necessary under symbolic mode. For example:

pg.permutate([
   lambda: pg.oneof([0, 1], name='sub_a'),
   lambda: pg.floatv(0.0, 1.0, name='sub_b'),
   lambda: pg.manyof(2, ['a', 'b', 'c'], name='sub_c')
], name='root')

See also:

• pygx.hyper.ManyOf
• pygx.oneof
• pygx.manyof
• pygx.floatv
• pygx.evolve

Parameters:

Name	Type	Description	Default
candidates	Iterable[Any]	Candidates to select from. Items of candidate can be any type, therefore it can have nested hyper primitives, which forms a hierarchical search space.	required
name	str | None	A name that identifies the decision point in the search space. Instances created with the same name refer to the same decision point (see pg.oneof) - this is needed when the code that instantiates the hyper primitive may be called multiple times under a pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply context. In symbolic mode, named decision points must be unique within a search space.	None
hints	Any | None	An optional value which acts as a hint for the controller.	None

Returns:

Type	Description
Any	In symbolic mode, this function returns a ManyOf object.
Any	In dynamic evaluation mode, this function returns a permutation of
Any	candidates:
Any	under a pg.hyper.DynamicEvaluationContext.apply scope, it returns
Any	a permutation of candidates based on controller decisions;
Any	under a pg.hyper.DynamicEvaluationContext.collect scope, it returns
Any	the first valid permutation. For example:
Any	```python
Any	Evaluates to [0, 1, 2, 3, 4] under collect.
Any	pg.permutate(range(5), name='numbers')
Any	```

Source code in pygx/hyper/_categorical.py

def permutate(
    candidates: Iterable[Any],
    name: str | None = None,
    hints: Any | None = None,
) -> Any:
    """Permutation of candidates.

    **Example:**

    ```python
    class A(pg.Object):
      x: int

    # Permutates the candidates.
    v = pg.permutate([1, 2, 3])

    # A complex type as candidate.
    v1 = pg.permutate(['a', {'x': 1}, A(x=1)])

    # A hierarchical categorical choice:
    v2 = pg.permutate([
        'foo',
        'bar',
        A(x=pg.oneof([1, 2, 3]))
    ])
    ```

    !!! note

        Under symbolic mode (the default), `pg.permutate` returns a
        [`pg.hyper.ManyOf`][pygx.hyper.ManyOf] object. Under dynamic
        evaluation mode, i.e. within the context of
        ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply``, it evaluates to a
        concrete permutation of the candidate values.

        To use conditional search space in dynamic evaluation mode, the
        candidate should be wrapped with a `lambda` function, which is not
        necessary under symbolic mode. For example:

        ```python
        pg.permutate([
           lambda: pg.oneof([0, 1], name='sub_a'),
           lambda: pg.floatv(0.0, 1.0, name='sub_b'),
           lambda: pg.manyof(2, ['a', 'b', 'c'], name='sub_c')
        ], name='root')
        ```

    See also:

      * [`pygx.hyper.ManyOf`][pygx.hyper.ManyOf]
      * [`pygx.oneof`][pygx.hyper.oneof]
      * [`pygx.manyof`][pygx.hyper.manyof]
      * [`pygx.floatv`][pygx.hyper.floatv]
      * [`pygx.evolve`][pygx.hyper.evolve]

    Args:
      candidates: Candidates to select from. Items of candidate can be any
        type, therefore it can have nested hyper primitives, which forms a
        hierarchical search space.
      name: A name that identifies the decision point in the search space.
        Instances created with the same name refer to the same decision point
        (see [`pg.oneof`][pygx.hyper.oneof]) - this is needed when the code
        that instantiates the hyper primitive may be called multiple times
        under a ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply`` context. In symbolic
        mode, named decision points must be unique within a search space.
      hints: An optional value which acts as a hint for the controller.

    Returns:
      In symbolic mode, this function returns a `ManyOf` object.
      In dynamic evaluation mode, this function returns a permutation of
      `candidates`:
      under a ``pg.hyper.DynamicEvaluationContext.apply`` scope, it returns
      a permutation of candidates based on controller decisions;
      under a ``pg.hyper.DynamicEvaluationContext.collect`` scope, it returns
      the first valid permutation. For example:

      ```python
      # Evaluates to [0, 1, 2, 3, 4] under `collect`.
      pg.permutate(range(5), name='numbers')
      ```

    """
    candidates = list(candidates)
    return manyof(
        len(candidates),
        candidates,
        choices_distinct=True,
        choices_sorted=False,
        name=name,
        hints=hints,
    )

reference

reference(reference_path: str) -> ValueReference

Creates a reference to the value at a relative path.

Parameters:

Name	Type	Description	Default
reference_path	str	Relative path of the referenced value, searched from the parent of the reference towards the root of the object tree.	required

Returns:

Type	Description
ValueReference	A ValueReference object that evaluates
ValueReference	to a copy of the referenced value.

Source code in pygx/hyper/_derived.py

def reference(reference_path: str) -> ValueReference:
    """Creates a reference to the value at a relative path.

    Args:
      reference_path: Relative path of the referenced value, searched from
        the parent of the reference towards the root of the object tree.

    Returns:
      A [`ValueReference`][pygx.hyper.ValueReference] object that evaluates
      to a copy of the referenced value.
    """
    return ValueReference(reference_paths=[reference_path])

dynamic_evaluate

dynamic_evaluate(
    evaluate_fn: Callable[[HyperValue], Any] | None,
    yield_value: Any = None,
    exit_fn: Callable[[], None] | None = None,
    per_thread: bool = True,
) -> Iterator[Any]

Eagerly evaluate hyper primitives within current scope.

Within the scope, creating a hyper primitive (e.g. pg.oneof) does not return the symbolic object; instead evaluate_fn is called with it and its result is returned.

Example:

def evaluate_fn(x: pg.hyper.HyperPrimitive):
  if isinstance(x, pg.hyper.OneOf):
    return x.candidates[0]
  raise NotImplementedError()

with pg.hyper.dynamic_evaluate(evaluate_fn):
  assert pg.oneof([0, 1, 2]) == 0

Please see apply as an example for using this method.

Parameters:

Name	Type	Description	Default
evaluate_fn	Callable[[HyperValue], Any] | None	A callable object that evaluates a hyper value (e.g. oneof, manyof, floatv) into a concrete value.	required
yield_value	Any	Value to yield return.	None
exit_fn	Callable[[], None] | None	A callable object to be called when exiting the context scope.	None
per_thread	bool	If True, the context manager will be applied to current thread only. Otherwise, it will be applied on current process.	True

Yields:

Type	Description
Any	yield_value from the argument.

Source code in pygx/hyper/_dynamic_evaluation.py

@contextlib.contextmanager
def dynamic_evaluate(
    evaluate_fn: Callable[[base.HyperValue], Any] | None,
    yield_value: Any = None,
    exit_fn: Callable[[], None] | None = None,
    per_thread: bool = True,
) -> Iterator[Any]:
    """Eagerly evaluate hyper primitives within current scope.

    Within the scope, creating a hyper primitive (e.g. `pg.oneof`) does not
    return the symbolic object; instead `evaluate_fn` is called with it and
    its result is returned.

    **Example:**

    ```python
    def evaluate_fn(x: pg.hyper.HyperPrimitive):
      if isinstance(x, pg.hyper.OneOf):
        return x.candidates[0]
      raise NotImplementedError()

    with pg.hyper.dynamic_evaluate(evaluate_fn):
      assert pg.oneof([0, 1, 2]) == 0
    ```

    Please see
    [`apply`][pygx.hyper.DynamicEvaluationContext.apply]
    as an example for using this method.

    Args:
      evaluate_fn: A callable object that evaluates a hyper value (e.g.
        oneof, manyof, floatv) into a concrete value.
      yield_value: Value to yield return.
      exit_fn: A callable object to be called when exiting the context scope.
      per_thread: If True, the context manager will be applied to current thread
        only. Otherwise, it will be applied on current process.

    Yields:
      `yield_value` from the argument.
    """
    if evaluate_fn is not None and not callable(evaluate_fn):
        raise ValueError(
            f"'evaluate_fn' must be either None or a callable object. "
            f'Encountered: {evaluate_fn!r}.'
        )
    if exit_fn is not None and not callable(exit_fn):
        raise ValueError(
            f"'exit_fn' must be a callable object. Encountered: {exit_fn!r}."
        )
    old_evaluate_fn = base.get_dynamic_evaluate_fn()
    has_errors = False
    try:
        base.set_dynamic_evaluate_fn(evaluate_fn, per_thread)
        yield yield_value
    except Exception:
        has_errors = True
        raise
    finally:
        base.set_dynamic_evaluate_fn(old_evaluate_fn, per_thread)
        if not has_errors and exit_fn is not None:
            exit_fn()

trace

trace(
    fun: Callable[[], Any],
    *,
    where: Callable[[HyperPrimitive], bool] | None = None,
    require_hyper_name: bool = False,
    per_thread: bool = True
) -> DynamicEvaluationContext

Trace the hyper primitives called within a function by executing it.

Example:

import pygx as pg

def fn():
  return pg.oneof([1, 2, 3]) + pg.floatv(0.0, 1.0)

context = pg.hyper.trace(fn)
with context.apply([0, 0.5]):
  assert fn() == 1.5

See more examples in DynamicEvaluationContext.

Parameters:

Name	Type	Description	Default
fun	Callable[[], Any]	Function in which the search space is defined.	required
where	Callable[[HyperPrimitive], bool] | None	A callable object that decide whether a hyper primitive should be included when being instantiated under collect. If None, all hyper primitives under collect will be included.	None
require_hyper_name	bool	If True, all hyper primitives defined in this scope will need to carry their names, which is usually a good idea when the function that instantiates the hyper primtives need to be called multiple times.	False
per_thread	bool	If True, the context manager will be applied to current thread only. Otherwise, it will be applied on current process.	True

Returns:

Type	Description
DynamicEvaluationContext	A DynamicEvaluationContext that can be passed to pg.sample.

Source code in pygx/hyper/_dynamic_evaluation.py

def trace(
    fun: Callable[[], Any],
    *,
    where: Callable[[base.HyperPrimitive], bool] | None = None,
    require_hyper_name: bool = False,
    per_thread: bool = True,
) -> DynamicEvaluationContext:
    """Trace the hyper primitives called within a function by executing it.

    **Example:**

    ```python
    import pygx as pg

    def fn():
      return pg.oneof([1, 2, 3]) + pg.floatv(0.0, 1.0)

    context = pg.hyper.trace(fn)
    with context.apply([0, 0.5]):
      assert fn() == 1.5
    ```

    See more examples in
    [`DynamicEvaluationContext`][pygx.hyper.DynamicEvaluationContext].

    Args:
      fun: Function in which the search space is defined.
      where: A callable object that decide whether a hyper primitive should be
        included when being instantiated under `collect`.
        If None, all hyper primitives under `collect` will be included.
      require_hyper_name: If True, all hyper primitives defined in this scope
        will need to carry their names, which is usually a good idea when the
        function that instantiates the hyper primtives need to be called multiple
        times.
      per_thread: If True, the context manager will be applied to current thread
        only. Otherwise, it will be applied on current process.

    Returns:
      A `DynamicEvaluationContext` that can be passed to ``pg.sample``.
    """
    context = DynamicEvaluationContext(
        where=where,
        require_hyper_name=require_hyper_name,
        per_thread=per_thread,
    )
    with context.collect():
        fun()
    return context

evolve

evolve(
    initial_value: Symbolic,
    node_transform: Callable[[KeyPath, Any, Symbolic], Any],
    *,
    weights: (
        None | Callable[[MutationType, KeyPath, Any, Symbolic], float]
    ) = None,
    name: str | None = None,
    hints: Any | None = None
) -> Evolvable

An evolvable symbolic value.

Example:

import random
import pygx as pg

class Foo(pg.Object):
  x: int
  y: int

class Bar(pg.Object):
  a: int
  b: int

# Defines possible transitions.
def node_transform(location, value, parent):
  if isinstance(value, Foo):
    return Bar(a=value.x, b=value.y)
  if location.key == 'x':
    return random.choice([1, 2, 3])
  if location.key == 'y':
    return random.choice([3, 4, 5])
  return value

v = pg.evolve(Foo(x=1, y=3), node_transform)

See also:

• pygx.hyper.Evolvable
• pygx.oneof
• pygx.manyof
• pygx.permutate
• pygx.floatv

Parameters:

Name	Type	Description	Default
initial_value	Symbolic	The initial value to evolve.	required
node_transform	Callable[[KeyPath, Any, Symbolic], Any]	A callable object that takes information of the value to operate (e.g. location, old value, parent node) and returns a new value as a replacement for the node. Such information allows users to not only access the mutation node, but the entire symbolic tree if needed, allowing complex mutation rules to be written with ease - for example - check adjacent nodes while modifying a list element. This function is designed to take care of both node replacements and node insertions. When insertion happens, the old value for the location will be pg.MISSING_VALUE.	required
weights	None | Callable[[MutationType, KeyPath, Any, Symbolic], float]	An optional callable object that returns the unnormalized (i.e. the probabilities don't have to sum to 1.0) mutation probabilities for all the nodes in the symbolic tree, based on (mutation type, location, old value, parent node). If None, all the locations and mutation types will be sampled uniformly.	None
name	str | None	An optional name of the decision point.	None
hints	Any | None	An optional hints for the decision point.	None

Returns:

Type	Description
Evolvable	A pg.hyper.Evolvable object.

Source code in pygx/hyper/_evolvable.py

def evolve(
    initial_value: symbolic.Symbolic,
    node_transform: Callable[
        [
            topology.KeyPath,  # Location.
            Any,  # Old value.
            # pg.MISSING_VALUE for insertion.
            symbolic.Symbolic,  # Parent node.
        ],
        Any,  # Replacement.
    ],
    *,
    weights: None
    | (
        Callable[
            [
                MutationType,  # Mutation type.
                topology.KeyPath,  # Location.
                Any,  # Value.
                symbolic.Symbolic,  # Parent.
            ],
            float,  # Mutation weight.
        ]
    ) = None,  # pylint: disable=bad-whitespace
    name: str | None = None,
    hints: Any | None = None,
) -> Evolvable:
    """An evolvable symbolic value.

    **Example:**

    ```python
    import random
    import pygx as pg

    class Foo(pg.Object):
      x: int
      y: int

    class Bar(pg.Object):
      a: int
      b: int

    # Defines possible transitions.
    def node_transform(location, value, parent):
      if isinstance(value, Foo):
        return Bar(a=value.x, b=value.y)
      if location.key == 'x':
        return random.choice([1, 2, 3])
      if location.key == 'y':
        return random.choice([3, 4, 5])
      return value

    v = pg.evolve(Foo(x=1, y=3), node_transform)
    ```

    See also:

      * [`pygx.hyper.Evolvable`][pygx.hyper.Evolvable]
      * [`pygx.oneof`][pygx.hyper.oneof]
      * [`pygx.manyof`][pygx.hyper.manyof]
      * [`pygx.permutate`][pygx.hyper.permutate]
      * [`pygx.floatv`][pygx.hyper.floatv]

    Args:
      initial_value: The initial value to evolve.
      node_transform: A callable object that takes information of the value to
        operate (e.g. location, old value, parent node) and returns a new value as
        a replacement for the node. Such information allows users to not only
        access the mutation node, but the entire symbolic tree if needed, allowing
        complex mutation rules to be written with ease - for example - check
        adjacent nodes while modifying a list element. This function is designed
        to take care of both node replacements and node insertions. When insertion
        happens, the old value for the location will be ``pg.MISSING_VALUE``.
      weights: An optional callable object that returns the unnormalized (i.e.
        the probabilities don't have to sum to 1.0) mutation
        probabilities for all the nodes in the symbolic tree, based on (mutation
        type, location, old value, parent node). If None, all the locations and
        mutation types will be sampled uniformly.
      name: An optional name of the decision point.
      hints: An optional hints for the decision point.

    Returns:
      A `pg.hyper.Evolvable` object.
    """
    return Evolvable(
        initial_value=initial_value,
        node_transform=node_transform,
        weights=weights,
        name=name,
        hints=hints,
    )

iterate

iterate(
    hyper_value: Any,
    num_examples: int | None = None,
    algorithm: DNAGenerator | None = None,
    where: Callable[[HyperPrimitive], bool] | None = None,
    force_feedback: bool = False,
) -> Iterator[Any]

Iterate a hyper value based on an algorithm.

Example:

hyper_dict = pg.Dict(x=pg.oneof([1, 2, 3]), y=pg.oneof(['a', 'b']))

# Get all examples from the hyper_dict.
assert list(pg.iter(hyper_dict)) == [
    pg.Dict(x=1, y='a'),
    pg.Dict(x=1, y='b'),
    pg.Dict(x=2, y='a'),
    pg.Dict(x=2, y='b'),
    pg.Dict(x=3, y='a'),
    pg.Dict(x=3, y='b'),
]

# Get the first two examples.
assert list(pg.iter(hyper_dict, 2)) == [
    pg.Dict(x=1, y='a'),
    pg.Dict(x=1, y='b'),
]

# Random sample examples, which is equivalent to `pg.random_sample`.
list(pg.iter(hyper_dict, 2, pg.geno.Random()))

# Iterate examples with feedback loop.
algo = pg.algo.evolution.regularized_evolution(
    pg.algo.evolution.mutators.Uniform())
for d, feedback in pg.iter(hyper_dict, 10, algo):
  feedback(d.x)

# Only materialize selected parts.
assert list(
    pg.iter(hyper_dict, where=lambda x: len(x.candidates) == 2)) == [
        pg.Dict(x=pg.oneof([1, 2, 3]), y='a'),
        pg.Dict(x=pg.oneof([1, 2, 3]), y='b'),
    ]

pg.iter differs from pg.sample in that it's designed for simple in-process iteration, which is handy for quickly generating examples from algorithms without maintaining trial states. On the contrary, pg.sample is designed for distributed sampling, with parallel workers and failover handling.

Parameters:

Name	Type	Description	Default
hyper_value	Any	A hyper value that represents a space of instances, or a pg.hyper.DynamicEvaluationContext. For the latter, each yielded example is a zero-arg callable returning a context manager, under which the traced function should be called again (see DynamicEvaluationContext).	required
num_examples	int | None	An optional integer as the max number of examples to yield. If None, iteration ends only when the algorithm is exhausted (e.g. Sweeping ends after enumerating the space; Random never ends).	None
algorithm	DNAGenerator | None	An optional DNA generator. If None, Sweeping will be used, which iterates examples in order.	None
where	Callable[[HyperPrimitive], bool] | None	Function to filter hyper primitives. If None, all hyper primitives from value will be included in the encoding/decoding process. Otherwise only the hyper primitives on which 'where' returns True will be included. where can be useful to partition a search space into separate optimization processes. Please see the pg.hyper.ObjectTemplate docstring for details.	None
force_feedback	bool	If True, always return the Feedback object together with the example, this is useful when the user want to pass different DNAGenerators to pg.iter and want to handle them uniformly.	False

Yields:

Type	Description
Any	A tuple of (example, feedback_fn) if the algorithm needs a feedback or
Any	force_feedback is True, otherwise the example.

Raises:

Type	Description
ValueError	when hyper_value is a constant value.

Source code in pygx/hyper/_iter.py

def iterate(
    hyper_value: Any,
    num_examples: int | None = None,
    algorithm: geno.DNAGenerator | None = None,
    where: Callable[[base.HyperPrimitive], bool] | None = None,
    force_feedback: bool = False,
) -> Iterator[Any]:
    """Iterate a hyper value based on an algorithm.

    **Example:**

    ```python
    hyper_dict = pg.Dict(x=pg.oneof([1, 2, 3]), y=pg.oneof(['a', 'b']))

    # Get all examples from the hyper_dict.
    assert list(pg.iter(hyper_dict)) == [
        pg.Dict(x=1, y='a'),
        pg.Dict(x=1, y='b'),
        pg.Dict(x=2, y='a'),
        pg.Dict(x=2, y='b'),
        pg.Dict(x=3, y='a'),
        pg.Dict(x=3, y='b'),
    ]

    # Get the first two examples.
    assert list(pg.iter(hyper_dict, 2)) == [
        pg.Dict(x=1, y='a'),
        pg.Dict(x=1, y='b'),
    ]

    # Random sample examples, which is equivalent to `pg.random_sample`.
    list(pg.iter(hyper_dict, 2, pg.geno.Random()))

    # Iterate examples with feedback loop.
    algo = pg.algo.evolution.regularized_evolution(
        pg.algo.evolution.mutators.Uniform())
    for d, feedback in pg.iter(hyper_dict, 10, algo):
      feedback(d.x)

    # Only materialize selected parts.
    assert list(
        pg.iter(hyper_dict, where=lambda x: len(x.candidates) == 2)) == [
            pg.Dict(x=pg.oneof([1, 2, 3]), y='a'),
            pg.Dict(x=pg.oneof([1, 2, 3]), y='b'),
        ]
    ```

    ``pg.iter`` differs from `pg.sample` in that it's designed
    for simple in-process iteration, which is handy for quickly generating
    examples from algorithms without maintaining trial states. On the contrary,
    `pg.sample` is designed for distributed sampling, with parallel workers and
    failover handling.

    Args:
      hyper_value: A hyper value that represents a space of instances, or a
        `pg.hyper.DynamicEvaluationContext`. For the latter, each yielded
        example is a zero-arg callable returning a context manager, under
        which the traced function should be called again (see
        [`DynamicEvaluationContext`][pygx.hyper.DynamicEvaluationContext]).
      num_examples: An optional integer as the max number of examples to
        yield. If None, iteration ends only when the algorithm is exhausted
        (e.g. `Sweeping` ends after enumerating the space; `Random` never
        ends).
      algorithm: An optional DNA generator. If None, Sweeping will be used, which
          iterates examples in order.
      where: Function to filter hyper primitives. If None, all hyper primitives
        from `value` will be included in the encoding/decoding process. Otherwise
        only the hyper primitives on which 'where' returns True will be included.
        `where` can be useful to partition a search space into separate
        optimization processes. Please see the `pg.hyper.ObjectTemplate`
        docstring for details.
      force_feedback: If True, always return the Feedback object together
        with the example, this is useful when the user want to pass different
        DNAGenerators to `pg.iter` and want to handle them uniformly.

    Yields:
      A tuple of (example, feedback_fn) if the algorithm needs a feedback or
      `force_feedback` is True, otherwise the example.

    Raises:
      ValueError: when `hyper_value` is a constant value.
    """
    if isinstance(hyper_value, dynamic_evaluation.DynamicEvaluationContext):
        dynamic_evaluation_context = hyper_value
        spec = hyper_value.dna_spec
        t = None
    else:
        t = object_template.template(hyper_value, where)
        if t.is_constant:
            raise ValueError(
                f"'hyper_value' is a constant value: {hyper_value!r}."
            )
        dynamic_evaluation_context = None
        spec = t.dna_spec()

    if algorithm is None:
        algorithm = geno.Sweeping()

    # NOTE(daiyip): algorithm can continue if it's already set up with the same
    # DNASpec, or we will setup the algorithm with the DNASpec from the template.
    if algorithm.dna_spec is None:
        algorithm.setup(spec)
    elif symbolic.ne(spec, algorithm.dna_spec):
        raise ValueError(
            f'{algorithm!r} has been set up with a different DNASpec. '
            f'Existing: {algorithm.dna_spec!r}, New: {spec!r}.'
        )

    count = 0
    while num_examples is None or count < num_examples:
        try:
            count += 1
            dna = algorithm.propose()
            if t is not None:
                example = t.decode(dna)
            else:
                assert dynamic_evaluation_context is not None
                # pylint: disable-next=unnecessary-lambda-assignment
                example = lambda: dynamic_evaluation_context.apply(dna)
            if force_feedback or algorithm.needs_feedback:
                yield example, Feedback(algorithm, dna)
            else:
                yield example
        except StopIteration:
            return

random_sample

random_sample(
    value: Any,
    num_examples: int | None = None,
    where: Callable[[HyperPrimitive], bool] | None = None,
    seed: int | None = None,
) -> Iterator[Any]

Returns an iterator of random sampled examples.

Example:

hyper_dict = pg.Dict(x=pg.oneof(range(3)), y=pg.floatv(0.0, 1.0))

# Generate one random example from the hyper_dict.
d = next(pg.random_sample(hyper_dict))

# Generate 5 random examples with random seed.
ds = list(pg.random_sample(hyper_dict, 5, seed=1))

# Generate 3 random examples of `x` with `y` intact.
ds = list(pg.random_sample(hyper_dict, 3,
    where=lambda x: isinstance(x, pg.hyper.OneOf)))

Parameters:

Name	Type	Description	Default
value	Any	A (maybe) hyper value.	required
num_examples	int | None	An optional integer as number of examples to yield. If None, the returned iterator iterates forever.	None
where	Callable[[HyperPrimitive], bool] | None	Function to filter hyper primitives. If None, all hyper primitives in value will be included in the encoding/decoding process. Otherwise only the hyper primitives on which 'where' returns True will be included. where can be useful to partition a search space into separate optimization processes. Please see the pg.hyper.ObjectTemplate docstring for details.	None
seed	int | None	An optional integer as random seed.	None

Returns:

Type	Description
Iterator[Any]	Iterator of random examples.

Source code in pygx/hyper/_iter.py

def random_sample(
    value: Any,
    num_examples: int | None = None,
    where: Callable[[base.HyperPrimitive], bool] | None = None,
    seed: int | None = None,
) -> Iterator[Any]:
    """Returns an iterator of random sampled examples.

    **Example:**

    ```python
    hyper_dict = pg.Dict(x=pg.oneof(range(3)), y=pg.floatv(0.0, 1.0))

    # Generate one random example from the hyper_dict.
    d = next(pg.random_sample(hyper_dict))

    # Generate 5 random examples with random seed.
    ds = list(pg.random_sample(hyper_dict, 5, seed=1))

    # Generate 3 random examples of `x` with `y` intact.
    ds = list(pg.random_sample(hyper_dict, 3,
        where=lambda x: isinstance(x, pg.hyper.OneOf)))
    ```

    Args:
      value: A (maybe) hyper value.
      num_examples: An optional integer as number of examples to yield. If
        None, the returned iterator iterates forever.
      where: Function to filter hyper primitives. If None, all hyper primitives in
        `value` will be included in the encoding/decoding process. Otherwise only
        the hyper primitives on which 'where' returns True will be included.
        `where` can be useful to partition a search space into separate
        optimization processes. Please see the `pg.hyper.ObjectTemplate`
        docstring for details.
      seed: An optional integer as random seed.

    Returns:
      Iterator of random examples.
    """
    return iterate(value, num_examples, geno.Random(seed), where=where)

floatv

floatv(
    min_value: float,
    max_value: float,
    scale: str | None = None,
    *,
    name: str | None = None,
    hints: Any | None = None
) -> Any

A continuous value within a range.

Example:

# A continuous value within [0.0, 1.0]
v = pg.floatv(0.0, 1.0)

See also:

• pygx.hyper.Float
• pygx.oneof
• pygx.manyof
• pygx.permutate
• pygx.evolve

Note

Under symbolic mode (the default), pg.floatv returns a pg.hyper.Float object. Under dynamic evaluation mode, i.e. within the context of pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply, it evaluates to a concrete float value.

Parameters:

Name	Type	Description	Default
min_value	float	Minimum acceptable value (inclusive).	required
max_value	float	Maximum acceptable value (inclusive).	required
scale	str | None	An optional string as the scale of the range. Supported values are None, 'linear', 'log', and 'rlog'. If None, the feasible space is unscaled. If linear, the feasible space is mapped to [0, 1] linearly. If log, the feasible space is mapped to [0, 1] logarithmically with formula x -> log(x / min) / log(max / min). If rlog, the feasible space is mapped to [0, 1] "reverse" logarithmically, resulting in values close to max_value spread out more than the points near the min_value, with formula: x -> 1.0 - log((max + min - x) / min) / log (max / min). min_value must be positive if scale is not None. Also, it depends on the search algorithm to decide whether this information is used or not.	None
name	str | None	A name that identifies the decision point in the search space. Instances created with the same name refer to the same decision point (see pg.oneof) - this is needed when the code that instantiates the hyper primitive may be called multiple times under a pg.hyper.DynamicEvaluationContext.collect or pg.hyper.DynamicEvaluationContext.apply context. In symbolic mode, named decision points must be unique within a search space.	None
hints	Any | None	An optional value which acts as a hint for the controller.	None

Returns:

Type	Description
Any	In symbolic mode, this function returns a Float object.
Any	In dynamic evaluation mode, this function returns a float value that is
Any	no less than the min_value and no greater than the max_value:
Any	under a pg.hyper.DynamicEvaluationContext.apply scope, it returns
Any	the chosen float value from the controller decisions;
Any	under a pg.hyper.DynamicEvaluationContext.collect scope, it returns
Any	min_value.

Source code in pygx/hyper/_numerical.py

def floatv(
    min_value: float,
    max_value: float,
    scale: str | None = None,
    *,
    name: str | None = None,
    hints: Any | None = None,
) -> Any:
    """A continuous value within a range.

    **Example:**

    ```python
    # A continuous value within [0.0, 1.0]
    v = pg.floatv(0.0, 1.0)
    ```

    See also:

      * [`pygx.hyper.Float`][pygx.hyper.Float]
      * [`pygx.oneof`][pygx.hyper.oneof]
      * [`pygx.manyof`][pygx.hyper.manyof]
      * [`pygx.permutate`][pygx.hyper.permutate]
      * [`pygx.evolve`][pygx.hyper.evolve]

    !!! note

        Under symbolic mode (the default), `pg.floatv` returns a
        [`pg.hyper.Float`][pygx.hyper.Float] object. Under dynamic evaluation
        mode, i.e. within the context of
        ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply``, it evaluates to a
        concrete float value.

    Args:
      min_value: Minimum acceptable value (inclusive).
      max_value: Maximum acceptable value (inclusive).
      scale: An optional string as the scale of the range. Supported values
        are None, 'linear', 'log', and 'rlog'.
        If None, the feasible space is unscaled.
        If `linear`, the feasible space is mapped to [0, 1] linearly.
        If `log`, the feasible space is mapped to [0, 1] logarithmically with
          formula `x -> log(x / min) / log(max / min)`.
        If `rlog`, the feasible space is mapped to [0, 1] "reverse"
          logarithmically, resulting in values close to `max_value` spread
          out more than the points near the `min_value`, with formula:
          x -> 1.0 - log((max + min - x) / min) / log (max / min).
        `min_value` must be positive if `scale` is not None.
        Also, it depends on the search algorithm to decide whether this
        information is used or not.
      name: A name that identifies the decision point in the search space.
        Instances created with the same name refer to the same decision point
        (see [`pg.oneof`][pygx.hyper.oneof]) - this is needed when the code
        that instantiates the hyper primitive may be called multiple times
        under a ``pg.hyper.DynamicEvaluationContext.collect`` or
        ``pg.hyper.DynamicEvaluationContext.apply`` context. In symbolic
        mode, named decision points must be unique within a search space.
      hints: An optional value which acts as a hint for the controller.

    Returns:
      In symbolic mode, this function returns a `Float` object.
      In dynamic evaluation mode, this function returns a float value that is
      no less than the `min_value` and no greater than the `max_value`:
      under a ``pg.hyper.DynamicEvaluationContext.apply`` scope, it returns
      the chosen float value from the controller decisions;
      under a ``pg.hyper.DynamicEvaluationContext.collect`` scope, it returns
      `min_value`.
    """
    return Float(
        min_value=min_value,
        max_value=max_value,
        scale=scale,  # pyright: ignore[reportArgumentType]
        name=name,
        hints=hints,
    )

dna_spec

dna_spec(
    value: Any, where: Callable[[HyperPrimitive], bool] | None = None
) -> DNASpec

Returns the DNASpec from a (maybe) hyper value.

Example:

hyper = pg.Dict(x=pg.oneof([1, 2, 3]), y=pg.oneof(['a', 'b']))
spec = pg.dna_spec(hyper)

assert spec.space_size == 6
assert len(spec.decision_points) == 2
print(spec.decision_points)

# Select a partial space with `where` argument.
spec = pg.dna_spec(hyper, where=lambda x: len(x.candidates) == 2)

assert spec.space_size == 2
assert len(spec.decision_points) == 1

See also:

• pygx.DNASpec
• pygx.DNA

Parameters:

Name	Type	Description	Default
value	Any	A (maybe) hyper value.	required
where	Callable[[HyperPrimitive], bool] | None	Function to filter hyper primitives. If None, all hyper primitives from value will be included in the encoding/decoding process. Otherwise only the hyper primitives on which 'where' returns True will be included. where can be very useful to partition a search space into separate optimization processes. Please see the ObjectTemplate docstring for details.	None

Returns:

Type	Description
DNASpec	A DNASpec object, which represents the search space from algorithm's view.

Source code in pygx/hyper/_object_template.py

def dna_spec(
    value: Any, where: Callable[[base.HyperPrimitive], bool] | None = None
) -> geno.DNASpec:
    """Returns the DNASpec from a (maybe) hyper value.

    **Example:**

    ```python
    hyper = pg.Dict(x=pg.oneof([1, 2, 3]), y=pg.oneof(['a', 'b']))
    spec = pg.dna_spec(hyper)

    assert spec.space_size == 6
    assert len(spec.decision_points) == 2
    print(spec.decision_points)

    # Select a partial space with `where` argument.
    spec = pg.dna_spec(hyper, where=lambda x: len(x.candidates) == 2)

    assert spec.space_size == 2
    assert len(spec.decision_points) == 1
    ```

    See also:

      * [`pygx.DNASpec`][pygx.geno.DNASpec]
      * [`pygx.DNA`][pygx.geno.DNA]

    Args:
      value: A (maybe) hyper value.
      where: Function to filter hyper primitives. If None, all hyper
        primitives from `value` will be included in the encoding/decoding
        process. Otherwise only the hyper primitives on which 'where' returns
        True will be included. `where` can be very useful to partition a
        search space into separate optimization processes. Please see the
        `ObjectTemplate` docstring for details.

    Returns:
      A DNASpec object, which represents the search space from algorithm's view.
    """
    return template(value, where).dna_spec()

materialize

materialize(
    value: Any,
    parameters: DNA | dict[str, Any],
    use_literal_values: bool = False,
    where: Callable[[HyperPrimitive], bool] | None = None,
) -> Any

Materialize a (maybe) hyper value using a DNA or parameter dict.

Example:

hyper_dict = pg.Dict(x=pg.oneof(['a', 'b']), y=pg.floatv(0.0, 1.0))

# Materialize using DNA.
assert pg.materialize(
  hyper_dict, pg.DNA([0, 0.5])) == pg.Dict(x='a', y=0.5)

# Materialize using key value pairs.
# See `pg.DNA.from_dict` for more details.
assert pg.materialize(
  hyper_dict, {'x': 0, 'y': 0.5}) == pg.Dict(x='a', y=0.5)

# Partially materialize.
v = pg.materialize(
  hyper_dict, pg.DNA(0), where=lambda x: isinstance(x, pg.hyper.OneOf))
assert v == pg.Dict(x='a', y=pg.floatv(0.0, 1.0))

Parameters:

Name	Type	Description	Default
value	Any	A (maybe) hyper value	required
parameters	DNA | dict[str, Any]	A DNA object, or a dict from each decision point's ID (key path) or name to its decision. A decision can be an integer as the candidate index for geno.Choices, a float for geno.Float, or a string - either a candidate's literal value or the '/' form for geno.Choices. (See pg.DNA.from_dict for more details.)	required
use_literal_values	bool	If True, integer decisions in parameters are interpreted as literal values of the corresponding choice rather than candidate indices. String decisions are always accepted as either literal values or '/' forms.	False
where	Callable[[HyperPrimitive], bool] | None	Function to filter hyper primitives. If None, all hyper primitives from value will be included in the encoding/decoding process. Otherwise only the hyper primitives on which 'where' returns True will be included. where can be useful to partition a search space into separate optimization processes. Please see the ObjectTemplate docstring for details.	None

Returns:

Type	Description
Any	A materialized value.

Raises:

Type	Description
TypeError	if parameters is not a DNA or dict.
ValueError	if parameters cannot be decoded.

Source code in pygx/hyper/_object_template.py

def materialize(
    value: Any,
    parameters: geno.DNA | dict[str, Any],
    use_literal_values: bool = False,
    where: Callable[[base.HyperPrimitive], bool] | None = None,
) -> Any:
    """Materialize a (maybe) hyper value using a DNA or parameter dict.

    **Example:**

    ```python
    hyper_dict = pg.Dict(x=pg.oneof(['a', 'b']), y=pg.floatv(0.0, 1.0))

    # Materialize using DNA.
    assert pg.materialize(
      hyper_dict, pg.DNA([0, 0.5])) == pg.Dict(x='a', y=0.5)

    # Materialize using key value pairs.
    # See `pg.DNA.from_dict` for more details.
    assert pg.materialize(
      hyper_dict, {'x': 0, 'y': 0.5}) == pg.Dict(x='a', y=0.5)

    # Partially materialize.
    v = pg.materialize(
      hyper_dict, pg.DNA(0), where=lambda x: isinstance(x, pg.hyper.OneOf))
    assert v == pg.Dict(x='a', y=pg.floatv(0.0, 1.0))
    ```

    Args:
      value: A (maybe) hyper value
      parameters: A DNA object, or a dict from each decision point's ID (key
        path) or name to its decision. A decision can be an integer as the
        candidate index for `geno.Choices`, a float for `geno.Float`, or a
        string - either a candidate's literal value or the
        '<selected_index>/<num_candidates>' form for `geno.Choices`. (See
        `pg.DNA.from_dict` for more details.)
      use_literal_values: If True, integer decisions in `parameters` are
        interpreted as literal values of the corresponding choice rather
        than candidate indices. String decisions are always accepted as
        either literal values or '<index>/<num_candidates>' forms.
      where: Function to filter hyper primitives. If None, all hyper
        primitives from `value` will be included in the encoding/decoding
        process. Otherwise only the hyper primitives on which 'where' returns
        True will be included. `where` can be useful to partition a search
        space into separate optimization processes. Please see the
        `ObjectTemplate` docstring for details.

    Returns:
      A materialized value.

    Raises:
      TypeError: if parameters is not a DNA or dict.
      ValueError: if parameters cannot be decoded.
    """
    t = template(value, where)
    if isinstance(parameters, dict):
        dna = geno.DNA.from_parameters(
            parameters=parameters,
            dna_spec=t.dna_spec(),
            use_literal_values=use_literal_values,
        )
    else:
        dna = parameters

    if not isinstance(dna, geno.DNA):
        raise TypeError(
            f"'parameters' must be a DNA or a dict of string to DNA values. "
            f'Encountered: {dna!r}.'
        )
    return t.decode(dna)

template

template(
    value: Any, where: Callable[[HyperPrimitive], bool] | None = None
) -> ObjectTemplate

Creates an object template from the input.

Example:

d = pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=pg.manyof(2, range(4)))
t = pg.template(d)

assert t.encode(pg.Dict(x='a', y=[0, 1])) == pg.DNA([0, [0, 1]])
assert t.decode(pg.DNA([0, [0, 1]])) == pg.Dict(x='a', y=[0, 1])

With a where filter, only the matching hyper primitives take part in encoding/decoding; the rest stay in place:

t = pg.template(d, where=lambda x: isinstance(x, pg.hyper.ManyOf))

assert t.encode(
    pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=[0, 1])) == pg.DNA([0, 1])
assert t.decode(pg.DNA([0, 1])) == pg.Dict(
    x=pg.oneof(['a', 'b', 'c']), y=[0, 1])

Parameters:

Name	Type	Description	Default
value	Any	A value based on which the template is created.	required
where	Callable[[HyperPrimitive], bool] | None	Function to filter hyper values. If None, all hyper primitives from value will be included in the encoding/decoding process. Otherwise only the hyper values on which 'where' returns True will be included. where can be useful to partition a search space into separate optimization processes. Please see the ObjectTemplate docstring for details.	None

Returns:

Type	Description
ObjectTemplate	A template object.

Source code in pygx/hyper/_object_template.py

def template(
    value: Any, where: Callable[[base.HyperPrimitive], bool] | None = None
) -> ObjectTemplate:
    """Creates an object template from the input.

    **Example:**

    ```python
    d = pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=pg.manyof(2, range(4)))
    t = pg.template(d)

    assert t.encode(pg.Dict(x='a', y=[0, 1])) == pg.DNA([0, [0, 1]])
    assert t.decode(pg.DNA([0, [0, 1]])) == pg.Dict(x='a', y=[0, 1])
    ```

    With a `where` filter, only the matching hyper primitives take part in
    encoding/decoding; the rest stay in place:

    ```python
    t = pg.template(d, where=lambda x: isinstance(x, pg.hyper.ManyOf))

    assert t.encode(
        pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=[0, 1])) == pg.DNA([0, 1])
    assert t.decode(pg.DNA([0, 1])) == pg.Dict(
        x=pg.oneof(['a', 'b', 'c']), y=[0, 1])
    ```

    Args:
      value: A value based on which the template is created.
      where: Function to filter hyper values. If None, all hyper primitives
        from `value` will be included in the encoding/decoding process.
        Otherwise only the hyper values on which 'where' returns True will be
        included. `where` can be useful to partition a search space into
        separate optimization processes. Please see the `ObjectTemplate`
        docstring for details.

    Returns:
      A template object.
    """
    return ObjectTemplate(value, compute_derived=True, where=where)

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2