trading_agent.devices.push_channel

Pushes the events worth a phone notification to every phone running the Android app.

It listens to the event bus like the Telegram channel does and words most notifications with the same NotificationFormatter, so a fill reads the same on Telegram and on the phone. The user chooses the kinds in trading_agent.toml's [push] notifications list. Requests that need the user (a permission, a question or a plan) are pushed too, so the user can answer them in the app before they lapse.

Each notification carries the desk's name, so tapping it opens that desk's chat.

Typical usage example:

channel = PushChannel(store, sender, formatter, event_bus, ['fills', 'requests']) await channel.run()

NotificationSender

Bases: Protocol

Anything that can send one notification to one phone, such as FirebaseSender.

send async

send(token, title, body, data)

Sends one notification.

Parameters:

Name	Type	Description	Default
token	str	The phone's token.	required
title	str	The title.	required
body	str	The text.	required
data	dict[str, str]	Values the app reads when the notification is tapped.	required

Returns:

Type	Description
str	"sent", "unregistered" or "failed".

Source code in src/trading_agent/devices/push_channel.py

async def send(
    self,
    token: str,
    title: str,
    body: str,
    data: dict[str, str],
) -> str:
    """Sends one notification.

    Args:
        token (str): The phone's token.
        title (str): The title.
        body (str): The text.
        data (dict[str, str]): Values the app reads when the notification is tapped.

    Returns:
        str: "sent", "unregistered" or "failed".
    """

PushChannel

PushChannel(store, sender, formatter, event_bus, notifications)

The worker that turns events into phone notifications.

Subscribes to the bus without sending anything yet.

Parameters:

Name	Type	Description	Default
store	DeviceStore	The registered phones.	required
sender	NotificationSender	Sends to one phone.	required
formatter	NotificationFormatter	Words the events.	required
event_bus	EventBus	The bus to listen to.	required
notifications	Sequence[str]	The kinds to push, from [push] notifications; "requests" pushes questions, plans and permission requests.	required

Source code in src/trading_agent/devices/push_channel.py

def __init__(
    self,
    store: device_store_module.DeviceStore,
    sender: NotificationSender,
    formatter: notification_formatter_module.NotificationFormatter,
    event_bus: event_bus_module.EventBus,
    notifications: Sequence[str],
):
    """Subscribes to the bus without sending anything yet.

    Args:
        store (device_store_module.DeviceStore): The registered phones.
        sender (NotificationSender): Sends to one phone.
        formatter (notification_formatter_module.NotificationFormatter): Words the events.
        event_bus (event_bus_module.EventBus): The bus to listen to.
        notifications (Sequence[str]): The kinds to push, from [push] notifications; "requests" pushes questions, plans and permission requests.
    """
    self._store = store
    self._sender = sender
    self._formatter = formatter
    self._subscription = event_bus.subscribe(maximum_queued=_QUEUE_SIZE)
    self._push_requests = 'requests' in notifications
    self._trigger_sources: dict[str, tuple[str, str]] = {}

run async

run()

Pushes notifications as events arrive, until cancelled.

Source code in src/trading_agent/devices/push_channel.py

async def run(self) -> None:
    """Pushes notifications as events arrive, until cancelled."""
    while True:
        published_event = await self._subscription.next_event()
        try:
            await self.handle(published_event)
        except Exception:
            _LOGGER.exception('Could not push an event to the phones')

handle async

handle(published_event)

Pushes one event if it is worth a notification.

Parameters:

Name	Type	Description	Default
published_event	Event	The event.	required

Source code in src/trading_agent/devices/push_channel.py

async def handle(self, published_event: event_module.Event) -> None:
    """Pushes one event if it is worth a notification.

    Args:
        published_event (event_module.Event): The event.
    """
    kind = published_event.kind
    payload = published_event.payload
    if kind == event_module.EventKind.TEXT_DELTA:
        return
    if kind == event_module.EventKind.TRIGGER_RECEIVED:
        self._remember_trigger(payload)
        return
    if kind == event_module.EventKind.USER_REQUEST:
        if self._push_requests:
            await self._push_request(payload)
        return
    source, trigger_text = self._trigger_sources.get(
        str(payload.get('trigger_id')), (None, None)
    )
    text = self._formatter.format(published_event, source, trigger_text)
    if text is None:
        return
    desk = published_event.desk or ''
    await self.push(
        desk or _APP_TITLE,
        self._without_desk_prefix(desk, text),
        {
            'desk': desk,
            'kind': str(kind),
        },
    )

push async

push(title, body, data)

Sends one notification to every registered phone, forgetting phones Firebase no longer knows.

Parameters:

Name	Type	Description	Default
title	str	The title.	required
body	str	The text.	required
data	dict[str, str]	Values the app reads when the notification is tapped.	required

Returns:

Type	Description
int	How many phones it was sent to.

Source code in src/trading_agent/devices/push_channel.py

async def push(self, title: str, body: str, data: dict[str, str]) -> int:
    """Sends one notification to every registered phone, forgetting phones Firebase no longer knows.

    Args:
        title (str): The title.
        body (str): The text.
        data (dict[str, str]): Values the app reads when the notification is tapped.

    Returns:
        int: How many phones it was sent to.
    """
    sent = 0
    for device in self._store.devices():
        if await self.push_to(device, title, body, data):
            sent += 1
    return sent

push_to async

push_to(device, title, body, data)

Sends one notification to one phone, forgetting it when Firebase no longer knows it.

Parameters:

Name	Type	Description	Default
device	PhoneDevice	The phone.	required
title	str	The title.	required
body	str	The text.	required
data	dict[str, str]	Values the app reads when the notification is tapped.	required

Returns:

Type	Description
bool	True when Firebase accepted the notification.

Source code in src/trading_agent/devices/push_channel.py

async def push_to(
    self,
    device: device_store_module.PhoneDevice,
    title: str,
    body: str,
    data: dict[str, str],
) -> bool:
    """Sends one notification to one phone, forgetting it when Firebase no longer knows it.

    Args:
        device (device_store_module.PhoneDevice): The phone.
        title (str): The title.
        body (str): The text.
        data (dict[str, str]): Values the app reads when the notification is tapped.

    Returns:
        bool: True when Firebase accepted the notification.
    """
    outcome = await self._sender.send(device.token, title, body, data)
    if outcome == _UNREGISTERED:
        _LOGGER.info(
            'Firebase no longer knows the phone %s; removing it',
            device.name,
        )
        self._store.forget_token(device.token)
    return outcome == _SENT