def sweep_from_proto(
    msg: run_context_pb2.Sweep,
    sweep_transformer: Callable[[sweeps.SingleSweep], sweeps.SingleSweep] = lambda x: x,
) -> cirq.Sweep:
    """Creates a Sweep from a v2 protobuf message.

    Args:
        msg: Serialized sweep message.
        sweep_transformer: A function called on Linspace, Point, and ConstValue.

    """
    which = msg.WhichOneof('sweep')
    if which is None:
        return cirq.UnitSweep
    if which == 'sweep_function':
        factors = [sweep_from_proto(m, sweep_transformer) for m in msg.sweep_function.sweeps]
        func_type = msg.sweep_function.function_type
        if func_type == run_context_pb2.SweepFunction.PRODUCT:
            return cirq.Product(*factors)
        if func_type == run_context_pb2.SweepFunction.ZIP:
            return cirq.Zip(*factors)
        if func_type == run_context_pb2.SweepFunction.ZIP_LONGEST:
            return cirq.ZipLongest(*factors)
        if func_type == run_context_pb2.SweepFunction.CONCAT:
            return cirq.Concat(*factors)

        raise ValueError(f'invalid sweep function type: {func_type}')
    if which == 'single_sweep':
        key = msg.single_sweep.parameter_key
        metadata: DeviceParameter | Metadata | None
        if msg.single_sweep.HasField("parameter"):
            metadata = DeviceParameter(
                path=msg.single_sweep.parameter.path,
                idx=(
                    msg.single_sweep.parameter.idx
                    if msg.single_sweep.parameter.HasField("idx")
                    else None
                ),
                units=(
                    msg.single_sweep.parameter.units
                    if msg.single_sweep.parameter.HasField("units")
                    else None
                ),
            )
        elif msg.single_sweep.HasField("metadata"):
            metadata = metadata_from_proto(msg.single_sweep.metadata)
        else:
            metadata = None

        if msg.single_sweep.WhichOneof('sweep') == 'linspace':
            unit: float | tunits.Value = 1.0
            if msg.single_sweep.linspace.HasField('unit'):
                unit = tunits.Value.from_proto(msg.single_sweep.linspace.unit)
            return sweep_transformer(
                cirq.Linspace(
                    key=key,
                    start=msg.single_sweep.linspace.first_point * unit,  # type: ignore[arg-type]
                    stop=msg.single_sweep.linspace.last_point * unit,  # type: ignore[arg-type]
                    length=msg.single_sweep.linspace.num_points,
                    metadata=metadata,
                )
            )
        if msg.single_sweep.WhichOneof('sweep') == 'points':
            unit = 1.0
            if msg.single_sweep.points.HasField('unit'):
                unit = tunits.Value.from_proto(msg.single_sweep.points.unit)
            return sweep_transformer(
                cirq.Points(
                    key=key,
                    points=[p * unit for p in msg.single_sweep.points.points],
                    metadata=metadata,
                )
            )
        if msg.single_sweep.WhichOneof('sweep') == 'const_value':
            return sweep_transformer(
                cirq.Points(
                    key=key,
                    points=[_recover_sweep_const(msg.single_sweep.const_value)],
                    metadata=metadata,
                )
            )

        raise ValueError(f'single sweep type not set: {msg}')

    raise ValueError(f'sweep type not set: {msg}')  # pragma: no cover
Creates a Sweep from a v2 protobuf message.

    Args:
        msg: Serialized sweep message.
        sweep_transformer: A function called on Linspace, Point, and ConstValue.

    
cirq-google/cirq_google/api/v2/sweeps.py