Skip to content

pybrook.__main__

PyBrook CLI entrypoint.

ModelChangeEventHandler

Bases: FileSystemEventHandler

Handles model hot-reloading.

Source code in pybrook/__main__.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class ModelChangeEventHandler(FileSystemEventHandler):
    """
    Handles model hot-reloading.
    """

    def __init__(self, brook: PyBrook):
        """

        Args:
            brook: A PyBrook instance.
        """
        self.brook = brook
        self.modified = False

    def on_modified(self, event: Union[DirModifiedEvent, FileModifiedEvent]):
        """

        Args:
            event: Event representing file/directory modification.
        """
        logger.info("File change detected, reloading...")
        self.modified = True
        self.brook.terminate()

brook = brook instance-attribute

modified = False instance-attribute

__init__(brook)

Parameters:

Name Type Description Default
brook PyBrook

A PyBrook instance.

 required
Source code in pybrook/__main__.py 
54
55
56
57
58
59
60
61
def __init__(self, brook: PyBrook):
    """

    Args:
        brook: A PyBrook instance.
    """
    self.brook = brook
    self.modified = False

on_modified(event)

Parameters:

Name Type Description Default
event Union[DirModifiedEvent, FileModifiedEvent]

Event representing file/directory modification.

 required
Source code in pybrook/__main__.py 
63
64
65
66
67
68
69
70
71
def on_modified(self, event: Union[DirModifiedEvent, FileModifiedEvent]):
    """

    Args:
        event: Event representing file/directory modification.
    """
    logger.info("File change detected, reloading...")
    self.modified = True
    self.brook.terminate()

add_consumer_args(parser, consumers)

Parameters:

Name Type Description Default
parser ArgumentParser

The main argument parser.

 required
consumers list[BaseStreamConsumer]

List of consumers to generate CLI options for.

 required

Returns:

Type Description
dict[str, ConsumerConfig]

A dictionary of consumer configs filled with defaults.
dict[str, ConsumerConfig]

Consumer group names are used as keys.
Source code in pybrook/__main__.py 
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
def add_consumer_args(
    parser: argparse.ArgumentParser, consumers: list[BaseStreamConsumer]
) -> dict[str, ConsumerConfig]:
    """

    Args:
        parser: The main argument parser.
        consumers: List of consumers to generate CLI options for.

    Returns:
        A dictionary of consumer configs filled with defaults.
        Consumer group names are used as keys.

    """
    workers_config = {}
    for c in consumers:
        consumer_config = ConsumerConfig()
        with suppress(ArgumentError):
            parser.add_argument(
                f"--{c.consumer_group_name}-workers",
                type=int,
                help="(default: %(default)s)",  # noqa: WPS323
                default=consumer_config.workers,
            )
        workers_config[c.consumer_group_name] = consumer_config
    return workers_config

main()

CLI Entrypoint.

Starts PyBrook workers.

Examples:

 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
```bash
❯ pybrook pybrook.examples.demo:brook --help
usage: pybrook [-h]
       [--location-report:dr-workers LOCATION_REPORT:DR_WORKERS]
       [--direction-report:dr-workers DIRECTION_REPORT:DR_WORKERS]
       [--brigade-report:dr-workers BRIGADE_REPORT:DR_WORKERS]
       [--direction:dr-workers DIRECTION:DR_WORKERS]
       [--direction:fg-workers DIRECTION:FG_WORKERS]
       APP

positional arguments:
  APP

options:
  -h, --help
  --location-report:dr-workers LOCATION_REPORT:DR_WORKERS
                        (default: 4)
  --direction-report:dr-workers DIRECTION_REPORT:DR_WORKERS
                        (default: 4)
  --brigade-report:dr-workers BRIGADE_REPORT:DR_WORKERS
                        (default: 4)
  --direction:dr-workers DIRECTION:DR_WORKERS
                        (default: 4)
  --direction:fg-workers DIRECTION:FG_WORKERS
                        (default: 4)

```
Source code in pybrook/__main__.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def main() -> None:
    """
    CLI Entrypoint.

    Starts PyBrook workers.

    Examples:

        ```bash
        ❯ pybrook pybrook.examples.demo:brook --help
        usage: pybrook [-h]
               [--location-report:dr-workers LOCATION_REPORT:DR_WORKERS]
               [--direction-report:dr-workers DIRECTION_REPORT:DR_WORKERS]
               [--brigade-report:dr-workers BRIGADE_REPORT:DR_WORKERS]
               [--direction:dr-workers DIRECTION:DR_WORKERS]
               [--direction:fg-workers DIRECTION:FG_WORKERS]
               APP

        positional arguments:
          APP

        options:
          -h, --help
          --location-report:dr-workers LOCATION_REPORT:DR_WORKERS
                                (default: 4)
          --direction-report:dr-workers DIRECTION_REPORT:DR_WORKERS
                                (default: 4)
          --brigade-report:dr-workers BRIGADE_REPORT:DR_WORKERS
                                (default: 4)
          --direction:dr-workers DIRECTION:DR_WORKERS
                                (default: 4)
          --direction:fg-workers DIRECTION:FG_WORKERS
                                (default: 4)

        ```
    """

    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("-h", "--help", action="store_true")
    parser.add_argument("APP", nargs=1)
    args: argparse.Namespace
    unknown: list[str]
    args, unknown = parser.parse_known_args()
    app_arg = args.APP[-1].split(":") if args.APP else None
    if not app_arg and args.help:
        parser.print_help()
        return
    model_module = import_module(app_arg[0])
    modified = True
    while modified:
        brook: PyBrook = (
            getattr(model_module, app_arg[1])
            if len(app_arg) > 1
            else model_module.brook
        )
        brook.process_model()
        workers_config = add_consumer_args(parser, brook.consumers)
        args = parser.parse_args()
        if args.help:
            parser.print_help()
            return
        update_workers_config(args, workers_config)
        handler = ModelChangeEventHandler(brook)
        observer = Observer()
        observer.schedule(handler, model_module.__file__)  # noqa: WPS609
        observer.start()
        brook.run(config=workers_config)
        observer.stop()
        observer.join()
        reload(model_module)
        modified = handler.modified

update_workers_config(args, workers_config)

Updates workers_config with settings loaded from argparse arguments.

Parameters:

Name Type Description Default
args Namespace

An argparse Namespace

 required
workers_config dict[str, ConsumerConfig]

A dictionary of consumer configs to update using settings loaded from the args argument.

 required

Returns:

Source code in pybrook/__main__.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def update_workers_config(
    args: argparse.Namespace, workers_config: dict[str, ConsumerConfig]
):
    """
    Updates `workers_config` with settings loaded from argparse arguments.

    Args:
        args: An argparse `Namespace`
        workers_config:  A dictionary of
            consumer configs to update using
            settings loaded from the `args` argument.

    Returns:

    """
    for c in workers_config:
        for arg in ("workers",):
            arg_name: str = c.replace("-", "_") + "_" + arg
            setattr(workers_config[c], arg, getattr(args, arg_name))