trading_agent.devices.device_store

The phones that receive push notifications, kept in one JSON file.

The Android app registers its Firebase Cloud Messaging token after signing in, with the gateway's access token, so only the owner's phones are added. The user can remove a phone on the Settings page. The file sits in workspace/shared, which the desks can read but never write.

Typical usage example:

store = DeviceStore(layout.phone_devices_file) phone = store.register(firebase_token, 'Galaxy M53') store.remove(phone.device_id)

DeviceError

Bases: ValueError

A phone that is not registered, or one more phone than the store keeps.

PhoneDevice dataclass

PhoneDevice(device_id, token, name, registered_at)

One phone that receives notifications.

Attributes:

Name	Type	Description
device_id	str	A short random id for the phone, used to remove it.
token	str	The Firebase Cloud Messaging token that addresses the app on this phone.
name	str	The phone's name as the app reports it, such as "Samsung SM-M536B".
registered_at	str	When it was first registered, as an ISO 8601 string in UTC.

DeviceStore

DeviceStore(path, clock=None)

The registered phones, kept in one JSON file.

Creates the store without reading the file.

Parameters:

Name	Type	Description	Default
path	Path	The JSON file.	required
clock	Callable[[], datetime] | None	Returns the current time in UTC; None uses the system clock.	None

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

def __init__(
    self, path: Path, clock: Callable[[], datetime.datetime] | None = None
):
    """Creates the store without reading the file.

    Args:
        path (Path): The JSON file.
        clock (Callable[[], datetime.datetime] | None): Returns the current time in UTC; None uses the system clock.
    """
    self._path = path
    self._clock = clock
    self._lock = threading.Lock()

register

register(token, name)

Adds a phone, or renames it when its token is already known.

Parameters:

Name	Type	Description	Default
token	str	The phone's Firebase Cloud Messaging token.	required
name	str	The phone's name.	required

Returns:

Type	Description
PhoneDevice	The phone as stored.

Raises:

Type	Description
DeviceError	The token is empty, or ten phones are already registered.

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

def register(self, token: str, name: str) -> PhoneDevice:
    """Adds a phone, or renames it when its token is already known.

    Args:
        token (str): The phone's Firebase Cloud Messaging token.
        name (str): The phone's name.

    Returns:
        PhoneDevice: The phone as stored.

    Raises:
        DeviceError: The token is empty, or ten phones are already registered.
    """
    token = token.strip()
    name = name.strip()[:_LONGEST_NAME] or 'Android phone'
    if not token:
        raise DeviceError('A phone needs its Firebase token to register')
    with self._lock:
        devices = self._read()
        for position, device in enumerate(devices):
            if device.token == token:
                renamed = dataclasses.replace(device, name=name)
                devices[position] = renamed
                self._write(devices)
                return renamed
        if len(devices) >= _MOST_DEVICES:
            raise DeviceError(
                f'{_MOST_DEVICES} phones are already registered; remove one on the Settings page first'
            )
        device = PhoneDevice(
            device_id=secrets.token_hex(4),
            token=token,
            name=name,
            registered_at=self._now().isoformat(timespec='seconds'),
        )
        devices.append(device)
        self._write(devices)
        return device

remove

remove(device_id)

Removes a phone, so it gets no more notifications.

Parameters:

Name	Type	Description	Default
device_id	str	The phone's id.	required

Returns:

Type	Description
PhoneDevice	The phone that was removed.

Raises:

Type	Description
DeviceError	No phone has that id.

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

def remove(self, device_id: str) -> PhoneDevice:
    """Removes a phone, so it gets no more notifications.

    Args:
        device_id (str): The phone's id.

    Returns:
        PhoneDevice: The phone that was removed.

    Raises:
        DeviceError: No phone has that id.
    """
    with self._lock:
        devices = self._read()
        for device in devices:
            if device.device_id == device_id:
                devices.remove(device)
                self._write(devices)
                return device
    raise DeviceError(f'No phone with id {device_id!r} is registered')

forget_token

forget_token(token)

Removes the phone with a token Firebase says is no longer valid, such as after the app was uninstalled.

Parameters:

Name	Type	Description	Default
token	str	The token.	required

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

def forget_token(self, token: str) -> None:
    """Removes the phone with a token Firebase says is no longer valid, such as after the app was uninstalled.

    Args:
        token (str): The token.
    """
    with self._lock:
        devices = self._read()
        kept = []
        for device in devices:
            if device.token != token:
                kept.append(device)
        if len(kept) != len(devices):
            self._write(kept)

devices

devices()

Every registered phone.

Returns:

Type	Description
list[PhoneDevice]	The phones, in the order they were registered.

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

def devices(self) -> list[PhoneDevice]:
    """Every registered phone.

    Returns:
        list[PhoneDevice]: The phones, in the order they were registered.
    """
    with self._lock:
        return self._read()