Skip to content

trading_agent.terminal.screens.channels_screen

The Channels page: the Telegram channel's state, the chats waiting to be paired, and the paired chats.

It copies the web page's ChannelsPage.tsx: a waiting pairing is approved with a, a code can be typed, and a paired chat is revoked with Delete.

Typical usage example:

app.push_screen(ChannelsScreen('general'))

ChannelsScreen 

ChannelsScreen(desk_name)

Bases: PageScreen

The Channels page.

Source code in src/trading_agent/terminal/screens/page_screen.py 
32
33
34
35
36
37
38
39
def __init__(self, desk_name: str):
    """Creates the page.

    Args:
        desk_name (str): The desk the chat shows, which a desk's page starts on.
    """
    super().__init__(classes='page')
    self.desk_name = desk_name

compose_page 

compose_page()

Builds the status, the pairing tables and the code box.

Yields:

Type Description
Widget

The sections, in a scroll area.
Source code in src/trading_agent/terminal/screens/channels_screen.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def compose_page(self):
    """Builds the status, the pairing tables and the code box.

    Yields:
        Widget: The sections, in a scroll area.
    """
    with containers.VerticalScroll(classes='page-body'):
        yield widgets.Static(
            '', id='channels-status', classes='page-summary'
        )
        yield widgets.Static(
            '[b]Waiting to be paired[/b] [dim]a approves[/dim]',
            classes='page-heading',
        )
        yield widgets.DataTable(id='channels-pending', cursor_type='row')
        yield widgets.Input(
            placeholder='Or type the code the bot sent, then press enter',
            id='channels-code',
        )
        yield widgets.Static(
            '[b]Paired chats[/b] [dim]delete revokes[/dim]',
            classes='page-heading',
        )
        yield widgets.DataTable(id='channels-paired', cursor_type='row')

on_mount 

on_mount()

Sets up the tables, then reads the channel.

Source code in src/trading_agent/terminal/screens/channels_screen.py 
59
60
61
62
63
64
65
66
67
def on_mount(self) -> None:
    """Sets up the tables, then reads the channel."""
    self.table('channels-pending').add_columns(
        'Code', 'Name', 'Chat', 'Asked'
    )
    self.table('channels-paired').add_columns(
        'Name', 'Chat', 'Desk', 'Paired'
    )
    super().on_mount()

refresh_page async 

refresh_page()

Reads the Telegram channel.

Source code in src/trading_agent/terminal/screens/channels_screen.py 
 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
@work(exclusive=True, group='page')
async def refresh_page(self) -> None:
    """Reads the Telegram channel."""
    try:
        telegram = await self.app.gateway.telegram()
    except gateway_client.GatewayError as error:
        self.problem(error.message)
        return
    clock = formatter.Formatter()
    state = str(telegram.get('state', ''))
    if telegram.get('error'):
        state += f' ({telegram["error"]})'
    self.query_one('#channels-status', widgets.Static).update(
        Content.from_markup(
            'Telegram [b]$enabled[/b] · token $token · state [b]$state[/b] · @$name · default desk $desk\n[dim]Notifications: $notifications[/dim]',
            enabled='on' if telegram.get('enabled') else 'off',
            token='set' if telegram.get('token_present') else 'missing',
            state=state,
            name=str(telegram.get('username') or '—'),
            desk=str(telegram.get('default_desk', '')),
            notifications=', '.join(
                str(item) for item in telegram.get('notifications') or []
            ),
        )
    )
    pending = self.table('channels-pending')
    pending.clear()
    for item in telegram.get('pending') or []:
        pending.add_row(
            str(item.get('code', '')),
            str(item.get('name', '')),
            str(item.get('chat_id', '')),
            clock.day_time(
                str(
                    item.get('requested_at') or item.get('created_at') or ''
                )
            ),
            key=str(item.get('code', '')),
        )
    paired = self.table('channels-paired')
    paired.clear()
    for chat in telegram.get('paired') or []:
        paired.add_row(
            str(chat.get('name', '')),
            str(chat.get('chat_id', '')),
            str(chat.get('desk', '')),
            clock.day_time(str(chat.get('paired_at', ''))),
            key=str(chat.get('chat_id', '')),
        )

action_approve 

action_approve()

Approves the highlighted pairing.

Source code in src/trading_agent/terminal/screens/channels_screen.py 
119
120
121
122
123
def action_approve(self) -> None:
    """Approves the highlighted pairing."""
    code = self.selected_key('channels-pending')
    if code is not None:
        self._approve(code)

on_input_submitted 

on_input_submitted(event)

Approves a typed code.

Parameters:

Name Type Description Default
event Submitted

The code.

 required
Source code in src/trading_agent/terminal/screens/channels_screen.py 
125
126
127
128
129
130
131
132
133
134
def on_input_submitted(self, event: widgets.Input.Submitted) -> None:
    """Approves a typed code.

    Args:
        event (widgets.Input.Submitted): The code.
    """
    event.stop()
    if event.value.strip():
        self._approve(event.value.strip())
        event.input.value = ''

action_revoke 

action_revoke()

Revokes the highlighted paired chat.

Source code in src/trading_agent/terminal/screens/channels_screen.py 
136
137
138
139
140
141
142
def action_revoke(self) -> None:
    """Revokes the highlighted paired chat."""
    chat_id = self.selected_key('channels-paired')
    if chat_id is not None and self.focused is self.table(
        'channels-paired'
    ):
        self._revoke(int(chat_id))