Skip to content

trading_agent.terminal.attachment_opener

Downloads a file the desk sent, or a picture a tool showed it, and opens it with the desktop's default program.

Files are saved under the state folder's "downloads" folder and opened with xdg-open, so a picture opens in the image viewer and a PDF in the PDF reader, as a new browser tab does on the web page.

Typical usage example:

opener = AttachmentOpener(gateway) path = await opener.open_attachment('general', attachment)

AttachmentOpener

AttachmentOpener(gateway, folder=None)

Saves desk files locally and opens them.

Creates the opener.

Parameters:

Name Type Description Default
gateway GatewayClient

The gateway, to download from.

required
folder Path | None

Where to save files, or None for the state folder's "downloads" folder.

None
Source code in src/trading_agent/terminal/attachment_opener.py
def __init__(
    self,
    gateway: gateway_client.GatewayClient,
    folder: Path | None = None,
):
    """Creates the opener.

    Args:
        gateway (gateway_client.GatewayClient): The gateway, to download from.
        folder (Path | None): Where to save files, or None for the state folder's "downloads" folder.
    """
    self._gateway = gateway
    if folder is None:
        folder = (
            terminal_settings.TerminalSettings.state_directory()
            / 'downloads'
        )
    self.folder = folder

save_attachment async

save_attachment(desk_name, attachment)

Downloads an attachment.

Parameters:

Name Type Description Default
desk_name str

The desk it belongs to.

required
attachment dict[str, Any]

The attachment, with attachment_id and file_name.

required

Returns:

Type Description
Path

Where it was saved.

Raises:

Type Description
GatewayError

The gateway could not send it.

OSError

The file could not be written.

Source code in src/trading_agent/terminal/attachment_opener.py
async def save_attachment(
    self, desk_name: str, attachment: dict[str, Any]
) -> Path:
    """Downloads an attachment.

    Args:
        desk_name (str): The desk it belongs to.
        attachment (dict[str, Any]): The attachment, with attachment_id and file_name.

    Returns:
        Path: Where it was saved.

    Raises:
        gateway_client.GatewayError: The gateway could not send it.
        OSError: The file could not be written.
    """
    data = await self._gateway.attachment_bytes(
        desk_name, attachment['attachment_id']
    )
    name = f'{attachment["attachment_id"]}-{attachment.get("file_name", "file")}'
    return self._write(name, data)

save_picture async

save_picture(desk_name, picture_path)

Downloads a picture from inside a desk's folder.

Parameters:

Name Type Description Default
desk_name str

The desk.

required
picture_path str

The picture's path relative to the desk folder.

required

Returns:

Type Description
Path

Where it was saved.

Raises:

Type Description
GatewayError

The gateway could not send it.

OSError

The file could not be written.

Source code in src/trading_agent/terminal/attachment_opener.py
async def save_picture(self, desk_name: str, picture_path: str) -> Path:
    """Downloads a picture from inside a desk's folder.

    Args:
        desk_name (str): The desk.
        picture_path (str): The picture's path relative to the desk folder.

    Returns:
        Path: Where it was saved.

    Raises:
        gateway_client.GatewayError: The gateway could not send it.
        OSError: The file could not be written.
    """
    data = await self._gateway.picture_bytes(desk_name, picture_path)
    return self._write(f'{desk_name}-{picture_path}', data)

open

open(path)

Opens a saved file with the desktop's default program, without waiting for it.

Parameters:

Name Type Description Default
path Path

The file.

required

Returns:

Type Description
bool

True when xdg-open was started, False when it is not installed.

Source code in src/trading_agent/terminal/attachment_opener.py
def open(self, path: Path) -> bool:
    """Opens a saved file with the desktop's default program, without waiting for it.

    Args:
        path (Path): The file.

    Returns:
        bool: True when xdg-open was started, False when it is not installed.
    """
    program = shutil.which('xdg-open')
    if program is None:
        return False
    subprocess.Popen(
        [
            program,
            str(path),
        ],
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    return True