API

  • SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.

  • SPDX-License-Identifier: Apache-2.0

  • Licensed under the Apache License, Version 2.0 (the “License”);

  • you may not use this file except in compliance with the License.

  • You may obtain a copy of the License at

  • https://www.apache.org/licenses/LICENSE-2.0

  • Unless required by applicable law or agreed to in writing, software

  • distributed under the License is distributed on an “AS IS” BASIS,

  • WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

  • See the License for the specific language governing permissions and

  • limitations under the License.

class omni.flux.utils.common.Converter(key: str, claim_func: typing.Callable[[typing.Any], bool], serialize_hook: typing.Callable[[omni.flux.utils.common.serialize.T], typing.Union[dict[str, 'JSON'], list['JSON'], str, int, float, bool, None, omni.flux.utils.common.serialize.Primitive]] = <function Converter.<lambda>>, deserialize_hook: typing.Callable[[typing.Union[dict[str, 'JSON'], list['JSON'], str, int, float, bool, None, omni.flux.utils.common.serialize.Primitive]], omni.flux.utils.common.serialize.T] = <function Converter.<lambda>>)

Bases: Generic[omni.flux.utils.common.serialize.T]

A dataclass which holds methods used to convert data from one format to another.

__init__(key: str, claim_func: typing.Callable[[typing.Any], bool], serialize_hook: typing.Callable[[omni.flux.utils.common.serialize.T], typing.Union[dict[str, 'JSON'], list['JSON'], str, int, float, bool, None, omni.flux.utils.common.serialize.Primitive]] = <function Converter.<lambda>>, deserialize_hook: typing.Callable[[typing.Union[dict[str, 'JSON'], list['JSON'], str, int, float, bool, None, omni.flux.utils.common.serialize.Primitive]], omni.flux.utils.common.serialize.T] = <function Converter.<lambda>>) → None
claim_func: Callable[[Any], bool]
deserialize_hook()
key: str
serialize_hook()
class omni.flux.utils.common.Event(*args, copy: bool = False, **kwargs)

Bases: set

A list of callable objects. Calling an instance of this will cause a call to each item in the list in ascending order by index.

__init__(*args, copy: bool = False, **kwargs)

Init of a set function

Parameters

  • *args – any args

  • copy – if True, it will execute callback(s) from a copy of the set. Why? It can happen that we have a circular pattern where an event execute a callback that will re-create/delete the event itself. For example, imagine a button that sets an event when we click on it. And the event is that this is re-creating the button itself (and the event). As a result, you could have an error like RuntimeError: Set changed size during iteration. We don’t set this to True by default to force the dev to be aware of the pattern he is doing.

  • **kwargs – any kwargs

class omni.flux.utils.common.EventSubscription(event, function)

Bases: object

Event subscription.

_Event has callback while this object exists.

__init__(event, function)

Save the function, the event, and add the function to the event.

class omni.flux.utils.common.Serializer

Bases: Generic[omni.flux.utils.common.serialize.T]

A utility class that provides methods to serialize and deserialize objects using custom registered hooks.

Examples

There are a few ways to configure the serializer to handle custom types.

[Option 1 : Direct Converter instance] >>> import datetime … >>> serializer = Serializer() … >>> serializer.register_converter( … Converter( … key=”datetime.datetime”, … claim_func=lambda x: isinstance(x, datetime.datetime), … serialize_hook=lambda dt: dt.isoformat(), … deserialize_hook=datetime.datetime.fromisoformat, … ) … )

[Option 2 : Register serialize/deserialize hook decorators] >>> import datetime … >>> serializer = Serializer() … >>> @serializer.register_serialize_hook(datetime.datetime) >>> def serialize_datetime(dt): … return dt.isoformat() … >>> @serializer.register_deserialize_hook(datetime.datetime) >>> def deserialize_datetime(s): … return datetime.datetime.fromisoformat(s)

__init__()
deserialize(data: dict[str, 'JSON'] | list['JSON'] | str | int | float | bool | None) → omni.flux.utils.common.serialize.T

Deserialize json data back into native types.

dumps(obj: omni.flux.utils.common.serialize.T) → dict[str, 'JSON'] | list['JSON'] | str | int | float | bool | None

Wrapper for json.dumps which will serialize any complex types using the registered converters.

loads(s: dict[str, 'JSON'] | list['JSON'] | str | int | float | bool | None) → omni.flux.utils.common.serialize.T

Wrapper for json.loads which will deserialize into native types using the registered converters.

register_converter(converter: omni.flux.utils.common.serialize.Converter)

Add a converter to the registry of available converters.

register_deserialize_hook(claim_func_or_type: Union[Type, Callable[[Any], bool]], key: Optional[str] = None)

Decorator used to register a deserialize hook for specific type(s).

register_serialize_hook(claim_func_or_type: Union[Type, Callable[[Any], bool]], key: Optional[str] = None)

Decorator used to register a serialize hook for specific type(s).

serialize(obj: omni.flux.utils.common.serialize.T) → dict[str, 'JSON'] | list['JSON'] | str | int | float | bool | None

Serialize obj to something json friendly.

This method is called recursively for nested types such as tuple, list, set, and dict.

to_primitive(obj: omni.flux.utils.common.serialize.T) → Union[omni.flux.utils.common.serialize.Primitive[omni.flux.utils.common.serialize.T], omni.flux.utils.common.serialize.T]

Check for any registered converters for obj, if one exists, then call its serialize hook to convert it into a primitive type.

omni.flux.utils.common.async_wrap(func) → Callable

Wrap a function into an async executor

omni.flux.utils.common.reset_default_attrs(obj)