Merge pull request #173 from gdsfactory/cli

kf CLI

changelog.d/+d501e164.added.md

@@ -0,0 +1 @@
+Added a cli based on [typer](https://typer.tiangolo.com) to allow running of functions (taking int/float/str args) and allow upload/update of gdatasea edafiles

pyproject.toml

@@ -18,14 +18,15 @@ authors = [
     {name = "gdsfactory community", email = "[email protected]"},
 ]
 dependencies = [
-	"klayout >= 0.28.9.post2",
+	"klayout >= 0.28.10",
 	"scipy",
 	"ruamel.yaml",
 	"cachetools >= 5.2.0",
-  "pydantic >= 2.0b3",
-	"pydantic-settings >= 2.0b1",
+  "pydantic >= 2.0.2, < 3",
+	"pydantic-settings >= 2.0.1, < 3",
 	"loguru",
 	"tomli",
+	"typer[all]",
 ]
 
 [project.optional-dependencies]
@@ -42,6 +43,7 @@ dev = [
 	"python-lsp-ruff",
 	"types-cachetools",
 	"python-lsp-black",
+	"pytest",
 	"kfactory[git]",
 	"towncrier",
 	"tbump",
@@ -74,6 +76,8 @@ ipy = [
 	"ipyevents",
 ]
 
+[project.scripts]
+kf = "kfactory.cli:app"
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/kfactory/__init__.py

@@ -26,8 +26,6 @@
 from . import cells, placer, routing, utils, port, pdk, technology
 from .conf import config
 
-# from .pdk import Pdk
-
 __version__ = "0.8.4"
 
 logger = config.logger

src/kfactory/cli/__init__.py

@@ -0,0 +1,31 @@
+"""CLI interface for kfactory.
+
+Use `kf --help` for more info.
+"""
+from typing import Annotated
+
+# import click
+import typer
+
+from .. import __version__
+from .runshow import run, show
+from .sea import app as sea
+
+app = typer.Typer(name="kf")
+
+
+app.command()(run)
+app.command()(show)
+app.add_typer(sea)
+
+
+@app.callback(invoke_without_command=True)
+def version_callback(
+    version: Annotated[
+        bool, typer.Option("--version", "-V", help="Show version of the CLI")
+    ] = False,
+) -> None:
+    """Show the version of the cli."""
+    if version:
+        print(f"KFactory CLI Version: {__version__}")
+        raise typer.Exit()

src/kfactory/cli/runshow.py

@@ -0,0 +1,110 @@
+"""CLI interface for kfactory.
+
+Use `kf --help` for more info.
+"""
+import importlib
+import os
+import runpy
+import sys
+from enum import Enum
+from pathlib import Path
+from typing import Annotated, Optional
+
+import typer
+
+from ..conf import logger
+from ..kcell import KCell
+from ..kcell import show as kfshow
+
+# app = typer.Typer(name="show")
+# show = typer.Typer(name="show")
+# run = typer.Typer(name="run")
+
+
+class RunType(str, Enum):
+    file = "file"
+    module = "module"
+    function = "function"
+
+
+def show(file: str) -> None:
+    """Show a GDS or OAS file in KLayout through klive."""
+    path = Path(file)
+    logger.debug("Path = {}", path.resolve())
+    if not path.exists():
+        logger.critical("{file} does not exist, exiting", file=file)
+        return
+    if not path.is_file():
+        logger.critical("{file} is not a file, exiting", file=file)
+        return
+    if not os.access(path, os.R_OK):
+        logger.critical("No permission to read file {file}, exiting", file=file)
+        return
+    kfshow(path)
+
+
+def run(
+    file: Annotated[
+        str,
+        typer.Argument(default=..., help="The file|module|function to execute"),
+    ],
+    func_kwargs: Annotated[
+        Optional[list[str]],  # noqa: UP007
+        typer.Argument(
+            help="Arguments used for --type function."
+            " Doesn't have any influence for other types"
+        ),
+    ] = None,
+    type: Annotated[
+        RunType,
+        typer.Option(
+            help="Run a file or a module (`python -m <module_name>`) or a function"
+        ),
+    ] = RunType.file,
+    show: Annotated[
+        bool, typer.Option(help="Show the file through klive in KLayout")
+    ] = True,
+) -> None:
+    """Run a python modules __main__ or a function if specified."""
+    path = sys.path.copy()
+    sys.path.append(os.getcwd())
+    match type:
+        case RunType.file:
+            runpy.run_path(file, run_name="__main__")
+        case RunType.module:
+            runpy.run_module(file, run_name="__main__")
+        case RunType.function:
+            mod, func = file.rsplit(".", 1)
+            logger.debug(f"{mod=},{func=}")
+            try:
+                spec = importlib.util.find_spec(mod)
+                if spec is None or spec.loader is None:
+                    raise ImportError
+                _mod = importlib.util.module_from_spec(spec)
+                sys.modules[mod] = _mod
+                spec.loader.exec_module(_mod)
+                kwargs = {}
+
+                old_arg = ""
+                if func_kwargs is not None:
+                    for i, file in enumerate(func_kwargs):
+                        if i % 2:
+                            try:
+                                value: int | float | str = int(file)
+                            except ValueError:
+                                try:
+                                    value = float(file)
+                                except ValueError:
+                                    value = file
+                            kwargs[old_arg] = value
+                        else:
+                            old_arg = file
+
+                cell = getattr(_mod, func)(**kwargs)
+                if show and isinstance(cell, KCell):
+                    cell.show()
+            except ImportError:
+                logger.critical(
+                    f"Couldn't import function '{func}' from module '{mod}'"
+                )
+    sys.path = path

src/kfactory/cli/sea.py

@@ -0,0 +1,95 @@
+"""CLI interface for kfactory.
+
+Use `kf sea --help` for more info.
+"""
+import json
+from pathlib import Path
+from typing import Annotated, Optional
+
+import requests
+import rich
+import typer
+
+from ..kcell import KCLayout
+
+app = typer.Typer(name="sea")
+
+
+@app.command()
+def upload(
+    file: str,
+    name: Optional[str] = None,  # noqa: UP007
+    description: Optional[str] = None,  # noqa: UP007
+    base_url: Annotated[
+        str, typer.Option(envvar="GDATASEA_URL")
+    ] = "http://localhost:3131",
+) -> None:
+    """Upload a new eda file to gdatasea."""
+    if name is None:
+        kcl = KCLayout()
+        kcl.read(file)
+        assert len(kcl.top_cells()) > 0, (
+            "Cannot automatically determine name of gdatasea edafile if"
+            " there is no name given and the gds is empty"
+        )
+        name = kcl.top_cells()[0].name
+
+    url = f"{base_url}/edafile"
+    if description:
+        params = {"name": name, "description": description}
+    else:
+        params = {"name": name}
+
+    fp = Path(file)
+    assert fp.exists()
+    with open(fp, "rb") as f:
+        r = requests.post(url, params=params, files={"eda_file": f})
+        msg = f"Response from {url}: "
+        try:
+            msg += rich.pretty.pretty_repr(json.loads(r.text))
+            msg = msg.replace("'success': 200", "[green]success: 200[/green]").replace(
+                "422", "[red]422[/red]"
+            )
+        except json.JSONDecodeError:
+            msg += rich.pretty.pretty_repr(f"[red]{r.text}[/red]")
+        rich.print(msg)
+
+
+@app.command()
+def update(
+    file: str,
+    edafile_id: Annotated[int, typer.Option("--edafile_id", "--id", "-id")],
+    name: Optional[str] = None,  # noqa: UP007
+    description: Optional[str] = None,  # noqa: UP007
+    base_url: Annotated[
+        str, typer.Option(envvar="GDATASEA_URL")
+    ] = "http://localhost:3131",
+) -> None:
+    """Upload a new eda file to gdatasea."""
+    if name is None:
+        kcl = KCLayout()
+        kcl.read(file)
+        assert len(kcl.top_cells()) > 0, (
+            "Cannot automatically determine name of gdatasea edafile if"
+            " there is no name given and the gds is empty"
+        )
+        name = kcl.top_cells()[0].name
+
+    url = f"{base_url}/edafile/{edafile_id}"
+    params = {"name": name}
+    if description:
+        params["description"] = description
+
+    fp = Path(file)
+    assert fp.exists()
+    with open(fp, "rb") as f:
+        r = requests.put(url, params=params, files={"eda_file": f})
+        msg = f"Response from {url}: "
+        try:
+            msg += rich.pretty.pretty_repr(json.loads(r.text))
+            msg = msg.replace("'success': 200", "[green]success: 200[/green]").replace(
+                "422", "[red]422[/red]"
+            )
+        except json.JSONDecodeError:
+            msg += rich.pretty.pretty_repr(f"[red]{r.text}[/red]")
+        rich.print(msg)

src/kfactory/conf.py

@@ -6,6 +6,7 @@
 import re
 import sys
 import traceback
+from enum import Enum
 from itertools import takewhile
 from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
@@ -52,15 +53,23 @@ def tracing_formatter(record: loguru.Record) -> str:
         )
 
 
+class LogLevel(str, Enum):
+    TRACE = "TRACE"
+    DEBUG = "DEBUG"
+    INFO = "INFO"
+    SUCCESS = "SUCCESS"
+    WARNING = "WARNING"
+    ERROR = "ERROR"
+    CRITICAL = "CRITICAL"
+
+
 class LogFilter(BaseModel):
     """Filter certain messages by log level or regex.
 
     Filtered messages are not evaluated and discarded.
     """
 
-    level: Literal[
-        "TRACE", "DEBUG", "INFO", "SUCCESS", "WARNING", "ERROR", "CRITICAL"
-    ] = "INFO"
+    level: LogLevel = LogLevel.INFO
     regex: str | None = None
 
     def __call__(self, record: loguru.Record) -> bool:
@@ -103,7 +112,7 @@ def __init__(self, **data: Any):
         super().__init__(**data)
         self.logger.remove()
         self.logger.add(sys.stdout, format=tracing_formatter, filter=self.logfilter)
-        self.logger.info("LogLevel: {}", self.logfilter.level)
+        self.logger.debug("LogLevel: {}", self.logfilter.level)
 
     class Config:
         """Pydantic settings."""

src/kfactory/kcell.py

@@ -3378,7 +3378,7 @@ def show(
             delete = True
 
     elif isinstance(gds, str | Path):
-        gds_file = Path(gds)
+        gds_file = Path(gds).resolve()
     else:
         raise NotImplementedError(f"unknown type {type(gds)} for streaming to KLayout")
     if not gds_file.is_file():
@@ -3401,7 +3401,23 @@ def show(
         msg = ""
         try:
             msg = conn.recv(1024).decode("utf-8")
-            config.logger.info(f"Message from klive: {msg}")
+            try:
+                jmsg = json.loads(msg)
+                match jmsg["type"]:
+                    case "open":
+                        config.logger.info(
+                            "klive v{version}: Opened file '{file}'",
+                            version=jmsg["version"],
+                            file=jmsg["file"],
+                        )
+                    case "reload":
+                        config.logger.info(
+                            "klive v{version}: Reloaded file '{file}'",
+                            version=jmsg["version"],
+                            file=jmsg["file"],
+                        )
+            except json.JSONDecodeError:
+                config.logger.info(f"Message from klive: {msg}")
         except OSError:
             config.logger.warning("klive didn't send data, closing")
         finally: