import enum
import requests
import threading
from platypush.plugins import Plugin, action
[docs]
class InterruptionFilterPolicy(enum.Enum):
ALLOW_ALL = 1
PRIORITY_ONLY = 2
BLOCK_ALL = 3
ALARM_ONLY = 4
[docs]
class MobileJoinPlugin(Plugin):
"""
Control mobile devices and other devices linked to Join (https://joaoapps.com/join/api/).
It requires you to have either the Join app installed on an Android device or the Join
browser extension installed.
The ``device`` parameter in the actions can be:
- A device ID or name
- A list or comma-separated list of device IDs/names
- A group name or a list of group names
Supported groups:
- group.all
- group.android
- group.windows10
- group.phone
- group.tablet
- group.pc
"""
_base_url = 'https://joinjoaomgcd.appspot.com/_ah/api'
_push_url = '{}/messaging/v1/sendPush'.format(_base_url)
_devices_url = '{}/registration/v1/listDevices'.format(_base_url)
_groups = {
'group.all',
'group.android',
'group.windows10',
'group.phone',
'group.tablet',
'group.pc',
}
[docs]
def __init__(self, api_key: str, **kwargs):
"""
:param api_key: Join API key. Get your at https://joinjoaomgcd.appspot.com/.
"""
super().__init__(**kwargs)
self._api_key = api_key
self._devices = {}
self._devices_by_id = {}
self._devices_by_name = {}
self._devices_lock = threading.RLock()
def _send_request(self, url, device=None, params: dict = None, **kwargs):
if not params:
params = {}
params['apikey'] = self._api_key
if device:
params['deviceIds'] = self._get_device_ids(device)
response = requests.get(url, params=params, **kwargs)
if not response.ok:
response.raise_for_status()
response = response.json()
if response.get('userAuthError'):
raise PermissionError('Invalid api_key provided')
del response['userAuthError']
return response
def _init_devices(self):
with self._devices_lock:
if not self._devices:
self._devices = self.get_devices().output
self._devices_by_id = {dev['id']: dev for dev in self._devices}
self._devices_by_name = {dev['name']: dev for dev in self._devices}
def _get_device_ids(self, device):
devices = [dev.strip() for dev in device.split(',')] \
if isinstance(device, str) else device
assert isinstance(devices, list)
has_unknown_devices = True
cache_refreshed = False
device_ids = []
while has_unknown_devices:
has_unknown_devices = False
for dev in devices:
if dev in self._devices_by_id:
device_ids.append(self._devices_by_id[dev]['id'])
elif dev in self._devices_by_name:
device_ids.append(self._devices_by_name[dev]['id'])
elif dev in self._groups:
device_ids.append(dev)
else:
has_unknown_devices = True
if has_unknown_devices:
if cache_refreshed:
raise KeyError('No such device: {}'.format(device))
self._init_devices()
cache_refreshed = True
return device_ids
[docs]
@action
def push(self, device, text=None, url=None, file=None):
"""
Push a URL or file to one or more devices
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param text: Optional description text for the URL or file
:param url: URL to be pushed
:param file: A publicly accessible URL of a file. You can also send the url of a file on your personal
Google Drive
"""
assert url or file
params = {}
if text:
params['text'] = text
if url:
params['url'] = url
if file:
params['file'] = file
return self._send_request(self._push_url, device=device, params=params)
[docs]
@action
def set_clipboard(self, device, text):
"""
Write to the clipboard of a device
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param text: Text to be set
"""
return self._send_request(self._push_url, device=device, params={'clipboard': text})
[docs]
@action
def send_sms(self, device, text: str, number: str = None, contact_name: str = None):
"""
Send an sms through a mobile device connected to Join
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param text: Text to be sent
:param number: Phone number
:param contact_name: Alternatively to the phone number, you can specify a contact name
"""
assert (number or contact_name) and not (number and contact_name)
params = {'smstext': text}
if number:
params['smsnumber'] = number
else:
params['smscontactname'] = contact_name
return self._send_request(self._push_url, device=device, params=params)
[docs]
@action
def send_mms(self, device, file: str = None, text: str = None, subject: str = '',
number: str = None, contact_name: str = None, urgent: bool = False):
"""
Send an MMS through a mobile device connected to Join
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param text: Text to be sent
:param number: Phone number
:param contact_name: Alternatively to the phone number, you can specify a contact name
:param file: File attached to the message. Must be a local (to the phone) file or a publicly accessible URL
:param subject: MMS subject
:param urgent: Set to True if this is an urgent MMS. This will make the sent message be an MMS instead of an SMS
"""
assert (number or contact_name) and not (number and contact_name)
params = {
'mmssubject': subject,
'mmsurgent': int(urgent),
}
if number:
params['smsnumber'] = number
else:
params['smscontactname'] = contact_name
if file:
params['mmsfile'] = file
if text:
params['smstext'] = text
return self._send_request(self._push_url, device=device, params=params)
[docs]
@action
def call_number(self, device, number: str):
"""
Call a phone number through a mobile device connected to Join
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param number: Phone number
"""
return self._send_request(self._push_url, device=device, params={'callnumber': number})
[docs]
@action
def set_wallpaper(self, device, wallpaper: str):
"""
Set the wallpaper on a device connected to Join
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param wallpaper: A publicly accessible URL of an image file. Will set the wallpaper on the receiving device
"""
return self._send_request(self._push_url, device=device, params={'wallpaper': wallpaper})
[docs]
@action
def set_lock_wallpaper(self, device, wallpaper: str):
"""
Set the lock wallpaper on a device connected to Join
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param wallpaper: A publicly accessible URL of an image file. Will set the lockscreen wallpaper on the receiving
device if the device has Android 7 or above
"""
return self._send_request(self._push_url, device=device, params={'wallpaper': wallpaper})
[docs]
@action
def find(self, device):
"""
Make a device ring loudly
:param device: Device ID or name, or list of device IDs/names, or group name(s)
"""
return self._send_request(self._push_url, device=device, params={'find': True})
[docs]
@action
def set_ring_volume(self, device, volume: float):
"""
Set ring volume
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param volume: Volume
"""
return self._send_request(self._push_url, device=device, params={'ringVolume': volume})
[docs]
@action
def set_alarm_volume(self, device, volume: float):
"""
Set alarm volume
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param volume: Volume
"""
return self._send_request(self._push_url, device=device, params={'alarmVolume': volume})
[docs]
@action
def set_interruption_filter(self, device, policy: str):
"""
Set interruption filter on one or more devices
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param policy: Possible values:
- 'allow_all': Allow all notifications
- 'priority_only': Allow only priority notifications and calls
- 'alarm_only': Allow only alarm-related interruptions
- 'block_all': Do not allow any interruptions
"""
try:
policy = getattr(InterruptionFilterPolicy, policy.upper())
except AttributeError:
raise AttributeError('Invalid policy: {}. Supported values: {}'.format(
policy, [i.name for i in InterruptionFilterPolicy]
))
return self._send_request(self._push_url, device=device, params={'interruptionFilter': policy})
[docs]
@action
def say(self, device, text: str, language: str = None):
"""
Say some text through a device's TTS engine
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param text: Text to say
:param language: Language code
"""
params = {'say': text}
if language:
params['language'] = language
return self._send_request(self._push_url, device=device, params=params)
[docs]
@action
def launch_app(self, device, name: str = None, package: str = None):
"""
Launch an app on a device
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param name: Application name
:param package: Alternatively to the application name, you can also specify the application
package name. You can check the package name for an app by going to its Google Play page and checking
the end of the URL. Example: for YouTube this is the URL
(https://play.google.com/store/apps/details?id=com.google.android.youtube) and this is the package name
(com.google.android.youtube)
"""
assert (name or package) and not (name and package)
params = {}
if name:
params['app'] = name
else:
params['appPackage'] = package
return self._send_request(self._push_url, device=device, params=params)
[docs]
@action
def send_notification(self, device, title: str = None, text: str = None, url: str = None, file: str = None,
icon: str = None, small_icon: str = None, priority: int = 2, vibration_pattern=None,
dismiss_on_touch: bool = False, image: str = None, group: str = None,
sound: str = None, actions=None):
"""
Send a notification to a device
:param device: Device ID or name, or list of device IDs/names, or group name(s)
:param title: Notification title
:param text: Notification text
:param url: URL to be opened on touch
:param file: A publicly accessible URL of a file that will be opened or downloaded on touch. You can also
send the url of a file on your personal Google Drive.
:param icon: If a notification is created on the receiving device and this is set, then it'll be used as
the notification’s icon. If this image has transparency, it’ll also be used as the status bar icon is
smallicon is not set. It’ll also be used to tint the notification with its dominating color
:param small_icon: If a notification is created on the receiving device and this is set, then it’ll be used as
the notification’s status bar icon
:param priority: Control how your notification is displayed: lower priority notifications are usually
displayed lower in the notification list. Values from -2 (lowest priority) to 2 (highest priority).
Default is 2.
:param vibration_pattern: If the notification is received on an Android device, the vibration pattern in this
field will change the way the device vibrates with it. You can easily create a pattern by going to the
`AutoRemote notification page <https://autoremotejoaomgcd.appspot.com/AutoRemoteNotification.html>`_
and generate the pattern in the Vibration Pattern field
:type vibration_pattern: str (comma-separated float values) or list[float]
:param dismiss_on_touch: If set the notification will be dismissed when touched (default: False)
:param image: Publicly available URL for an image to show up in the notification
:param group: Unique ID to group your notifications with
:param sound: Publicly available URL for a sound to play with the notification
:param actions: Set notification buttons with customized behaviour. This parameter is a list of Join actions
configured on the target device that will be mapped to notification input elements.
More info `on the Joaoapps notifications page <https://joaoapps.com/join/actions/#notifications>`_
"""
params = {'dismissOnTouch': dismiss_on_touch}
if title:
params['title'] = title
if text:
params['text'] = text
if url:
params['url'] = url
if file:
params['file'] = file
if icon:
params['icon'] = icon
if small_icon:
params['smallIcon'] = small_icon
if priority:
params['priority'] = priority
if vibration_pattern:
params['vibrationPattern'] = [i for i in vibration_pattern.split(',') if len(i)] \
if isinstance(vibration_pattern, str) else vibration_pattern
assert isinstance(params['vibrationPattern'], list)
if image:
params['image'] = image
if group:
params['group'] = group
if sound:
params['sound'] = sound
if actions:
actions = [a.strip() for a in actions.split('|||' if '|||' in actions else ',') if len(a.strip())] \
if isinstance(actions, str) else actions
assert isinstance(actions, list) and len(actions) > 0
params['actions'] = '|||'.join(actions)
return self._send_request(self._push_url, device=device, params=params)
[docs]
@action
def get_devices(self):
"""
:return: List of connected devices, each containing 'id', 'name', 'user' and 'has_tasker' attributes
"""
return [
{
'id': dev['deviceId'],
'name': dev['deviceName'],
'user': dev['userAccount'],
'has_tasker': dev['hasTasker'],
}
for dev in self._send_request(self._devices_url)['records']
]
# vim:sw=4:ts=4:et:
