NeMo-Run/test/dummy_factory.py at main · sudostock/NeMo-Run · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
# 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
#
# http://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.

from dataclasses import dataclass
from typing import List, ForwardRef

import nemo_run as run


@dataclass
class DummyModel:
    hidden: int = 100
    activation: str = "relu"


class DummyTrainer:
    def __init__(self, num_epochs: int = 10):
        self.num_epochs = num_epochs

    def __hash__(self) -> int:
        return hash((self.num_epochs))

    def __eq__(self, value: object) -> bool:
        return isinstance(value, DummyTrainer) and self.num_epochs == value.num_epochs


@dataclass
class NestedModel:
    dummy: DummyModel


@dataclass(kw_only=True)
class DummyPlugin(run.Plugin):
    some_arg: int = 10


@dataclass(kw_only=True)
class AnotherPlugin(run.Plugin):
    another_arg: int = 10


@run.cli.factory
@run.autoconvert
def dummy_factory_for_entrypoint() -> DummyModel:
    return DummyModel(hidden=1000)


@run.cli.factory
def dummy_model_config() -> run.Config[DummyModel]:
    return run.Config(DummyModel, hidden=2000, activation="tanh")


@run.cli.factory
@run.autoconvert
def my_dummy_model(hidden=2000) -> DummyModel:
    return DummyModel(hidden=hidden, activation="tanh")


@run.cli.entrypoint(namespace="dummy", skip_confirmation=True)
def dummy_entrypoint(dummy: DummyModel):
    NestedModel(dummy=dummy)


@run.cli.factory(target=dummy_entrypoint)
def dummy_recipe() -> run.Partial[dummy_entrypoint]:
    return run.Partial(dummy_entrypoint, dummy=dummy_model_config())


@run.cli.factory
@run.autoconvert
def local_executor() -> run.Executor:
    return run.LocalExecutor()


@run.cli.factory
@run.autoconvert
def dummy_plugin(some_arg: int = 20) -> run.Plugin:
    return DummyPlugin(some_arg=some_arg)


@run.cli.factory
@run.autoconvert
def plugin_list(arg: int = 20) -> List[run.Plugin]:
    return [
        dummy_plugin(arg),
        AnotherPlugin(another_arg=arg),
    ]


@run.cli.factory
@run.autoconvert
def tokenizer_spec() -> ForwardRef("TokenizerSpec"):
    return DummyModel(hidden=1000)


def dummy_train(dummy_model: DummyModel, dummy_trainer: DummyTrainer): ...


if __name__ == "__main__":
    run.cli.main(dummy_entrypoint)