Editor

elva.apps.editor

ELVA editor app.

Classes:

• LogStatus –

  Widget reflecting the state of logging.

• StoreStatus –

  Widget reflecting the state of storage.

• RendererStatus –

  Widget reflecting the state of rendering.

• ProviderStatus –

  Widget reflecting the state of a provider.

• UI –

  User interface.

Functions:

• get_provider –

  Get the provider from the message parameter in the merged configuration.

• encode_content –

  Encode data for inclusion in the displayed QR code for sharing.

• cli –

  Edit text documents collaboratively in real-time.

Attributes:

• LOG_LEVEL_MAP –

  Mapping of logging levels to their corresponding names.

• LANGUAGES –

  Supported languages.

LOG_LEVEL_MAP = dict(FATAL=logging.FATAL, ERROR=logging.ERROR, WARNING=logging.WARNING, INFO=logging.INFO, DEBUG=logging.DEBUG) module-attribute

Mapping of logging levels to their corresponding names.

LANGUAGES = {'py': 'python', 'md': 'markdown', 'sh': 'bash', 'js': 'javascript', 'rs': 'rust', 'yml': 'yaml'} module-attribute

Supported languages.

LogStatus(params, *args, config=None, rename=None, **kwargs)

Bases: FeatureStatus

Widget reflecting the state of logging.

Methods:

• apply –

  Apply the new state.

• on_button_pressed –

  Hook called on a button pressed event.

Attributes:

• is_ready (bool) –

  Flag whether all conditions for successful logging are fulfilled.

Source code in src/elva/widgets/status.py

def __init__(
    self,
    params: list[str],
    *args: tuple,
    config: None | dict[str, Any] = None,
    rename: None | dict[str, str] = None,
    **kwargs: dict,
):
    """
    Arguments:
        params: list of configuration parameters associated with this feature.
        config: mapping of configuration parameters to their values.
        rename: mapping of old parameter names to new parameter names.
        args: positional arguments passed to [`Button`][textual.widgets.Button].
        kwargs: keyword arguments passed to [`Button`][textual.widgets.Button].
    """
    super().__init__(*args, **kwargs)
    self.params = params
    self.rename = rename
    self.config = self.trim(config)

is_ready property

Flag whether all conditions for successful logging are fulfilled.

Returns:

• bool –

  True if all conditions are fulfilled, else False.

apply()

Apply the new state.

If is_ready returns True, logging gets reinitialized with updated parameters. Else, logging is stopped.

Source code in src/elva/apps/editor.py

def apply(self):
    """
    Apply the new state.

    If [`is_ready`][elva.apps.editor.LogStatus.is_ready] returns `True`, logging gets reinitialized with updated parameters.
    Else, logging is stopped.
    """
    if self.is_ready:
        c = self.config
        path = c.get("path")

        for handler in log.handlers[:]:
            log.removeHandler(handler)

        handler = logging.FileHandler(path)
        handler.setFormatter(DefaultFormatter())
        log.addHandler(handler)
        log.setLevel(c.get("level") or logging.INFO)
        self.variant = "success"
    else:
        for handler in log.handlers[:]:
            log.removeHandler(handler)
        self.variant = "default"

on_button_pressed(message)

Hook called on a button pressed event.

Basically, this method toggles the state of the logging feature.

Parameters:

• message (Message) –

  an object holding information about the button pressed event.

Source code in src/elva/apps/editor.py

def on_button_pressed(self, message: Message):
    """
    Hook called on a button pressed event.

    Basically, this method toggles the state of the logging feature.

    Arguments:
        message: an object holding information about the button pressed event.
    """
    if self.variant == "success":
        for handler in log.handlers[:]:
            log.removeHandler(handler)
        self.variant = "default"
    else:
        self.apply()

StoreStatus(yobject, params, *args, **kwargs)

Bases: ComponentStatus

Widget reflecting the state of storage.

Methods:

• on_worker_state_changed –

  Hook called on a worker state change event.

Attributes:

• component –

  Instance of the controlled SQLiteStore component.

• is_ready (bool) –

  Flag whether all conditions for successful storing are fulfilled.

Source code in src/elva/widgets/status.py

def __init__(
    self,
    yobject: Doc | Text | Array | Map,
    params: list[str],
    *args: tuple,
    **kwargs: dict,
):
    """
    Arguments:
        yobject: instance of the Y object being used by the internal component.
        params: list of configuration parameters associated with this component.
        args: positional arguments passed to [`FeatureStatus`][elva.widgets.status.FeatureStatus].
        kwargs: keyword arguments passed to [`FeatureStatus`][elva.widgets.status.FeatureStatus].
    """
    self.yobject = yobject
    super().__init__(params, *args, **kwargs)

component = SQLiteStore class-attribute instance-attribute

Instance of the controlled SQLiteStore component.

is_ready property

Flag whether all conditions for successful storing are fulfilled.

Returns:

• bool –

  True if all conditions are fulfilled, else False.

on_worker_state_changed(message)

Hook called on a worker state change event.

This method changes the status depending on the state of the worker the SQLiteStore component is running in.

Parameters:

• message (Message) –

  an object holding information about the worker state change event.

Source code in src/elva/apps/editor.py

def on_worker_state_changed(self, message: Message):
    """
    Hook called on a worker state change event.

    This method changes the status depending on the state of the worker the [`SQLiteStore`][elva.store.SQLiteStore] component is running in.

    Arguments:
        message: an object holding information about the worker state change event.
    """
    if message.worker.name == "component":
        match message.state:
            case WorkerState.RUNNING:
                self.variant = "success"
            case WorkerState.ERROR:
                self.variant = "error"
                self.control = None
            case WorkerState.CANCELLED | WorkerState.SUCCESS:
                self.variant = "default"
                self.control = None

RendererStatus(yobject, params, *args, **kwargs)

Bases: ComponentStatus

Widget reflecting the state of rendering.

Methods:

• on_worker_state_changed –

  Hook called on a worker state change event.

Attributes:

• component –

  Instance of the controlled TextRenderer component.

• is_ready (bool) –

  Flag whether all conditions for successful rendering are fulfilled.

Source code in src/elva/widgets/status.py

def __init__(
    self,
    yobject: Doc | Text | Array | Map,
    params: list[str],
    *args: tuple,
    **kwargs: dict,
):
    """
    Arguments:
        yobject: instance of the Y object being used by the internal component.
        params: list of configuration parameters associated with this component.
        args: positional arguments passed to [`FeatureStatus`][elva.widgets.status.FeatureStatus].
        kwargs: keyword arguments passed to [`FeatureStatus`][elva.widgets.status.FeatureStatus].
    """
    self.yobject = yobject
    super().__init__(params, *args, **kwargs)

component = TextRenderer class-attribute instance-attribute

Instance of the controlled TextRenderer component.

is_ready property

Flag whether all conditions for successful rendering are fulfilled.

Returns:

• bool –

  True if all conditions are fulfilled, else False.

on_worker_state_changed(message)

Hook called on a worker state change event.

This method changes the status depending on the state of the worker the TextRenderer component is running in.

Parameters:

• message (Message) –

  an object holding information about the worker state change event.

Source code in src/elva/apps/editor.py

def on_worker_state_changed(self, message: Message):
    """
    Hook called on a worker state change event.

    This method changes the status depending on the state of the worker the [`TextRenderer`][elva.renderer.TextRenderer] component is running in.

    Arguments:
        message: an object holding information about the worker state change event.
    """
    if message.worker.name == "component":
        match message.state:
            case WorkerState.RUNNING:
                self.variant = "success"
            case WorkerState.ERROR:
                self.variant = "error"
                self.control = None
            case WorkerState.CANCELLED | WorkerState.SUCCESS:
                self.variant = "default"
                self.control = None

ProviderStatus(yobject, params, *args, **kwargs)

Bases: ComponentStatus

Widget reflecting the state of a provider.

Methods:

• on_worker_state_changed –

  Hook called on a worker state change event.

• watch_connection_status –

  Watch the connection status of the controlled provider component.

Attributes:

• component –

  Instance of the controlled provider component.

• is_ready (bool) –

  Flag whether all conditions for successful connection are fulfilled.

Source code in src/elva/widgets/status.py

def __init__(
    self,
    yobject: Doc | Text | Array | Map,
    params: list[str],
    *args: tuple,
    **kwargs: dict,
):
    """
    Arguments:
        yobject: instance of the Y object being used by the internal component.
        params: list of configuration parameters associated with this component.
        args: positional arguments passed to [`FeatureStatus`][elva.widgets.status.FeatureStatus].
        kwargs: keyword arguments passed to [`FeatureStatus`][elva.widgets.status.FeatureStatus].
    """
    self.yobject = yobject
    super().__init__(params, *args, **kwargs)

component = get_provider class-attribute instance-attribute

Instance of the controlled provider component.

is_ready property

Flag whether all conditions for successful connection are fulfilled.

Returns:

• bool –

  True if all conditions are fulfilled, else False.

on_worker_state_changed(message)

Hook called on a worker state change event.

This method changes the status depending on the state of the worker the provider component is running in.

Parameters:

• message (Message) –

  an object holding information about the worker state change event.

Source code in src/elva/apps/editor.py

def on_worker_state_changed(self, message: Message):
    """
    Hook called on a worker state change event.

    This method changes the status depending on the state of the worker the provider component is running in.

    Arguments:
        message: an object holding information about the worker state change event.
    """
    if message.worker.name == "component":
        match message.state:
            case WorkerState.RUNNING:
                self.variant = "warning"
                self.run_worker(self.watch_connection_status(), group="event")
            case WorkerState.ERROR:
                self.variant = "error"
                self.control = None
            case WorkerState.CANCELLED | WorkerState.SUCCESS:
                self.variant = "default"
                self.control = None

watch_connection_status() async

Watch the connection status of the controlled provider component.

This method listens for connection or disconnection events and update the status accordingly.

Source code in src/elva/apps/editor.py

async def watch_connection_status(self):
    """
    Watch the connection status of the controlled provider component.

    This method listens for connection or disconnection events and update the status accordingly.
    """
    while True:
        await self.control.connected.wait()
        self.variant = "success"

        await self.control.disconnected.wait()
        self.variant = "warning"

UI(config)

Bases: App

User interface.

Parameters:

• config (dict) –

  mapping of configuration parameters to their values.

Methods:

• on_config_panel_applied –

  Hook called on an applied event from the config panel.

• on_exception –

  Hook called on an exception raised in the provider component.

• on_mount –

  Hook called on mounting the app.

• compose –

  Hook arranging child widgets.

• on_button_pressed –

  Hook called on a pressed button event.

• on_config_view_saved –

  Hook called on a Saved event from a config view.

• on_config_view_changed –

  Hook called on a Changed event from a config view.

• update_qrcode –

  Update the QR code for sharing.

• action_render –

  Hook called on an invoked render action.

• action_save –

  Hook called on an invoked save action.

Attributes:

• CSS_PATH –

  Path to the used textual CSS file.

• BINDINGS –

  Key bindings for actions of the app.

• language (str) –

  The language the text document is written in.

Source code in src/elva/apps/editor.py

def __init__(self, config: dict):
    """
    Arguments:
        config: mapping of configuration parameters to their values.
    """
    self.config = c = config

    ansi_color = c.get("ansi_color")
    super().__init__(ansi_color=ansi_color if ansi_color is not None else False)

    identifier = c.get("identifier")
    server = c.get("server")
    messages = c.get("messages")

    qr = (
        encode_content(
            dict(identifier=identifier, server=server, messages=messages)
        )
        if all(map(lambda x: x is not None, [identifier, server, messages]))
        else None
    )

    c["qr"] = qr

    # document structure
    self.ydoc = Doc()
    self.ytext = Text()
    self.ydoc["ytext"] = self.ytext

    self._language = c.get("language")

CSS_PATH = 'editor.tcss' class-attribute instance-attribute

Path to the used textual CSS file.

BINDINGS = [Binding('ctrl+s', 'save'), Binding('ctrl+r', 'render')] class-attribute instance-attribute

Key bindings for actions of the app.

language property

The language the text document is written in.

on_config_panel_applied(message) async

Hook called on an applied event from the config panel.

This method transfers the new configuration to the status widgets.

Parameters:

• message (Message) –

  an object holding information about the applied event.

Source code in src/elva/apps/editor.py

async def on_config_panel_applied(self, message: Message):
    """
    Hook called on an applied event from the config panel.

    This method transfers the new configuration to the status widgets.

    Arguments:
        message: an object holding information about the applied event.
    """
    for status_id in ["#provider", "#store", "#renderer", "#logger"]:
        self.query_one(status_id).update(message.config)

on_exception(exc) async

Hook called on an exception raised in the provider component.

This method opens the config panel and updates the provider status widget to signal the issue to the user.

Parameters:

• exc (Exception) –

  the exception that the provide component raised.

Raises:

• exc –

  the exception that was raised by the provider component.

Source code in src/elva/apps/editor.py

async def on_exception(self, exc: Exception) -> Exception:
    """
    Hook called on an exception raised in the provider component.

    This method opens the config panel and updates the provider status widget to signal the issue to the user.

    Arguments:
        exc: the exception that the provide component raised.

    Raises:
        exc: the exception that was raised by the provider component.
    """
    if isinstance(exc, InvalidStatus) and exc.response.status_code == 401:
        self.query_one(ConfigPanel).remove_class("hidden")
        self.query_one("#provider").variant = "error"

    # reraise to break connection loop
    raise exc

on_mount() async

Hook called on mounting the app.

Source code in src/elva/apps/editor.py

async def on_mount(self):
    """
    Hook called on mounting the app.
    """
    self.query_one(ConfigPanel).add_class("hidden")
    self.query_one(StoreStatus).disabled = True
    self.query_one(RendererStatus).disabled = True

compose()

Hook arranging child widgets.

Source code in src/elva/apps/editor.py

def compose(self):
    """
    Hook arranging child widgets.
    """
    c = self.config

    yield YTextArea(
        self.ytext,
        tab_behavior="indent",
        show_line_numbers=True,
        id="editor",
        language=self.language,
    )

    yield ConfigPanel(
        [
            QRCodeView(
                c.get("qr"),
                name="share",
                id="view-share",
            ),
            TextInputView(
                value=c.get("identifier"),
                name="identifier",
                id="view-identifier",
            ),
            URLInputView(
                value=c.get("server"),
                name="server",
                id="view-server",
                validators=WebsocketsURLValidator(),
                validate_on=["changed"],
            ),
            RadioSelectView(
                list(zip(["yjs", "elva"], ["yjs", "elva"])),
                value=c.get("messages") or "yjs",
                name="messages",
                id="view-messages",
            ),
            TextInputView(
                value=c.get("name"),
                name="name",
            ),
            TextInputView(
                value=c.get("user"),
                name="user",
                id="view-user",
            ),
            TextInputView(
                value=c.get("password"),
                name="password",
                password=True,
                id="view-password",
            ),
            PathInputView(
                value=c.get("file"),
                suggester=PathSuggester(),
                name="file_path",
            ),
            PathInputView(
                value=c.get("render"),
                suggester=PathSuggester(),
                name="render_path",
            ),
            SwitchView(
                value=c.get("auto_render"),
                name="auto_render",
                animate=False,
            ),
            PathInputView(
                value=c.get("log"),
                suggester=PathSuggester(),
                name="log",
            ),
            RadioSelectView(
                list(LOG_LEVEL_MAP.items()),
                value=c.get("level") or LOG_LEVEL_MAP["INFO"],
                name="level",
            ),
        ],
        label="X - Cut, C - Copy, S - Save",
    )

    with StatusBar():
        yield Button("=", id="config")
        yield ProviderStatus(
            self.ydoc,
            [
                "identifier",
                "server",
                "messages",
                "user",
                "password",
            ],
            "P",
            config=self.config,
            id="provider",
        )
        yield StoreStatus(
            self.ydoc,
            [
                "identifier",
                "file",
            ],
            "S",
            config=self.config,
            rename={"file": "path"},
            id="store",
        )
        yield RendererStatus(
            self.ytext,
            [
                "render",
                "auto_render",
            ],
            "R",
            config=self.config,
            rename={"render": "path"},
            id="renderer",
        )
        yield LogStatus(
            [
                "log",
                "level",
            ],
            "L",
            config=self.config,
            rename={"log": "path"},
            id="logger",
        )

on_button_pressed(message)

Hook called on a pressed button event.

This methods toggles the visibility of the config panel.

Parameters:

• message (Message) –

  an object holding information about the button pressed event.

Source code in src/elva/apps/editor.py

def on_button_pressed(self, message: Message):
    """
    Hook called on a pressed button event.

    This methods toggles the visibility of the config panel.

    Arguments:
        message: an object holding information about the button pressed event.
    """
    button = message.button
    match button.id:
        case "config":
            self.query_one(ConfigPanel).toggle_class("hidden")

on_config_view_saved(message)

Hook called on a Saved event from a config view.

This methods saves the config view's key-value-pair as metadata entry to the ELVA SQLite database file.

Parameters:

• message (Message) –

  an object holding information about the button pressed event.

Source code in src/elva/apps/editor.py

def on_config_view_saved(self, message: Message):
    """
    Hook called on a Saved event from a config view.

    This methods saves the config view's key-value-pair as metadata entry to the ELVA SQLite database file.

    Arguments:
        message: an object holding information about the button pressed event.
    """
    c = self.query_one(ConfigPanel).state
    file_path = c.get("file")
    if file_path is not None and file_path.suffix and not file_path.is_dir():
        SQLiteStore.set_metadata(file_path, {message.name: message.value})

on_config_view_changed(message)

Hook called on a Changed event from a config view.

This methods keeps the QR Code updated.

Parameters:

• message (Message) –

  an object holding information about the button pressed event.

Source code in src/elva/apps/editor.py

def on_config_view_changed(self, message: Message):
    """
    Hook called on a [`Changed`][elva.widgets.config.ConfigView.Changed] event from a config view.

    This methods keeps the QR Code updated.

    Arguments:
        message: an object holding information about the button pressed event.
    """
    if message.name in ["identifier", "server", "messages"]:
        self.update_qrcode()

update_qrcode()

Update the QR code for sharing.

This method queries the current values from the identifier, server and message views, encodes them and updates the QR Code with that.

Source code in src/elva/apps/editor.py

def update_qrcode(self):
    """
    Update the QR code for sharing.

    This method queries the current values from the identifier, server and message views, encodes them and updates the QR Code with that.
    """
    identifier = self.query_one("#view-identifier").value
    server = self.query_one("#view-server").value
    messages = self.query_one("#view-messages").value

    content = encode_content(
        dict(identifier=identifier, server=server, messages=messages)
    )
    self.query_one("#view-share").value = content

action_render()

Hook called on an invoked render action.

If a renderer component is running, its write method is called. Otherwise, the config panel opens to let the user fill in missing information.

Source code in src/elva/apps/editor.py

def action_render(self):
    """
    Hook called on an invoked render action.

    If a renderer component is running, its `write` method is called.
    Otherwise, the config panel opens to let the user fill in missing information.
    """
    renderer = self.query_one("#renderer").control
    if renderer is not None:
        self.run_worker(renderer.write())
    else:
        self.query_one(ConfigPanel).remove_class("hidden")

action_save()

Hook called on an invoked save action.

If a store component is running, nothing happens - everything gets written automatically. Otherwise, the config panel opens to let the user fill in missing information.

Source code in src/elva/apps/editor.py

def action_save(self):
    """
    Hook called on an invoked save action.

    If a store component is running, nothing happens - everything gets written automatically.
    Otherwise, the config panel opens to let the user fill in missing information.
    """
    store = self.query_one("#store").control
    if store is None:
        self.query_one(ConfigPanel).remove_class("hidden")

get_provider(self, ydoc, **config)

Get the provider from the message parameter in the merged configuration.

Parameters:

• ydoc (Doc) –

  instance of a Y Document passed to the provider class.

• config (dict, default: {} ) –

  merged configuration parameters.

Returns:

• WebsocketProvider | ElvaWebsocketProvider –

  instance of a provider.

Source code in src/elva/apps/editor.py

def get_provider(
    self, ydoc: Doc, **config: dict
) -> WebsocketProvider | ElvaWebsocketProvider:
    """
    Get the provider from the message parameter in the merged configuration.

    Arguments:
        ydoc: instance of a Y Document passed to the provider class.
        config: merged configuration parameters.

    Returns:
        instance of a provider.
    """
    try:
        msg = config.pop("messages")
    except KeyError:
        msg = None

    match msg:
        case "yjs" | None:
            provider = WebsocketProvider(ydoc, **config)
        case "elva":
            provider = ElvaWebsocketProvider(ydoc, **config)

    return provider

encode_content(data)

Encode data for inclusion in the displayed QR code for sharing.

Currently, the data are written out in TOML syntax.

Parameters:

• data (dict) –

  mapping to encode.

Returns:

• str –

  data encoded in TOML syntax.

Source code in src/elva/apps/editor.py

def encode_content(data: dict) -> str:
    """
    Encode data for inclusion in the displayed QR code for sharing.

    Currently, the data are written out in TOML syntax.

    Arguments:
        data: mapping to encode.

    Returns:
        data encoded in TOML syntax.
    """
    return tomli_w.dumps(data)

cli(ctx, auto_render, apply, ansi_color, file)

Edit text documents collaboratively in real-time. 

Parameters:

• ctx (Context) –

  the click context holding the configuration parameter mapping.

• auto_render (bool) –

  flag whether to render on closing.

• apply (bool) –

  flag whether to mark the configuration as applied.

• ansi_color (bool) –

  flag whether to use the terminal's ANSI color codes.

• file (None | Path) –

  path to the ELVA SQLite database file.

Source code in src/elva/apps/editor.py

@click.command()
@click.option(
    "--auto-render",
    "auto_render",
    is_flag=True,
    help="Enable rendering of the data file.",
)
@click.option(
    "--apply",
    "apply",
    is_flag=True,
    help="Apply the config on startup.",
)
@click.option(
    "--ansi-color",
    "ansi_color",
    is_flag=True,
    help="Use the terminal ANSI colors for the Textual colortheme.",
)
@click.argument(
    "file",
    required=False,
    type=click.Path(path_type=Path, dir_okay=False),
)
@click.pass_context
def cli(
    ctx: click.Context,
    auto_render: bool,
    apply: bool,
    ansi_color: bool,
    file: None | Path,
):
    """
    Edit text documents collaboratively in real-time.
    \f

    Arguments:
        ctx: the click context holding the configuration parameter mapping.
        auto_render: flag whether to render on closing.
        apply: flag whether to mark the configuration as applied.
        ansi_color: flag whether to use the terminal's ANSI color codes.
        file: path to the ELVA SQLite database file.
    """

    c = ctx.obj

    # gather info
    gather_context_information(ctx, file=file, app="editor")

    if c.get("auto_render") is None or auto_render:
        # the flag has been explicitly set by the user
        c["auto_render"] = auto_render
    c["apply"] = apply
    c["ansi_color"] = ansi_color

    # logging
    LOGGER_NAME.set(__name__)
    log_path = c.get("log")
    level = c.get("level")
    if level is not None and log_path is not None:
        handler = logging.FileHandler(log_path)
        handler.setFormatter(DefaultFormatter())
        log.addHandler(handler)
        log.setLevel(level)

    # run app
    ui = UI(c)
    ui.run()