`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 an symbolic object that is 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).

.. graphviz:: :align: center

digraph hypers {
  node [shape="box"];
  edge [arrowtail="empty" arrowhead="none" dir="back" style="dashed"];
  hyper [label="HyperValue" href="hyper_value.html"];
  template [label="ObjectTemplate" href="object_template.html"];
  primitive [label="HyperPrimitive" href="hyper_primitive.html"];
  choices [label="Choices" href="choices.html"];
  oneof [label="OneOf" href="oneof_class.html"];
  manyof [label="ManyOf" href="manyof_class.html"];
  float [label="Float" href="float.html"];
  custom [label="CustomHyper" href="custom_hyper.html"];
  evolve [label="Evolvable" href="evolvable.html"];
  hyper -> template;
  hyper -> primitive;
  primitive -> choices;
  choices -> oneof;
  choices -> manyof;
  primitive -> float;
  primitive -> custom;
  custom -> evolve;
}

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

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 correspond to a decision point (pygx.geno.DecisionPoint) in the algorithm's view.

Child classes:

pygx.hyper.Choices

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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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 object (see geno.py). 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:

.. graphviz::

digraph relationship { template [label="ObjectTemplate" href="object_template.html"]; primitive [label="HyperPrimitive" href="hyper_primitive.html"]; choices [label="OneOf/ManyOf" href="choices.html"]; float [label="Float" href="float_class.html"]; custom [label="CustomHyper" href="custom_hyper.html"]; template -> primitive [label="elements (1:)"]; primitive -> choices [dir="back" arrowtail="empty" style="dashed"]; primitive -> float [dir="back" arrowtail="empty" style="dashed"]; primitive -> custom [dir="back" arrowtail="empty" style="dashed"]; choices -> template [label="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 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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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.

Example::

# DNA of an object containing a single OneOf.
# {'a': 1} => DNA(0)
{
   'a': one_of([1, 2])
}


# DNA of an object containing multiple OneOfs.
# {'b': 1, 'c': bar} => DNA([0, 1])
{
   'b': pg.oneof([1, 2]),
   'c': pg.oneof(['foo', 'bar'])
}

# DNA of an object containing conditional space.
# {'a': {'b': 1} => DNA(0, 0, 0)])
# {'a': {'b': [4, 7]} => DNA(1, [(0, 1), 2])
# {'a': {'b': 'bar'} => DNA(2)
{
   'a': {
      'b': pg.oneof([
        pg.oneof([
          pg.oneof([1, 2]),
          pg.oneof(3, 4)]),
          pg.manyof(2, [
            pg.oneof([4, 5]),
            6,
            7
          ]),
        ]),
        'bar',
      ])
   }
}

Parameters:

Name	Type	Description	Default
`value`	`list[Any]`	A list of value 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.

    Example::

        # DNA of an object containing a single OneOf.
        # {'a': 1} => DNA(0)
        {
           'a': one_of([1, 2])
        }


        # DNA of an object containing multiple OneOfs.
        # {'b': 1, 'c': bar} => DNA([0, 1])
        {
           'b': pg.oneof([1, 2]),
           'c': pg.oneof(['foo', 'bar'])
        }

        # DNA of an object containing conditional space.
        # {'a': {'b': 1} => DNA(0, 0, 0)])
        # {'a': {'b': [4, 7]} => DNA(1, [(0, 1), 2])
        # {'a': {'b': 'bar'} => DNA(2)
        {
           'a': {
              'b': pg.oneof([
                pg.oneof([
                  pg.oneof([1, 2]),
                  pg.oneof(3, 4)]),
                  pg.manyof(2, [
                    pg.oneof([4, 5]),
                    6,
                    7
                  ]),
                ]),
                'bar',
              ])
           }
        }

    Args:
      value: A list of value 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.sym_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.sym_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.sym_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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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.

To use user-defined hyper primitive is simple, the user should:

1) Subclass CustomHyper and implements the pygx.hyper.CustomHyper.custom_decode method. It's optional to implement the pygx.hyper.CustomHyper.custom_encode method, which is only necessary when the user want to encoder 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(): 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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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

Subclasses should override this method to support pg.Sweeping.

Source code in pygx/hyper/_custom.py

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

random_dna ¶

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

Subclasses should override this method to support 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:
    """Subclasses should override this method to support 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 of value that references to other values in 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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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.sym_parent
        while parent is not None and not reference_path.exists(parent):
            parent = getattr(parent, 'sym_parent', None)
        if parent is None:
            raise ValueError(
                f"Cannot resolve '{reference_path}': parent not found."
            )
        resolved_paths.append((parent, parent.sym_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::

import pygx as pg

# 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 the 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): # Call to example returns a context manager # under which the program 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 `colelct` 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 `colelct` 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

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.

Example::

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

# Will print 1 + 4 = 5. Meanwhile 2 hyper primitives will be 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.

    Example::

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

      # Will print 1 + 4 = 5. Meanwhile 2 hyper primitives will be 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())

    if (
        name in self._name_to_hyper
        and hyper_primitive != self._name_to_hyper[name]
    ):
        raise ValueError(
            f'Found different hyper primitives under the same name {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.

Example::

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

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

with context.apply([0, 1]):
  # Will print 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 currrent 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.

      Example::

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

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

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

    Args:
      decisions: A DNA or a list of numbers or strings as decisions for currrent
        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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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,
)

Internal class that encapsulates the information for a mutation point.

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 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 `cls.__sym_options__.frozen`.
      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:
      KeyError: When required key(s) are missing.
      ValueError: When value(s) are not acceptable by their value spec.
    """
    # 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 (varkw) 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 `varkw='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_sym_mode=cls.__sym_options__.sym,
    )
    self._sym_attributes.sym_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.sym_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.sym_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.sym_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 object with some parts placeheld by hyper primitives. For example::

x = A(a=0, b=pg.oneof(['foo', 'bar']), c=pg.manyof(2, [1, 2, 3, 4, 5, 6]), d=pg.floatv(0.1, 0.5), e=pg.oneof([ { 'f': pg.oneof([True, False]), } { 'g': pg.manyof(2, [B(), C(), D()], distinct=False), 'h': pg.manyof(2, [0, 1, 2], sorted=True), } ]) }) t = pg.template(x)

In this example, the root template have 4 children hyper primitives associated with keys 'b', 'c', 'd' and 'e', while the hyper primitive 'e' have 3 children associated with keys 'f', 'g' and 'h', creating a conditional search space.

Thus the DNA shape is determined by the definition of template, described by geno.DNASpec. In this case, the DNA spec of this template looks like::

pg.geno.space([ pg.geno.oneof([ # Spec for 'b'. pg.geno.constant(), # A constant template for 'foo'. pg.geno.constant(), # A constant template for 'bar'. ]), pg.geno.manyof([ # Spec for 'c'. pg.geno.constant(), # A constant template for 1. pg.geno.constant(), # A constant template for 2. pg.geno.constant(), # A constant template for 3. pg.geno.constant(), # A constant template for 4. pg.geno.constant(), # A constant template for 5. pg.geno.constant(), # A constant template for 6. ]), pg.geno.floatv(0.1, 0.5), # Spec for 'd'. pg.geno.oneof([ # Spec for 'e'. pg.geno.space([ pg.geno.oneof([ # Spec for 'f'. pg.geno.constant(), # A constant template for True. pg.geno.constant(), # A constant template for False. ]) ]), pg.geno.space([ pg.geno.manyof(2, [ # Spec for 'g'. pg.geno.constant(), # A constant template for B(). pg.geno.constant(), # A constant template for C(). pg.geno.constant(), # A constant template for D(). ], distinct=False) # choices of the same value can # be selected multiple times. pg.geno.manyof(2, [ # Spec for 'h'. pg.geno.constant(), # A constant template for 0. pg.geno.constant(), # A constant template for 1. pg.geno.constant(), # A constant template for 2. ], sorted=True) # acceptable choices needs to be sorted, # which enables using choices as set (of # possibly repeated values). ]) ])

It may generate DNA as the following

DNA([0, [0, 2], 0.1, (0, 0)])

A template can also work only on a subset of hyper primitives from the input value through the where function. This is useful to partition a search space into parts for separate optimization.

For example::

t = pg.hyper.ObjectTemplate( A(a=pg.oneof([1, 2]), b=pg.oneof([3, 4])), where=lambda e: e.root_path == 'a') assert t.dna_spec() == pg.geno.space([ pg.geno.oneof(location='a', candidates=[ pg.geno.constant(), # For a=1 pg.geno.constant(), # For a=2 ], literal_values=['(0/2) 1', '(½) 2']) ]) 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.oneof. 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::

      # 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.oneof.
      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 = []

    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}'.")
        if isinstance(template_value, base.HyperValue) and (
            not self._where or self._where(template_value)  # pyright: ignore[reportArgumentType]
        ):
            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 raise Exception.

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 raise Exception."""
    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(1)])

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

.. note::

Under symbolic mode (by default), pg.manyof returns a pg.hyper.ManyOf object. Under dynamic evaluation mode, which is called under the context of pygx.hyper.DynamicEvaluationContext.collect or pygx.hyper.DynamicEvaluationContext.apply, it evaluates to a concrete candidate value.

To use conditional search space in dynamic evaluate 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 `choice_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 can be used to identify a decision point in the search space. This is needed when the code to instantiate the same hyper primitive may be called multiple times under a `pg.DynamicEvaluationContext.collect` context or a `pg.DynamicEvaluationContext.apply` context.	`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 `Choices`.
`Any`	In dynamic evaluate mode, this function returns a list of items in
`Any`	`candidates`.
`Any`	If evaluated under a `pg.DynamicEvaluationContext.apply` scope,
`Any`	this function will return a list of selected candidates.
`Any`	If evaluated under a `pg.DynamicEvaluationContext.collect`
`Any`	scope, it will return a list of the first valid combination from the
`Any`	`candidates`. For example:: # Evaluates to [0, 1, 2]. manyof(3, range(5)) # Evaluates to [0, 0, 0]. manyof(3, range(5), distinct=False)

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::

      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(1)])

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

    .. note::

      Under symbolic mode (by default), `pg.manyof` returns a ``pg.hyper.ManyOf``
      object. Under dynamic evaluation mode, which is called under the context of
      [`pygx.hyper.DynamicEvaluationContext.collect`][pygx.hyper.DynamicEvaluationContext.collect] or
      [`pygx.hyper.DynamicEvaluationContext.apply`][pygx.hyper.DynamicEvaluationContext.apply], it evaluates to
      a concrete candidate value.

      To use conditional search space in dynamic evaluate 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.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 `choice_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 can be used to identify a decision point in the search
        space. This is needed when the code to instantiate the same hyper
        primitive may be called multiple times under a
        `pg.DynamicEvaluationContext.collect` context or a
        `pg.DynamicEvaluationContext.apply` context.
      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 `Choices`.
      In dynamic evaluate mode, this function returns a list of items in
      `candidates`.
      If evaluated under a `pg.DynamicEvaluationContext.apply` scope,
      this function will return a list of selected candidates.
      If evaluated under a `pg.DynamicEvaluationContext.collect`
      scope, it will return a list of the first valid combination from the
      `candidates`. For example::

          # Evaluates to [0, 1, 2].
          manyof(3, range(5))

          # Evaluates to [0, 0, 0].
          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(1)])

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

See also:

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

.. note::

Under symbolic mode (by default), pg.oneof returns a pg.hyper.OneOf object. Under dynamic evaluation mode, which is called under the context of pygx.hyper.DynamicEvaluationContext.collect or pygx.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 can be used to identify a decision point in the search space. This is needed when the code to instantiate the same hyper primitive may be called multiple times under a `pg.DynamicEvaluationContext.collect` context or under a `pg.DynamicEvaluationContext.apply` context.	`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 `ChoiceValue`.
`Any`	In dynamic evaluation mode, this function returns one of the items in
`Any`	`candidates`.
`Any`	If evaluated under a `pg.DynamicEvaluationContext.apply` scope,
`Any`	this function will return the selected candidate.
`Any`	If evaluated under a `pg.DynamicEvaluationContext.collect`
`Any`	scope, it will return 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::

      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(1)])

      # A hierarchical categorical choice:
      v2 = pg.oneof([
          'foo',
          'bar',
          A(pg.oneof([1, 2, 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 (by default), `pg.oneof` returns a ``pg.hyper.OneOf``
      object. Under dynamic evaluation mode, which is called under the context of
      [`pygx.hyper.DynamicEvaluationContext.collect`][pygx.hyper.DynamicEvaluationContext.collect] or
      [`pygx.hyper.DynamicEvaluationContext.apply`][pygx.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')

    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 can be used to identify a decision point in the search
        space. This is needed when the code to instantiate the same hyper
        primitive may be called multiple times under a
        `pg.DynamicEvaluationContext.collect` context or under a
        `pg.DynamicEvaluationContext.apply` context.
      hints: An optional value which acts as a hint for the controller.

    Returns:
      In symbolic mode, this function returns a `ChoiceValue`.
      In dynamic evaluation mode, this function returns one of the items in
      `candidates`.
      If evaluated under a `pg.DynamicEvaluationContext.apply` scope,
      this function will return the selected candidate.
      If evaluated under a `pg.DynamicEvaluationContext.collect`
      scope, it will return 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

Permuatation 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(1)])

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

.. note::

Under symbolic mode (by default), pg.manyof returns a pg.hyper.ManyOf object. Under dynamic evaluate mode, which is called under the context of pygx.hyper.DynamicEvaluationContext.collect or pygx.hyper.DynamicEvaluationContext.apply, it evaluates to a concrete candidate value.

To use conditional search space in dynamic evaluate 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 can be used to identify a decision point in the search space. This is needed when the code to instantiate the same hyper primitive may be called multiple times under a `pg.DynamicEvaluationContext.collect` context or a `pg.DynamicEvaluationContext.apply` context.	`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 `Choices`.
`Any`	In dynamic evaluate mode, this function returns a permutation from
`Any`	`candidates`.
`Any`	If evaluated under an `pg.DynamicEvaluationContext.apply` scope,
`Any`	this function will return a permutation of candidates based on controller
`Any`	decisions.
`Any`	If evaluated under a `pg.DynamicEvaluationContext.collect`
`Any`	scope, it will return the first valid permutation.
`Any`	For example:: Evaluates to [0, 1, 2, 3, 4].¶ permutate(range(5), name='numbers')

Source code in pygx/hyper/_categorical.py

def permutate(
    candidates: Iterable[Any],
    name: str | None = None,
    hints: Any | None = None,
) -> Any:
    """Permuatation 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(1)])

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

    .. note::

      Under symbolic mode (by default), `pg.manyof` returns a ``pg.hyper.ManyOf``
      object. Under dynamic evaluate mode, which is called under the context of
      [`pygx.hyper.DynamicEvaluationContext.collect`][pygx.hyper.DynamicEvaluationContext.collect] or
      [`pygx.hyper.DynamicEvaluationContext.apply`][pygx.hyper.DynamicEvaluationContext.apply], it evaluates to
      a concrete candidate value.

      To use conditional search space in dynamic evaluate 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.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 can be used to identify a decision point in the search
        space. This is needed when the code to instantiate the same hyper
        primitive may be called multiple times under a
        `pg.DynamicEvaluationContext.collect` context or a
        `pg.DynamicEvaluationContext.apply` context.
      hints: An optional value which acts as a hint for the controller.

    Returns:
      In symbolic mode, this function returns a `Choices`.
      In dynamic evaluate mode, this function returns a permutation from
      `candidates`.
      If evaluated under an `pg.DynamicEvaluationContext.apply` scope,
      this function will return a permutation of candidates based on controller
      decisions.
      If evaluated under a `pg.DynamicEvaluationContext.collect`
      scope, it will return the first valid permutation.
      For example::

        # Evaluates to [0, 1, 2, 3, 4].
        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

Create a referenced value from a referenced path.

Source code in pygx/hyper/_derived.py

def reference(reference_path: str) -> ValueReference:
    """Create a referenced value from a referenced path."""
    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.

Example::

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

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

Please see pygx.DynamicEvaluationContext.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 such as oneof, manyof, floatv, and etc. 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.

    Example::

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

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

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

    Args:
      evaluate_fn: A callable object that evaluates a hyper value such as
        oneof, manyof, floatv, and etc. 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.

See examples in pygx.hyper.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`	An 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.

    See examples in [`pygx.hyper.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:
        An 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::

@pg.symbolize @dataclasses.dataclass class Foo: x: int y: int

@pg.symbolize @dataclasses.dataclass class Bar: a: int b: int

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

v = pg.evolve(Foo(1, 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`. See `pg.composing.SeenObjectReplacer` as an example.	required
`weights`	`None \| Callable[[MutationType, KeyPath, Any, Symbolic], float]`	An optional callable object that returns the unnormalized (e.g. the sum of all 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::

      @pg.symbolize
      @dataclasses.dataclass
      class Foo:
        x: int
        y: int

      @pg.symbolize
      @dataclasses.dataclass
      class Bar:
        a: int
        b: int

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

      v = pg.evolve(Foo(1, 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`. See
        `pg.composing.SeenObjectReplacer` as an example.
      weights: An optional callable object that returns the unnormalized (e.g.
        the sum of all 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. for d, feedback in pg.iter( hyper_dict, 10, pg.algo.evolution.regularized_evolution(pg.algo.evolution.mutators.Uniform())): 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 distinguishes from pg.sample in that it's designed for simple in-process iteration, which is handy for quickly generating examples from algorithms without maintaining trail 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.	required
`num_examples`	`int \| None`	An optional integer as the max number of examples to propose. If None, propose will return an iterator of infinite examples.	`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 'Template' docstr 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::

      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.
      for d, feedback in pg.iter(
          hyper_dict, 10,
          pg.algo.evolution.regularized_evolution(pg.algo.evolution.mutators.Uniform())):
        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`` distinguishes from `pg.sample` in that it's designed
    for simple in-process iteration, which is handy for quickly generating
    examples from algorithms without maintaining trail 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.
      num_examples: An optional integer as the max number of examples to
          propose. If None, propose will return an iterator of infinite examples.
      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 'Template' docstr 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 propose. If None, propose will return an iterator that 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 'Template' docstr 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::

      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 propose. If None,
        propose will return an iterator that 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 'Template' docstr 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 (by default), pg.floatv returns a pg.hyper.Float object. Under dynamic evaluate mode, which is called under the context of pygx.hyper.DynamicEvaluationContext.collect or pygx.hyper.DynamicEvaluationContext.apply, it evaluates to a concrete candidate 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 can be used to identify a decision point in the search space. This is needed when the code to instantiate the same hyper primitive may be called multiple times under a `pg.DynamicEvaluationContext.collect` context or a `pg.DynamicEvaluationContext.apply` context.	`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`.
`Any`	In dynamic evaluate mode, this function returns a float value that is no
`Any`	less than the `min_value` and no greater than the `max_value`.
`Any`	If evaluated under an `pg.DynamicEvaluationContext.apply` scope,
`Any`	this function will return a chosen float value from the controller
`Any`	decisions.
`Any`	If evaluated under a `pg.DynamicEvaluationContext.collect`
`Any`	scope, it will return `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::

      # 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 (by default), `pg.floatv` returns a ``pg.hyper.Float``
      object. Under dynamic evaluate mode, which is called under the context of
      [`pygx.hyper.DynamicEvaluationContext.collect`][pygx.hyper.DynamicEvaluationContext.collect] or
      [`pygx.hyper.DynamicEvaluationContext.apply`][pygx.hyper.DynamicEvaluationContext.apply], it evaluates to
      a concrete candidate 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 can be used to identify a decision point in the search
        space. This is needed when the code to instantiate the same hyper
        primitive may be called multiple times under a
        `pg.DynamicEvaluationContext.collect` context or a
        `pg.DynamicEvaluationContext.apply` context.
      hints: An optional value which acts as a hint for the controller.

    Returns:
      In symbolic mode, this function returns a `Float`.
      In dynamic evaluate mode, this function returns a float value that is no
      less than the `min_value` and no greater than the `max_value`.
      If evaluated under an `pg.DynamicEvaluationContext.apply` scope,
      this function will return a chosen float value from the controller
      decisions.
      If evaluated under a `pg.DynamicEvaluationContext.collect`
      scope, it will return `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:

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 'Template' docstr 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::

      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 'Template' docstr 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 = True,
    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 usign 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 of string (key path) to a string (in format of '/' for `geno.Choices`, or '' for `geno.Float`), or their literal values when `use_literal_values` is set to True.	required
`use_literal_values`	`bool`	Applicable when `parameters` is a dict. If True, the values in the dict will be from `geno.Choices.literal_values` for `geno.Choices`.	`True`
`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 'Template' docstr 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 = True,
    where: Callable[[base.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 usign 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 of string (key path) to a
        string (in format of '<selected_index>/<num_choices>' for
        `geno.Choices`, or '<float_value>' for `geno.Float`), or their literal
        values when `use_literal_values` is set to True.
      use_literal_values: Applicable when `parameters` is a dict. If True, the
        values in the dict will be from `geno.Choices.literal_values` for
        `geno.Choices`.
      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 'Template' docstr 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.dna_spec() == pg.geno.space([ pg.geno.oneof([ pg.geno.constant(), pg.geno.constant(), pg.geno.constant(), ], location='x'), pg.geno.manyof([ pg.geno.constant(), pg.geno.constant(), pg.geno.constant(), pg.geno.constant(), ], location='y') ])

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

t = pg.template(d, where=lambda x: isinstance(x, pg.hyper.ManyOf)) assert t.dna_spec() == pg.geno.space([ pg.geno.manyof([ pg.geno.constant(), pg.geno.constant(), pg.geno.constant(), pg.geno.constant(), ], location='y') ]) assert t.encode(pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=0)) == pg.DNA(0) assert t.decode(pg.DNA(0)) == pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=0)

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 'ObjectTemplate' docstr 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::

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

      assert t.dna_spec() == pg.geno.space([
          pg.geno.oneof([
              pg.geno.constant(),
              pg.geno.constant(),
              pg.geno.constant(),
          ], location='x'),
          pg.geno.manyof([
              pg.geno.constant(),
              pg.geno.constant(),
              pg.geno.constant(),
              pg.geno.constant(),
          ], location='y')
      ])

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

      t = pg.template(d, where=lambda x: isinstance(x, pg.hyper.ManyOf))
       assert t.dna_spec() == pg.geno.space([
          pg.geno.manyof([
              pg.geno.constant(),
              pg.geno.constant(),
              pg.geno.constant(),
              pg.geno.constant(),
          ], location='y')
      ])
      assert t.encode(pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=0)) == pg.DNA(0)
      assert t.decode(pg.DNA(0)) == pg.Dict(x=pg.oneof(['a', 'b', 'c']), y=0)

    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 'ObjectTemplate' docstr 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

pygx.hyper¶

hyper ¶

HyperPrimitive ¶

HyperValue ¶

dna property ¶

set_dna ¶

decode ¶

encode abstractmethod ¶

dna_spec abstractmethod ¶

Choices ¶

candidate_templates property ¶

is_leaf property ¶

on_sym_bound ¶

dna_spec ¶

encode ¶

ManyOf ¶

custom_apply ¶

OneOf ¶

on_sym_bound ¶

encode ¶

custom_apply ¶

CustomHyper ¶

custom_decode abstractmethod ¶

encode ¶

custom_encode ¶

dna_spec ¶

first_dna ¶

next_dna ¶

random_dna ¶

custom_apply ¶

DerivedValue ¶

derive abstractmethod ¶

resolve ¶

ValueReference ¶

on_sym_bound ¶

derive ¶

custom_apply ¶

DynamicEvaluationContext ¶

per_thread property ¶

dna_spec property ¶

is_external property ¶

hyper_dict property ¶

collect ¶

add_decision_point ¶

apply ¶

evaluate ¶

Evolvable ¶

mutation_points_and_weights ¶

first_dna ¶

random_dna ¶

mutate ¶

MutationPoint dataclass ¶

MutationType ¶

Float ¶

on_sym_bound ¶

dna_spec ¶

encode ¶

custom_apply ¶

is_leaf ¶

ObjectTemplate ¶

root_path property writable ¶

value property ¶

hyper_primitives property ¶

is_constant property ¶

dna_spec ¶

encode ¶

try_encode ¶

format ¶

custom_apply ¶

manyof ¶

oneof ¶

permutate ¶

Evaluates to [0, 1, 2, 3, 4].¶

reference ¶

dynamic_evaluate ¶

trace ¶

evolve ¶

iterate ¶

random_sample ¶

floatv ¶

`pygx.hyper`¶

dna `property` ¶

encode `abstractmethod` ¶

dna_spec `abstractmethod` ¶

candidate_templates `property` ¶

is_leaf `property` ¶

custom_decode `abstractmethod` ¶

derive `abstractmethod` ¶

per_thread `property` ¶

dna_spec `property` ¶

is_external `property` ¶

hyper_dict `property` ¶

MutationPoint `dataclass` ¶

root_path `property` `writable` ¶

value `property` ¶

hyper_primitives `property` ¶

is_constant `property` ¶