Skip to content

File: Broken/__main__.py

Broken.__main__

ProjectLanguage

Bases: BrokenEnum

Source code in Broken/__main__.py
39
40
41
42
43
44
class ProjectLanguage(BrokenEnum):
    Unknown = "unknown"
    Python  = "python"
    NodeJS  = "nodejs"
    Rust    = "rust"
    CPP     = "cpp"

Unknown

Unknown = 'unknown'

Python

Python = 'python'

NodeJS

NodeJS = 'nodejs'

Rust

Rust = 'rust'

CPP

CPP = 'cpp'

ProjectManager

Source code in Broken/__main__.py
 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
113
114
115
116
117
118
119
120
121
122
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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
@define
class ProjectManager:
    path: Path
    name: str = "Unknown"
    cli: BrokenTyper = None

    # # Main entry point

    def main(self, ctx: Context) -> None:
        self.cli = BrokenTyper(help=False)
        self.cli.command(self.update)
        self.cli.command(self.compile)
        self.cli.command(self.run, context=True)
        with BrokenPath.pushd(self.path, echo=False):
            self.cli(*ctx.args)

    # # Initialization

    def __attrs_post_init__(self):
        self.name = self.path.name

    def __eq__(self, other: Self) -> bool:
        return self.path == other.path

    # # Utility Attributes

    @property
    def version(self) -> str:
        import arrow
        now = arrow.utcnow().format("YYYY.M.D")
        return self.config.setdefault("version", now)

    @property
    def description(self) -> str:
        description = ""

        # Read Python's pyproject.toml
        if (config := self.path/"pyproject.toml").exists():
            description = (
                toml.loads(config.read_text())
                .get("project", {})
                .get("description", "")
            )

        # Read Rust's Cargo.toml
        elif (config := self.path/"Cargo.toml").exists():
            description = (
                toml.loads(config.read_text())
                .get("package", {})
                .get("description", "")
            )

        return description

    @property
    def languages(self) -> set[ProjectLanguage]:
        languages = set()

        # Best attempts to detect language
        if (self.path/"pyproject.toml").exists():
            languages.add(ProjectLanguage.Python)
        elif (self.path/"Cargo.toml").exists():
            languages.add(ProjectLanguage.Rust)
        elif (self.path/"meson.build").exists():
            languages.add(ProjectLanguage.CPP)
        else:
            languages.add(ProjectLanguage.Unknown)

        return languages

    @property
    def pyproject(self) -> DotMap:
        return DotMap(toml.loads((self.path/"pyproject.toml").read_text()))

    @property
    def cargo_toml(self) -> DotMap:
        return DotMap(toml.loads((self.path/"Cargo.toml").read_text()))

    @property
    def _pretty_language(self) -> str:
        if self.is_python: return f"🐍 (Python) {self.description}"
        if self.is_nodejs: return f"🟢 (NodeJS) {self.description}"
        if self.is_rust:   return f"🦀 (Rust  ) {self.description}"
        if self.is_cpp:    return f"🌀 (C/C++ ) {self.description}"
        return self.description

    # Shorthands for project language

    @property
    def is_known(self) -> bool:
        return ProjectLanguage.Unknown not in self.languages
    @property
    def is_python(self) -> bool:
        return ProjectLanguage.Python in self.languages
    @property
    def is_nodejs(self) -> bool:
        return ProjectLanguage.NodeJS in self.languages
    @property
    def is_rust(self) -> bool:
        return ProjectLanguage.Rust in self.languages
    @property
    def is_cpp(self) -> bool:
        return ProjectLanguage.CPP in self.languages

    # # Commands

    def update(self) -> None:
        """✨ Update this project's dependencies"""
        if self.is_python:
            outdated = shell("uv", "pip", "list", "--outdated", "--format=json", output=True)
            pyproject = (self.path/"pyproject.toml").read_text("utf8")

            # Replaces any package version of '~=', '>=', '^=' with latest
            for package in map(DotMap, json.loads(outdated)):
                pyproject = re.sub(
                    rf'({re.escape(package.name)}(?:\[[^\]]+\])?\s*(?:~=|>=|\^))\s*([^\"]*)"',
                    rf'\g<1>{package.latest_version}"',
                    pyproject
                )

            # Write changes
            (self.path/"pyproject.toml").write_text(pyproject, "utf8")
            shell("uv", "sync", "--all-packages")

        if self.is_nodejs:
            shell("pnpm", "update")
        if self.is_rust:
            shell("cargo", "update")
        if self.is_cpp:
            log.error("C++ projects are not supported yet")

    def run(self, ctx: Context,
        loop:  Annotated[bool, Option("--loop",  help="Press Enter after each run to run again")]=False,
        clear: Annotated[bool, Option("--clear", help="Clear terminal before running")]=False,
        debug: Annotated[bool, Option("--debug", help="Debug mode for Rust projects")]=False,
    ) -> None:
        """🔥 Run this project with all arguments that follow"""

        while True:
            BrokenPlatform.clear_terminal() if clear else None

            if self.is_python:
                log.info(f"Hey! Just type '{self.name.lower()}' to run the project directly, it's faster 😉")
                return

            elif self.is_rust:
                raise RuntimeError(log.error("Rust projects are not supported yet"))
                _status = shell(
                    "cargo", "run",
                    "--bin", self.name,
                    ["--profile", "release"] if not debug else [],
                    "--features", self.rust_features,
                    "--", ctx.args
                )

            elif self.is_cpp:
                BUILD_DIR = BROKEN.DIRECTORIES.REPO_BUILD/self.name
                if shell("meson", BUILD_DIR, "--reconfigure", "--buildtype", "release").returncode != 0:
                    exit(log.error(f"Could not build project ({self.name})") or 1)
                if shell("ninja", "-C", BUILD_DIR).returncode != 0:
                    exit(log.error(f"Could not build project ({self.name})") or 1)
                binary = next(BUILD_DIR.glob(f"{self.name.lower()}"))
                shell(binary, ctx.args)

            if not loop:
                break

            import rich.prompt
            log.success(f"Project ({self.name}) finished successfully")
            if not rich.prompt.Confirm.ask("(Infinite mode) Press Enter to run again", default=True):
                break

    # # Python shenanigans

    def compile(self,
        target: Annotated[list[PlatformEnum],
            Option("--target", "-t",
            help="Target platforms to build binaries for"
        )] = [BrokenPlatform.Host],

        tarball: Annotated[bool,
            Option("--tarball", "-z",
            help="Create a compressed tarball archive for unix releases",
        )] = False,

        standalone: Annotated[bool,
            Option("--standalone", "-s",
            help="(Standalone) Create self-contained distributions with all dependencies",
        )] = False,

        torch: Annotated[Optional[TorchRelease],
            Option("--torch", "-r",
            help="(Standalone) Bundle a specific PyTorch version with the project"
        )] = None,
    ) -> None:
        """
        📦 Release the Project as a distributable binary

        Note:
            - Requires mingw packages for Windows cross compilation from Linux
        """

        # Recurse on each target item
        if isinstance(target, list):
            for target in flatten(map(PlatformEnum.get_all, target)):
                ProjectManager.compile(**locals())
            return None

        # Filter invalid host -> target combinations of all targets
        if BrokenPlatform.OnLinux and (target.system == SystemEnum.MacOS):
            return log.skip(f"Linux can't [italic]easily[/] compile for {target.system}")
        elif BrokenPlatform.OnMacOS and (target.system != SystemEnum.MacOS):
            return log.skip("macOS can only [italic]easily[/] compile for itself")
        elif BrokenPlatform.OnWindows and (target.system != SystemEnum.Windows):
            return log.skip("Windows can only [italic]easily[/] compile for itself")
        elif (target == PlatformEnum.WindowsARM64):
            return log.skip("Windows on ARM is not widely supported")

        # Automatically bundle some torch on projects that needs it
        if (self.name == "DepthFlow"):
            torch = (torch or SimpleTorch.CPU.value)

        # Non-macOS ARM builds can be unstable/not tested, disable on CI
        if (target.arch.is_arm() and (target.system != SystemEnum.MacOS)):
            log.warning("ARM general support is only present in macOS")

        # Fixme: Wait for uv's implementation of pip wheel for my own sanity
        if (standalone and target != BrokenPlatform.Host):
            log.error("Standalone releases are best built in a host matching the target platform")
            log.error("• Awaiting implementation of (https://github.com/astral-sh/uv/issues/1681)")
            log.error(f"• Attempted to build for '{target.value}' on '{BrokenPlatform.Host.value}'")
            return

        log.note("Building Project Release for", target)

        if self.is_python:
            BrokenManager.rust()
            BUILD_DIR: Path = BROKEN.DIRECTORIES.REPO_BUILD/"Cargo"
            BUILD_WHL: Path = BROKEN.DIRECTORIES.BUILD_WHEELS
            PYTHON_VERSION: str = "3.12"

            # Remove previous build cache for pyapp
            for path in BUILD_DIR.rglob("pyapp*"):
                BrokenPath.remove(path)

            # Write a releases env config file
            (RELEASE_ENV := BROKEN.RESOURCES.ROOT/"Release.env").write_text('\n'.join(
                f"{key}={value}" for key, value in dict(
                    # Placeholder
                ).items()
            ))

            # Build wheels, find main and extra ones
            Environment.set("PYAPP_RELEASE", 1)
            WHEELS = BrokenManager().pypi(all=True)
            MAIN   = next(WHEELS.glob("broken_source*"))
            EXTRA  = set(WHEELS.glob("*.whl")) - {MAIN}

            if (standalone):

                # Fixme: Improve this with (https://github.com/astral-sh/uv/issues/1681)
                def fetch_wheel(
                    dependencies: Union[str, list[str]],
                    index: Optional[str]=None,
                    nodeps: bool=True,
                ) -> None:
                    if (returncode := shell(
                        sys.executable, "-m", "pip", "download", dependencies,
                        (("--platform", x) for x in target.pip_platform),
                        "--python-version", PYTHON_VERSION,
                        "--only-binary=:all:"*(not nodeps),
                        "--no-deps"*(nodeps),
                        "--prefer-binary",
                        every("--index", index),
                        "--dest", BUILD_WHL,
                    ).returncode) != 0:
                        log.error(f"Failed to download dependency ({dependencies})")
                        exit(returncode)

                from concurrent.futures import ThreadPoolExecutor

                with ThreadPoolExecutor(max_workers=10) as pool:
                    for dependency in filter(None, shell(
                        "uv", "export", "--all-packages",
                        "--format", "requirements-txt",
                        "--no-editable", "--no-hashes",
                        "--no-header", "--no-dev",
                        output=True
                    ).splitlines()):

                        # Skip editable packages
                        if (dependency.startswith(".")):
                            continue

                        # Skip audioop on Python 3.13+ as it was dropped from stdlib
                        if (PYTHON_VERSION == "3.13") and ("audioop" in dependency):
                            continue

                        # Ignore platform constraints
                        dependency = dependency.split(";")[0]

                        pool.submit(fetch_wheel, dependency)

                # Add all dependencies wheels and sdists to the extra list
                EXTRA |= set(BUILD_WHL.glob("*.whl")) - (EXTRA | {MAIN})
                EXTRA |= set(BUILD_WHL.glob("*.tar.gz"))

                # Why PyTorch can't be normal?
                if bool(torch):

                    # Help the linker deal with 3.2 GB Torch CUDA binaries..
                    Environment.append("RUSTFLAGS", "-C code-model=large")

                    fetch_wheel(
                        dependencies=torch.packages,
                        index=torch.index,
                        nodeps=False
                    )

                    # Remove new duplicate and list them on extra wheels
                    for file in set(BUILD_WHL.iterdir()) - (EXTRA | {MAIN}):

                        # Note: Need case insensitive enabled due shit like this:
                        # - https://pypi.org/project/Jinja2/3.1.4/#jinja2-3.1.4-py3-none-any.whl
                        # - https://download.pytorch.org/whl/Jinja2-3.1.4-py3-none-any.whl
                        duplicates = list(BUILD_WHL.glob(
                            pattern=f"{file.name.split("-")[0]}-*",
                            case_sensitive=False
                        ))

                        if len(duplicates) > 1:
                            log.info(f"Removing duplicate: {file}")
                            file.unlink()
                            continue

                        EXTRA |= {file}

            # Pyapp configuration
            Environment.update(
                PYAPP_PROJECT_PATH=str(MAIN),
                PYAPP_EXTRA_WHEELS=";".join(map(str, EXTRA)),
                PYAPP_PIP_EXTRA_ARGS=("--no-deps"*standalone),
                PYAPP_PYTHON_VERSION=PYTHON_VERSION,
                PYAPP_EXEC_MODULE=self.name,
                PYAPP_DISTRIBUTION_EMBED=1,
                PYAPP_PASS_LOCATION=1,
                PYAPP_UV_ENABLED=1,
                PYAPP_UV_EMBED=1,
            )

            # Rust configuration
            Environment.update(
                CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER=shutil.which("aarch64-linux-gnu-gcc"),
            )

            # Cache Rust compilation across projects
            Environment.set("CARGO_HOME", BUILD_DIR)
            shell("rustup", "target", "add", target.triple)

            # Cargo warning: We're not 'installing' a utility
            BrokenPath.add_to_path(BUILD_DIR/"bin")

            if (_PYAPP_FORK := True):
                if not (fork := BROKEN.DIRECTORIES.REPO_BUILD/"PyApp").exists():
                    shell("git", "clone", "https://github.com/BrokenSource/PyApp", fork, "-b", "custom")
                embed = (fork/"src"/"embed")

                # Remove previous embeddings if any
                for file in embed.glob("*.whl"):
                    file.unlink()
                for file in embed.glob("*.tar.gz"):
                    file.unlink()

                # Actually compile it
                if shell(
                    "cargo", "install",
                    "--path", fork, "--force",
                    "--root", BUILD_DIR,
                    "--target", target.triple,
                ).returncode != 0:
                    raise RuntimeError(log.error("Failed to compile PyApp"))
            else:
                if shell(
                    "cargo", "install",
                    "pyapp", "--force",
                    "--root", BUILD_DIR,
                    "--target", target.triple,
                ).returncode != 0:
                    raise RuntimeError(log.error("Failed to compile PyApp"))

            RELEASE_ENV.unlink()

            # Find the compiled binary
            binary = next((BUILD_DIR/"bin").glob("pyapp*"))
            log.info(f"Compiled Pyapp binary at ({binary})")
            BrokenPath.make_executable(binary)

            # Rename the compiled binary to the final release name
            release_path = BROKEN.DIRECTORIES.REPO_RELEASES / ''.join((
                f"{self.name.lower()}",
                f"-{target.value}",
                f"-v{BROKEN.VERSION}",
                f"-{torch.flavor}" if torch else "",
                "-standalone"*standalone,
                f"{target.extension}",
            ))
            BrokenPath.copy(src=binary, dst=release_path)
            BrokenPath.make_executable(release_path)

            # Release a tar.gz to keep chmod +x attributes
            if tarball and ("windows" not in target.name):
                release_path = BrokenPath.gzip(release_path, remove=True)

            log.success(f"Built Project Release at ({release_path})")

path

path: Path

name

name: str = 'Unknown'

cli

cli: BrokenTyper = None

main

main(ctx: Context) -> None
Source code in Broken/__main__.py
56
57
58
59
60
61
62
def main(self, ctx: Context) -> None:
    self.cli = BrokenTyper(help=False)
    self.cli.command(self.update)
    self.cli.command(self.compile)
    self.cli.command(self.run, context=True)
    with BrokenPath.pushd(self.path, echo=False):
        self.cli(*ctx.args)

__attrs_post_init__

__attrs_post_init__()
Source code in Broken/__main__.py
66
67
def __attrs_post_init__(self):
    self.name = self.path.name

__eq__

__eq__(other: Self) -> bool
Source code in Broken/__main__.py
69
70
def __eq__(self, other: Self) -> bool:
    return self.path == other.path

version

version: str

description

description: str

languages

languages: set[ProjectLanguage]

pyproject

pyproject: DotMap

cargo_toml

cargo_toml: DotMap

is_known

is_known: bool

is_python

is_python: bool

is_nodejs

is_nodejs: bool

is_rust

is_rust: bool

is_cpp

is_cpp: bool

update

update() -> None

✨ Update this project's dependencies

Source code in Broken/__main__.py
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def update(self) -> None:
    """✨ Update this project's dependencies"""
    if self.is_python:
        outdated = shell("uv", "pip", "list", "--outdated", "--format=json", output=True)
        pyproject = (self.path/"pyproject.toml").read_text("utf8")

        # Replaces any package version of '~=', '>=', '^=' with latest
        for package in map(DotMap, json.loads(outdated)):
            pyproject = re.sub(
                rf'({re.escape(package.name)}(?:\[[^\]]+\])?\s*(?:~=|>=|\^))\s*([^\"]*)"',
                rf'\g<1>{package.latest_version}"',
                pyproject
            )

        # Write changes
        (self.path/"pyproject.toml").write_text(pyproject, "utf8")
        shell("uv", "sync", "--all-packages")

    if self.is_nodejs:
        shell("pnpm", "update")
    if self.is_rust:
        shell("cargo", "update")
    if self.is_cpp:
        log.error("C++ projects are not supported yet")

run

run(
    ctx: Context,
    loop: Annotated[
        bool,
        Option(
            --loop,
            help="Press Enter after each run to run again",
        ),
    ] = False,
    clear: Annotated[
        bool,
        Option(
            --clear, help="Clear terminal before running"
        ),
    ] = False,
    debug: Annotated[
        bool,
        Option(
            --debug, help="Debug mode for Rust projects"
        ),
    ] = False,
) -> None

🔥 Run this project with all arguments that follow

Source code in Broken/__main__.py
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def run(self, ctx: Context,
    loop:  Annotated[bool, Option("--loop",  help="Press Enter after each run to run again")]=False,
    clear: Annotated[bool, Option("--clear", help="Clear terminal before running")]=False,
    debug: Annotated[bool, Option("--debug", help="Debug mode for Rust projects")]=False,
) -> None:
    """🔥 Run this project with all arguments that follow"""

    while True:
        BrokenPlatform.clear_terminal() if clear else None

        if self.is_python:
            log.info(f"Hey! Just type '{self.name.lower()}' to run the project directly, it's faster 😉")
            return

        elif self.is_rust:
            raise RuntimeError(log.error("Rust projects are not supported yet"))
            _status = shell(
                "cargo", "run",
                "--bin", self.name,
                ["--profile", "release"] if not debug else [],
                "--features", self.rust_features,
                "--", ctx.args
            )

        elif self.is_cpp:
            BUILD_DIR = BROKEN.DIRECTORIES.REPO_BUILD/self.name
            if shell("meson", BUILD_DIR, "--reconfigure", "--buildtype", "release").returncode != 0:
                exit(log.error(f"Could not build project ({self.name})") or 1)
            if shell("ninja", "-C", BUILD_DIR).returncode != 0:
                exit(log.error(f"Could not build project ({self.name})") or 1)
            binary = next(BUILD_DIR.glob(f"{self.name.lower()}"))
            shell(binary, ctx.args)

        if not loop:
            break

        import rich.prompt
        log.success(f"Project ({self.name}) finished successfully")
        if not rich.prompt.Confirm.ask("(Infinite mode) Press Enter to run again", default=True):
            break

compile

compile(
    target: Annotated[
        list[PlatformEnum],
        Option(
            --target,
            -t,
            help="Target platforms to build binaries for",
        ),
    ] = [BrokenPlatform.Host],
    tarball: Annotated[
        bool,
        Option(
            --tarball,
            -z,
            help="Create a compressed tarball archive for unix releases",
        ),
    ] = False,
    standalone: Annotated[
        bool,
        Option(
            --standalone,
            -s,
            help="(Standalone) Create self-contained distributions with all dependencies",
        ),
    ] = False,
    torch: Annotated[
        Optional[TorchRelease],
        Option(
            --torch,
            -r,
            help="(Standalone) Bundle a specific PyTorch version with the project",
        ),
    ] = None,
) -> None

📦 Release the Project as a distributable binary

Note
  • Requires mingw packages for Windows cross compilation from Linux
Source code in Broken/__main__.py
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
def compile(self,
    target: Annotated[list[PlatformEnum],
        Option("--target", "-t",
        help="Target platforms to build binaries for"
    )] = [BrokenPlatform.Host],

    tarball: Annotated[bool,
        Option("--tarball", "-z",
        help="Create a compressed tarball archive for unix releases",
    )] = False,

    standalone: Annotated[bool,
        Option("--standalone", "-s",
        help="(Standalone) Create self-contained distributions with all dependencies",
    )] = False,

    torch: Annotated[Optional[TorchRelease],
        Option("--torch", "-r",
        help="(Standalone) Bundle a specific PyTorch version with the project"
    )] = None,
) -> None:
    """
    📦 Release the Project as a distributable binary

    Note:
        - Requires mingw packages for Windows cross compilation from Linux
    """

    # Recurse on each target item
    if isinstance(target, list):
        for target in flatten(map(PlatformEnum.get_all, target)):
            ProjectManager.compile(**locals())
        return None

    # Filter invalid host -> target combinations of all targets
    if BrokenPlatform.OnLinux and (target.system == SystemEnum.MacOS):
        return log.skip(f"Linux can't [italic]easily[/] compile for {target.system}")
    elif BrokenPlatform.OnMacOS and (target.system != SystemEnum.MacOS):
        return log.skip("macOS can only [italic]easily[/] compile for itself")
    elif BrokenPlatform.OnWindows and (target.system != SystemEnum.Windows):
        return log.skip("Windows can only [italic]easily[/] compile for itself")
    elif (target == PlatformEnum.WindowsARM64):
        return log.skip("Windows on ARM is not widely supported")

    # Automatically bundle some torch on projects that needs it
    if (self.name == "DepthFlow"):
        torch = (torch or SimpleTorch.CPU.value)

    # Non-macOS ARM builds can be unstable/not tested, disable on CI
    if (target.arch.is_arm() and (target.system != SystemEnum.MacOS)):
        log.warning("ARM general support is only present in macOS")

    # Fixme: Wait for uv's implementation of pip wheel for my own sanity
    if (standalone and target != BrokenPlatform.Host):
        log.error("Standalone releases are best built in a host matching the target platform")
        log.error("• Awaiting implementation of (https://github.com/astral-sh/uv/issues/1681)")
        log.error(f"• Attempted to build for '{target.value}' on '{BrokenPlatform.Host.value}'")
        return

    log.note("Building Project Release for", target)

    if self.is_python:
        BrokenManager.rust()
        BUILD_DIR: Path = BROKEN.DIRECTORIES.REPO_BUILD/"Cargo"
        BUILD_WHL: Path = BROKEN.DIRECTORIES.BUILD_WHEELS
        PYTHON_VERSION: str = "3.12"

        # Remove previous build cache for pyapp
        for path in BUILD_DIR.rglob("pyapp*"):
            BrokenPath.remove(path)

        # Write a releases env config file
        (RELEASE_ENV := BROKEN.RESOURCES.ROOT/"Release.env").write_text('\n'.join(
            f"{key}={value}" for key, value in dict(
                # Placeholder
            ).items()
        ))

        # Build wheels, find main and extra ones
        Environment.set("PYAPP_RELEASE", 1)
        WHEELS = BrokenManager().pypi(all=True)
        MAIN   = next(WHEELS.glob("broken_source*"))
        EXTRA  = set(WHEELS.glob("*.whl")) - {MAIN}

        if (standalone):

            # Fixme: Improve this with (https://github.com/astral-sh/uv/issues/1681)
            def fetch_wheel(
                dependencies: Union[str, list[str]],
                index: Optional[str]=None,
                nodeps: bool=True,
            ) -> None:
                if (returncode := shell(
                    sys.executable, "-m", "pip", "download", dependencies,
                    (("--platform", x) for x in target.pip_platform),
                    "--python-version", PYTHON_VERSION,
                    "--only-binary=:all:"*(not nodeps),
                    "--no-deps"*(nodeps),
                    "--prefer-binary",
                    every("--index", index),
                    "--dest", BUILD_WHL,
                ).returncode) != 0:
                    log.error(f"Failed to download dependency ({dependencies})")
                    exit(returncode)

            from concurrent.futures import ThreadPoolExecutor

            with ThreadPoolExecutor(max_workers=10) as pool:
                for dependency in filter(None, shell(
                    "uv", "export", "--all-packages",
                    "--format", "requirements-txt",
                    "--no-editable", "--no-hashes",
                    "--no-header", "--no-dev",
                    output=True
                ).splitlines()):

                    # Skip editable packages
                    if (dependency.startswith(".")):
                        continue

                    # Skip audioop on Python 3.13+ as it was dropped from stdlib
                    if (PYTHON_VERSION == "3.13") and ("audioop" in dependency):
                        continue

                    # Ignore platform constraints
                    dependency = dependency.split(";")[0]

                    pool.submit(fetch_wheel, dependency)

            # Add all dependencies wheels and sdists to the extra list
            EXTRA |= set(BUILD_WHL.glob("*.whl")) - (EXTRA | {MAIN})
            EXTRA |= set(BUILD_WHL.glob("*.tar.gz"))

            # Why PyTorch can't be normal?
            if bool(torch):

                # Help the linker deal with 3.2 GB Torch CUDA binaries..
                Environment.append("RUSTFLAGS", "-C code-model=large")

                fetch_wheel(
                    dependencies=torch.packages,
                    index=torch.index,
                    nodeps=False
                )

                # Remove new duplicate and list them on extra wheels
                for file in set(BUILD_WHL.iterdir()) - (EXTRA | {MAIN}):

                    # Note: Need case insensitive enabled due shit like this:
                    # - https://pypi.org/project/Jinja2/3.1.4/#jinja2-3.1.4-py3-none-any.whl
                    # - https://download.pytorch.org/whl/Jinja2-3.1.4-py3-none-any.whl
                    duplicates = list(BUILD_WHL.glob(
                        pattern=f"{file.name.split("-")[0]}-*",
                        case_sensitive=False
                    ))

                    if len(duplicates) > 1:
                        log.info(f"Removing duplicate: {file}")
                        file.unlink()
                        continue

                    EXTRA |= {file}

        # Pyapp configuration
        Environment.update(
            PYAPP_PROJECT_PATH=str(MAIN),
            PYAPP_EXTRA_WHEELS=";".join(map(str, EXTRA)),
            PYAPP_PIP_EXTRA_ARGS=("--no-deps"*standalone),
            PYAPP_PYTHON_VERSION=PYTHON_VERSION,
            PYAPP_EXEC_MODULE=self.name,
            PYAPP_DISTRIBUTION_EMBED=1,
            PYAPP_PASS_LOCATION=1,
            PYAPP_UV_ENABLED=1,
            PYAPP_UV_EMBED=1,
        )

        # Rust configuration
        Environment.update(
            CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER=shutil.which("aarch64-linux-gnu-gcc"),
        )

        # Cache Rust compilation across projects
        Environment.set("CARGO_HOME", BUILD_DIR)
        shell("rustup", "target", "add", target.triple)

        # Cargo warning: We're not 'installing' a utility
        BrokenPath.add_to_path(BUILD_DIR/"bin")

        if (_PYAPP_FORK := True):
            if not (fork := BROKEN.DIRECTORIES.REPO_BUILD/"PyApp").exists():
                shell("git", "clone", "https://github.com/BrokenSource/PyApp", fork, "-b", "custom")
            embed = (fork/"src"/"embed")

            # Remove previous embeddings if any
            for file in embed.glob("*.whl"):
                file.unlink()
            for file in embed.glob("*.tar.gz"):
                file.unlink()

            # Actually compile it
            if shell(
                "cargo", "install",
                "--path", fork, "--force",
                "--root", BUILD_DIR,
                "--target", target.triple,
            ).returncode != 0:
                raise RuntimeError(log.error("Failed to compile PyApp"))
        else:
            if shell(
                "cargo", "install",
                "pyapp", "--force",
                "--root", BUILD_DIR,
                "--target", target.triple,
            ).returncode != 0:
                raise RuntimeError(log.error("Failed to compile PyApp"))

        RELEASE_ENV.unlink()

        # Find the compiled binary
        binary = next((BUILD_DIR/"bin").glob("pyapp*"))
        log.info(f"Compiled Pyapp binary at ({binary})")
        BrokenPath.make_executable(binary)

        # Rename the compiled binary to the final release name
        release_path = BROKEN.DIRECTORIES.REPO_RELEASES / ''.join((
            f"{self.name.lower()}",
            f"-{target.value}",
            f"-v{BROKEN.VERSION}",
            f"-{torch.flavor}" if torch else "",
            "-standalone"*standalone,
            f"{target.extension}",
        ))
        BrokenPath.copy(src=binary, dst=release_path)
        BrokenPath.make_executable(release_path)

        # Release a tar.gz to keep chmod +x attributes
        if tarball and ("windows" not in target.name):
            release_path = BrokenPath.gzip(release_path, remove=True)

        log.success(f"Built Project Release at ({release_path})")

BrokenManager

Bases: BrokenSingleton

Source code in Broken/__main__.py
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
@define
class BrokenManager(BrokenSingleton):
    projects: list[ProjectManager] = Factory(list)

    cli: BrokenTyper = Factory(lambda: BrokenTyper(description=(
        "🚀 Broken Source Software Monorepo development manager script\n\n"
        "• Tip: run \"broken (command) --help\" for options on commands or projects ✨\n\n"
    )))

    @property
    def python_projects(self) -> list[ProjectManager]:
        return list(filter(lambda project: project.is_python, self.projects))

    def __attrs_post_init__(self) -> None:
        self.projects.append(broken := ProjectManager(BROKEN.DIRECTORIES.REPOSITORY))
        self.find_projects(BROKEN.DIRECTORIES.REPO_PROJECTS)
        self.find_projects(BROKEN.DIRECTORIES.REPO_META)

        with self.cli.panel("🚀 Core"):
            self.cli.command(BrokenTorch.install)
            self.cli.command(self.insiders)
            self.cli.command(self.clone)
            self.cli.command(self.rust)

        with self.cli.panel("📦 Development"):
            self.cli.command(self.compile_all, hidden=True)
            self.cli.command(self.docker)
            self.cli.command(self.website)
            self.cli.command(self.pypi)
            self.cli.command(self.upgrade)
            self.cli.command(self.clean)
            self.cli.command(self.sync)

        self.cli.command(self.tremeschin, hidden=True)

        for project in self.projects:
            self.cli.command(
                target=project.main,
                name=project.name.lower(),
                description=project._pretty_language,
                panel=f"🔥 Projects at [bold]({project.path.parent})[/]",
                hidden=(project is broken),
                context=True,
                help=False,
            )

    def find_projects(self, path: Path, max_depth: int=1) -> None:
        if (not path.exists()):
            return
        if (max_depth <= 0):
            return

        # Special directories that could contain projects
        if (projects := path/"Projects").exists():
            self.find_projects(projects, 1)

        # Note: Avoid hidden, workspace, recursion
        for directory in (path := BrokenPath.get(path)).iterdir():
            if directory.is_file():
                continue
            if directory.is_symlink() or directory.is_dir():
                self.find_projects(directory, max_depth - 1)
            if (project := ProjectManager(directory)).is_known:
                self.projects.append(project)

    # ---------------------------------------------------------------------------------------------|
    # Meta Repositories

    def clone(self,
        repo:     Annotated[str,  Argument(help="URL of the Git Repository to Clone")],
        path:     Annotated[Path, Option("--path",     "-p", help="Path to clone the repository to")]=BROKEN.DIRECTORIES.REPO_META,
        recurse:  Annotated[bool, Option("--recurse",  "-r", help="Clone all submodules recursively")]=True,
        checkout: Annotated[str,  Option("--checkout", "-c", help="Checkout recursively branch or tag")]="main",
    ) -> Path:
        """🔗 Clone a project in the Meta directory"""
        from urllib.parse import urlparse

        # If the path isn't a repo, use the repository name
        if (path.exists()) and (not (Path(path)/".git").exists()):
            log.minor(f"Path {path} isn't a repository, appending the url name")
            path = (path/Path(urlparse(str(repo).removesuffix(".git")).path).stem)

        # Only attempt cloning if non-existent
        if (not path.exists()):
            with BrokenPath.pushd(path.parent, echo=False):
                shell("git", "clone", ("--recurse-submodules"*recurse), repo, path)

        # Not having .git is a failed clone
        if not (path/".git").exists():
            log.error(f"Invalid repository at ({path}), perhaps try removing it")
            exit(1)

        with BrokenPath.pushd(path, echo=False):
            shell("git", "submodule", "foreach", "--recursive", f"git checkout {checkout} || true")

        return path

    def insiders(self):
        """💎 Clone the Insiders repository (WIP, No content)"""
        self.clone("https://github.com/BrokenSource/Insiders", BROKEN.DIRECTORIES.INSIDERS)

    def tremeschin(self):
        Tremeschin = (BROKEN.DIRECTORIES.REPO_META/"Tremeschin")
        self.clone("https://github.com/Tremeschin/Personal", Tremeschin)
        self.clone("https://github.com/Tremeschin/Private",  Tremeschin/"Private")

    # ---------------------------------------------------------------------------------------------|
    # Core section

    def website(self, deploy: Annotated[bool, Option("--deploy", "-d", help="Deploy Unified Website to GitHub Pages")]=False) -> None:
        """📚 Serve or deploy the monorepo website"""
        if deploy:
            Environment.set("CODE_REFERENCE", 1)
            shell("mkdocs", "gh-deploy", "--force")
        else:
            shell("mkdocs", "serve")

    def compile_all(self,
        standalone: Annotated[bool, Option("--standalone", "-s")]=False,
    ) -> None:
        for project in self.projects[1:]:
            project.compile(
                target=[PlatformEnum._AllHost],
                standalone=standalone,
                tarball=True,
            )

    def pypi(self,
        publish: Annotated[bool, Option("--publish", "-p", help="Publish the wheel to PyPI")]=False,
        output:  Annotated[Path, Option("--output",  "-o", help="Output directory for wheels")]=BROKEN.DIRECTORIES.BUILD_WHEELS,
        all:     Annotated[bool, Option("--all",     "-a", help="Build all projects")]=True,
    ) -> Path:
        """🧀 Build all project wheels and publish to PyPI"""
        BrokenPath.recreate(output)
        shell("uv", "build", "--wheel", ("--all"*all), "--out-dir", output)
        shell("uv", "publish", f"{output}/*.whl", skip=(not publish))
        return Path(output)

    def docker(self,
        push:  Annotated[bool, Option("--push",  "-p", help="Push built images to GHCR")]=False,
        clean: Annotated[bool, Option("--clean", "-c", help="Remove local images after pushing")]=False,
    ) -> None:
        """🐳 Build and push docker images for all projects"""
        from Broken.Core.BrokenTorch import BrokenTorch

        for build in combinations(
            base_image=["ubuntu:24.04"],
            torch=BrokenTorch.docker(),
        ):
            # Warn: Must use same env vars as in docker-compose.yml
            Environment.set("BASE_IMAGE",    build.base_image)
            Environment.set("TORCH_VERSION", build.torch.number)
            Environment.set("TORCH_FLAVOR",  build.torch.flavor)
            shell("docker", "compose", "build")

            # Assumes all dockerfiles were built by docker compose, fails ok otherwise
            for dockerfile in BROKEN.DIRECTORIES.REPO_DOCKER.glob("*.dockerfile"):
                image:  str = dockerfile.stem
                latest: str = f"{image}:latest"
                flavor: str = build.torch.flavor

                # Tag a latest and versioned flavored images, optional push
                for tag in (f"latest-{flavor}", f"{__version__}-{flavor}"):
                    final: str = f"ghcr.io/brokensource/{image}:{tag}"
                    shell("docker", "tag",  latest, final)
                    shell("docker", "push", final,  skip=(not push))
                    shell("docker", "rmi",  final,  skip=(not clean))

                # No need for generic latest image
                shell("docker", "rmi", latest)

    def upgrade(self) -> None:
        """📦 Temporary solution to bump pyproject versions"""
        for project in self.projects:
            project.update()

    @staticmethod
    def rust(
        toolchain:   Annotated[str,  Option("--toolchain",   "-t", help="(Any    ) Rust toolchain to use (stable, nightly)")]="stable",
        build_tools: Annotated[bool, Option("--build-tools", "-b", help="(Windows) Install Visual C++ Build Tools")]=True,
    ):
        """🦀 Installs rustup and a rust toolchain"""
        import requests

        # Actions has its own workflow setup
        if (Runtime.GitHub):
            return

        # Install rustup based on platform
        if not shutil.which("rustup"):
            log.info("Rustup wasn't found, will install it")

            if BrokenPlatform.OnWindows:
                shell("winget", "install", "-e", "--id", "Rustlang.Rustup")
            elif BrokenPlatform.OnUnix:
                shell("sh", "-c", requests.get("https://sh.rustup.rs").text, "-y", echo=False)

            # If rustup isn't found, ask user to restart shell
            BrokenPath.add_to_path(Path.home()/".cargo"/"bin")

            if not BrokenPath.which("rustup"):
                log.warning("Rustup was likely installed but wasn't found adding '~/.cargo/bin' to Path")
                log.warning("• Maybe you changed the CARGO_HOME or RUSTUP_HOME environment variables")
                log.warning("• Please restart your shell for Rust toolchain to be on PATH")
                exit(0)

        # Install Visual C++ Build Tools on Windows
        if (BrokenPlatform.OnWindows and build_tools):
            log.warning("You must have Microsoft Visual C++ Build Tools installed to compile Rust projects")
            log.warning("• Broken will try installing it, you might need to restart your shell afterwards")
            shell("winget", "install", "-e", "--id", "Microsoft.VisualStudio.2022.BuildTools", "--override", (
                " --add Microsoft.VisualStudio.Component.VC.Tools.x86.x64"
                " --add Microsoft.VisualStudio.Component.Windows10SDK"
                " --add Microsoft.VisualStudio.Component.Windows11SDK.22000"
                "--wait --passive"
            ))

        class RustToolchain(BrokenEnum):
            Stable  = "stable"
            Nightly = "nightly"

        toolchain = RustToolchain.get(toolchain).value

        # Install or select the correct toolchain
        for line in shell("rustup", "toolchain", "list", output=True, echo=False).split("\n"):
            if ("no installed" in line) or (("default" in line) and (line.split("-")[0] != toolchain)):
                log.info(f"Defaulting Rust toolchain to ({toolchain})")
                shell("rustup", "default", toolchain)
        else:
            log.info(f"Rust toolchain is already the default ({toolchain})")

    def clean(self) -> None:
        """🧹 Remove pycaches, common blob directories"""
        root = BROKEN.DIRECTORIES.REPOSITORY

        for path in root.rglob("__pycache__"):
            BrokenPath.remove(path)

        # Fixed known blob directories
        BrokenPath.remove(BROKEN.DIRECTORIES.REPO_RELEASES)
        BrokenPath.remove(BROKEN.DIRECTORIES.REPO_BUILD)
        BrokenPath.remove(root/".cache")

    def sync(self) -> None:
        """♻️  Synchronize common resources files across all projects"""
        root = BROKEN.DIRECTORIES.REPOSITORY

        for project in self.projects[1:]:
            if (project.path/".github"/".nosync").exists():
                continue
            for file in flatten(
                ((root/".github").glob(ext) for ext in ("*.md", "*.yml")),
                (root/".github"/"ISSUE_TEMPLATE").glob("*.yml"),
                (root/".github"/"hatch_build.py"),
            ):
                target = project.path/file.relative_to(root)
                BrokenPath.copy(src=file, dst=target)

projects

projects: list[ProjectManager] = Factory(list)

cli

cli: BrokenTyper = Factory(
    lambda: BrokenTyper(
        description='🚀 Broken Source Software Monorepo development manager script\n\n• Tip: run "broken (command) --help" for options on commands or projects ✨\n\n'
    )
)

python_projects

python_projects: list[ProjectManager]

__attrs_post_init__

__attrs_post_init__() -> None
Source code in Broken/__main__.py
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
def __attrs_post_init__(self) -> None:
    self.projects.append(broken := ProjectManager(BROKEN.DIRECTORIES.REPOSITORY))
    self.find_projects(BROKEN.DIRECTORIES.REPO_PROJECTS)
    self.find_projects(BROKEN.DIRECTORIES.REPO_META)

    with self.cli.panel("🚀 Core"):
        self.cli.command(BrokenTorch.install)
        self.cli.command(self.insiders)
        self.cli.command(self.clone)
        self.cli.command(self.rust)

    with self.cli.panel("📦 Development"):
        self.cli.command(self.compile_all, hidden=True)
        self.cli.command(self.docker)
        self.cli.command(self.website)
        self.cli.command(self.pypi)
        self.cli.command(self.upgrade)
        self.cli.command(self.clean)
        self.cli.command(self.sync)

    self.cli.command(self.tremeschin, hidden=True)

    for project in self.projects:
        self.cli.command(
            target=project.main,
            name=project.name.lower(),
            description=project._pretty_language,
            panel=f"🔥 Projects at [bold]({project.path.parent})[/]",
            hidden=(project is broken),
            context=True,
            help=False,
        )

find_projects

find_projects(path: Path, max_depth: int = 1) -> None
Source code in Broken/__main__.py
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
def find_projects(self, path: Path, max_depth: int=1) -> None:
    if (not path.exists()):
        return
    if (max_depth <= 0):
        return

    # Special directories that could contain projects
    if (projects := path/"Projects").exists():
        self.find_projects(projects, 1)

    # Note: Avoid hidden, workspace, recursion
    for directory in (path := BrokenPath.get(path)).iterdir():
        if directory.is_file():
            continue
        if directory.is_symlink() or directory.is_dir():
            self.find_projects(directory, max_depth - 1)
        if (project := ProjectManager(directory)).is_known:
            self.projects.append(project)

clone

clone(
    repo: Annotated[
        str,
        Argument(help="URL of the Git Repository to Clone"),
    ],
    path: Annotated[
        Path,
        Option(
            --path,
            -p,
            help="Path to clone the repository to",
        ),
    ] = BROKEN.DIRECTORIES.REPO_META,
    recurse: Annotated[
        bool,
        Option(
            --recurse,
            -r,
            help="Clone all submodules recursively",
        ),
    ] = True,
    checkout: Annotated[
        str,
        Option(
            --checkout,
            -c,
            help="Checkout recursively branch or tag",
        ),
    ] = "main",
) -> Path

🔗 Clone a project in the Meta directory

Source code in Broken/__main__.py
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
def clone(self,
    repo:     Annotated[str,  Argument(help="URL of the Git Repository to Clone")],
    path:     Annotated[Path, Option("--path",     "-p", help="Path to clone the repository to")]=BROKEN.DIRECTORIES.REPO_META,
    recurse:  Annotated[bool, Option("--recurse",  "-r", help="Clone all submodules recursively")]=True,
    checkout: Annotated[str,  Option("--checkout", "-c", help="Checkout recursively branch or tag")]="main",
) -> Path:
    """🔗 Clone a project in the Meta directory"""
    from urllib.parse import urlparse

    # If the path isn't a repo, use the repository name
    if (path.exists()) and (not (Path(path)/".git").exists()):
        log.minor(f"Path {path} isn't a repository, appending the url name")
        path = (path/Path(urlparse(str(repo).removesuffix(".git")).path).stem)

    # Only attempt cloning if non-existent
    if (not path.exists()):
        with BrokenPath.pushd(path.parent, echo=False):
            shell("git", "clone", ("--recurse-submodules"*recurse), repo, path)

    # Not having .git is a failed clone
    if not (path/".git").exists():
        log.error(f"Invalid repository at ({path}), perhaps try removing it")
        exit(1)

    with BrokenPath.pushd(path, echo=False):
        shell("git", "submodule", "foreach", "--recursive", f"git checkout {checkout} || true")

    return path

insiders

insiders()

💎 Clone the Insiders repository (WIP, No content)

Source code in Broken/__main__.py
562
563
564
def insiders(self):
    """💎 Clone the Insiders repository (WIP, No content)"""
    self.clone("https://github.com/BrokenSource/Insiders", BROKEN.DIRECTORIES.INSIDERS)

tremeschin

tremeschin()
Source code in Broken/__main__.py
566
567
568
569
def tremeschin(self):
    Tremeschin = (BROKEN.DIRECTORIES.REPO_META/"Tremeschin")
    self.clone("https://github.com/Tremeschin/Personal", Tremeschin)
    self.clone("https://github.com/Tremeschin/Private",  Tremeschin/"Private")

website

website(
    deploy: Annotated[
        bool,
        Option(
            --deploy,
            -d,
            help="Deploy Unified Website to GitHub Pages",
        ),
    ] = False,
) -> None

📚 Serve or deploy the monorepo website

Source code in Broken/__main__.py
574
575
576
577
578
579
580
def website(self, deploy: Annotated[bool, Option("--deploy", "-d", help="Deploy Unified Website to GitHub Pages")]=False) -> None:
    """📚 Serve or deploy the monorepo website"""
    if deploy:
        Environment.set("CODE_REFERENCE", 1)
        shell("mkdocs", "gh-deploy", "--force")
    else:
        shell("mkdocs", "serve")

compile_all

compile_all(
    standalone: Annotated[
        bool, Option(--standalone, -s)
    ] = False,
) -> None
Source code in Broken/__main__.py
582
583
584
585
586
587
588
589
590
def compile_all(self,
    standalone: Annotated[bool, Option("--standalone", "-s")]=False,
) -> None:
    for project in self.projects[1:]:
        project.compile(
            target=[PlatformEnum._AllHost],
            standalone=standalone,
            tarball=True,
        )

pypi

pypi(
    publish: Annotated[
        bool,
        Option(
            --publish, -p, help="Publish the wheel to PyPI"
        ),
    ] = False,
    output: Annotated[
        Path,
        Option(
            --output,
            -o,
            help="Output directory for wheels",
        ),
    ] = BROKEN.DIRECTORIES.BUILD_WHEELS,
    all: Annotated[
        bool, Option(--all, -a, help="Build all projects")
    ] = True,
) -> Path

🧀 Build all project wheels and publish to PyPI

Source code in Broken/__main__.py
592
593
594
595
596
597
598
599
600
601
def pypi(self,
    publish: Annotated[bool, Option("--publish", "-p", help="Publish the wheel to PyPI")]=False,
    output:  Annotated[Path, Option("--output",  "-o", help="Output directory for wheels")]=BROKEN.DIRECTORIES.BUILD_WHEELS,
    all:     Annotated[bool, Option("--all",     "-a", help="Build all projects")]=True,
) -> Path:
    """🧀 Build all project wheels and publish to PyPI"""
    BrokenPath.recreate(output)
    shell("uv", "build", "--wheel", ("--all"*all), "--out-dir", output)
    shell("uv", "publish", f"{output}/*.whl", skip=(not publish))
    return Path(output)

docker

docker(
    push: Annotated[
        bool,
        Option(
            --push, -p, help="Push built images to GHCR"
        ),
    ] = False,
    clean: Annotated[
        bool,
        Option(
            --clean,
            -c,
            help="Remove local images after pushing",
        ),
    ] = False,
) -> None

🐳 Build and push docker images for all projects

Source code in Broken/__main__.py
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
def docker(self,
    push:  Annotated[bool, Option("--push",  "-p", help="Push built images to GHCR")]=False,
    clean: Annotated[bool, Option("--clean", "-c", help="Remove local images after pushing")]=False,
) -> None:
    """🐳 Build and push docker images for all projects"""
    from Broken.Core.BrokenTorch import BrokenTorch

    for build in combinations(
        base_image=["ubuntu:24.04"],
        torch=BrokenTorch.docker(),
    ):
        # Warn: Must use same env vars as in docker-compose.yml
        Environment.set("BASE_IMAGE",    build.base_image)
        Environment.set("TORCH_VERSION", build.torch.number)
        Environment.set("TORCH_FLAVOR",  build.torch.flavor)
        shell("docker", "compose", "build")

        # Assumes all dockerfiles were built by docker compose, fails ok otherwise
        for dockerfile in BROKEN.DIRECTORIES.REPO_DOCKER.glob("*.dockerfile"):
            image:  str = dockerfile.stem
            latest: str = f"{image}:latest"
            flavor: str = build.torch.flavor

            # Tag a latest and versioned flavored images, optional push
            for tag in (f"latest-{flavor}", f"{__version__}-{flavor}"):
                final: str = f"ghcr.io/brokensource/{image}:{tag}"
                shell("docker", "tag",  latest, final)
                shell("docker", "push", final,  skip=(not push))
                shell("docker", "rmi",  final,  skip=(not clean))

            # No need for generic latest image
            shell("docker", "rmi", latest)

upgrade

upgrade() -> None

📦 Temporary solution to bump pyproject versions

Source code in Broken/__main__.py
636
637
638
639
def upgrade(self) -> None:
    """📦 Temporary solution to bump pyproject versions"""
    for project in self.projects:
        project.update()

rust

rust(
    toolchain: Annotated[
        str,
        Option(
            --toolchain,
            -t,
            help="(Any    ) Rust toolchain to use (stable, nightly)",
        ),
    ] = "stable",
    build_tools: Annotated[
        bool,
        Option(
            --build - tools,
            -b,
            help="(Windows) Install Visual C++ Build Tools",
        ),
    ] = True,
)

🦀 Installs rustup and a rust toolchain

Source code in Broken/__main__.py
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
@staticmethod
def rust(
    toolchain:   Annotated[str,  Option("--toolchain",   "-t", help="(Any    ) Rust toolchain to use (stable, nightly)")]="stable",
    build_tools: Annotated[bool, Option("--build-tools", "-b", help="(Windows) Install Visual C++ Build Tools")]=True,
):
    """🦀 Installs rustup and a rust toolchain"""
    import requests

    # Actions has its own workflow setup
    if (Runtime.GitHub):
        return

    # Install rustup based on platform
    if not shutil.which("rustup"):
        log.info("Rustup wasn't found, will install it")

        if BrokenPlatform.OnWindows:
            shell("winget", "install", "-e", "--id", "Rustlang.Rustup")
        elif BrokenPlatform.OnUnix:
            shell("sh", "-c", requests.get("https://sh.rustup.rs").text, "-y", echo=False)

        # If rustup isn't found, ask user to restart shell
        BrokenPath.add_to_path(Path.home()/".cargo"/"bin")

        if not BrokenPath.which("rustup"):
            log.warning("Rustup was likely installed but wasn't found adding '~/.cargo/bin' to Path")
            log.warning("• Maybe you changed the CARGO_HOME or RUSTUP_HOME environment variables")
            log.warning("• Please restart your shell for Rust toolchain to be on PATH")
            exit(0)

    # Install Visual C++ Build Tools on Windows
    if (BrokenPlatform.OnWindows and build_tools):
        log.warning("You must have Microsoft Visual C++ Build Tools installed to compile Rust projects")
        log.warning("• Broken will try installing it, you might need to restart your shell afterwards")
        shell("winget", "install", "-e", "--id", "Microsoft.VisualStudio.2022.BuildTools", "--override", (
            " --add Microsoft.VisualStudio.Component.VC.Tools.x86.x64"
            " --add Microsoft.VisualStudio.Component.Windows10SDK"
            " --add Microsoft.VisualStudio.Component.Windows11SDK.22000"
            "--wait --passive"
        ))

    class RustToolchain(BrokenEnum):
        Stable  = "stable"
        Nightly = "nightly"

    toolchain = RustToolchain.get(toolchain).value

    # Install or select the correct toolchain
    for line in shell("rustup", "toolchain", "list", output=True, echo=False).split("\n"):
        if ("no installed" in line) or (("default" in line) and (line.split("-")[0] != toolchain)):
            log.info(f"Defaulting Rust toolchain to ({toolchain})")
            shell("rustup", "default", toolchain)
    else:
        log.info(f"Rust toolchain is already the default ({toolchain})")

clean

clean() -> None

🧹 Remove pycaches, common blob directories

Source code in Broken/__main__.py
696
697
698
699
700
701
702
703
704
705
706
def clean(self) -> None:
    """🧹 Remove pycaches, common blob directories"""
    root = BROKEN.DIRECTORIES.REPOSITORY

    for path in root.rglob("__pycache__"):
        BrokenPath.remove(path)

    # Fixed known blob directories
    BrokenPath.remove(BROKEN.DIRECTORIES.REPO_RELEASES)
    BrokenPath.remove(BROKEN.DIRECTORIES.REPO_BUILD)
    BrokenPath.remove(root/".cache")

sync

sync() -> None

♻️ Synchronize common resources files across all projects

Source code in Broken/__main__.py
708
709
710
711
712
713
714
715
716
717
718
719
720
721
def sync(self) -> None:
    """♻️  Synchronize common resources files across all projects"""
    root = BROKEN.DIRECTORIES.REPOSITORY

    for project in self.projects[1:]:
        if (project.path/".github"/".nosync").exists():
            continue
        for file in flatten(
            ((root/".github").glob(ext) for ext in ("*.md", "*.yml")),
            (root/".github"/"ISSUE_TEMPLATE").glob("*.yml"),
            (root/".github"/"hatch_build.py"),
        ):
            target = project.path/file.relative_to(root)
            BrokenPath.copy(src=file, dst=target)

main

main()
Source code in Broken/__main__.py
725
726
727
def main():
    with BrokenProfiler("BROKEN"):
        BrokenManager().cli(*sys.argv[1:])