Skip to content

Instantly share code, notes, and snippets.

@zed zed/morse_code_bot.py Secret
Last active Jul 26, 2018

Embed
What would you like to do?
Skype Morse code bot
#!/usr/bin/env python3
"""Skype bot that translates its input messages to International Morse code and back.
The standard:
https://www.itu.int/dms_pubrec/itu-r/rec/m/R-REC-M.1677-1-200910-I!!PDF-E.pdf
# XXX 3.3.2, 3.5.1, 4.3, Part II, no abbreviations
"""
import asyncio
import json
import logging
import os
import re
import sys
import urllib.parse
import aiohttp.web # $ pip install aiohttp
import werkzeug.contrib.cache # $ pip install werkzeug
__version__ = '0.2.2'
log = logging.getLogger(__name__).debug
letter_sep = ' '
word_sep = ' \n'
morse_code = {
'a': '.-', 'b': '-...', 'c': '-.-.',
'd': '-..', 'e': '.', 'f': '..-.',
'g': '--.', 'h': '....', 'i': '..',
'j': '.---', 'k': '-.-', 'l': '.-..',
'm': '--', 'n': '-.', 'o': '---',
'p': '.--.', 'q': '--.-', 'r': '.-.',
's': '...', 't': '-', 'u': '..-',
'v': '...-', 'w': '.--', 'x': '-..-',
'y': '-.--', 'z': '--..',
'é': '..-..',
'1': '.----', '2': '..---', '3': '...--',
'4': '....-', '5': '.....', '6': '-....',
'7': '--...', '8': '---..', '9': '----.',
'0': '-----',
'.': '.-.-.-', ',': '--..--', ':': '---...',
'?': '..--..', "'": '.----.', '-': '-....-',
'/': '-..-.', '(': '-.--.', ')': '-.--.-',
'"': '.-..-.', '=': '-...-',
'understood': '...-.',
'error': '.' * 8,
'+': '.-.-.',
'invitation to transmit': '-.-',
'wait': '.-...',
'end of work': '...-.-',
'starting signal': '-.-.-',
'*': '-..-', '@': '.--.-.',
}
morse_code['%'] = ' '.join(map(morse_code.get, '0/0'))
morse_code['‰'] = ' '.join(map(morse_code.get, '0/00'))
english_from_morse = dict(zip(morse_code.values(), morse_code.keys()))
# cache access token on disk
cache = werkzeug.contrib.cache.FileSystemCache('.cachedir', threshold=86400)
common_http_headers = {'User-Agent': f'morse-code-bot/{__version__}'}
def is_morse_code(text):
"""
>>> is_morse_code('-.-')
True
>>> is_morse_code('........ ........ ........')
True
>>> is_morse_code('abc')
False
"""
return frozenset(''.join(text.split())) <= {'-', '.'}
def morse2english(code):
r"""
>>> morse2english('........ ........ ........')
'errorerrorerror'
>>> morse2english('.... .. \n.--.-. .-.-.-')
'hi @.'
"""
return ' '.join(''.join(english_from_morse.get(morse_char_code, '\ufffd')
for morse_char_code in word.split())
for word in filter(None, re.split(r'\s{2,}', code)))
def english2morse(text):
r"""
>>> english2morse('hi @.')
'.... .. \n.--.-. .-.-.-'
"""
words = text.split()
return word_sep.join(letter_sep.join(
morse_code.get(c, c if c.isspace() else morse_code['error'])
for c in word)
for word in words)
async def get_access_token():
token = cache.get(key='token')
if not token:
# request access token
url = 'https://login.microsoftonline.com/common/oauth2/v2.0/token'
data = dict(client_id=os.environ['APP_ID'],
scope='https://graph.microsoft.com/.default',
grant_type='client_credentials',
client_secret=os.environ['APP_SECRET'])
headers = {**common_http_headers, **{'Cache-Control': 'no-cache'}}
async with aiohttp.post(url, data=data, headers=headers) as r:
assert 200 <= r.status < 300
token = await r.json()
cache.set('token', token, timeout=token['expires_in'])
return token['access_token']
async def send_message(message, skypeid):
"""
POST / v2 / conversations / 8: alice / activities HTTP / 1.1
Host: apis.skype.com
Authorization: Bearer < redacted oauth2 token >
{
"message": {"content": "Hi! (wave)"}
}
"""
url = f'https://apis.skype.com/v2/conversations/8:{skypeid}/activities'
token = await get_access_token()
headers = dict(common_http_headers, Authorization=f'Bearer {token}')
headers['Content-Type'] = headers['Accept'] = 'application/json'
data = json.dumps(dict(message=dict(content=message)))
async with aiohttp.post(url, data=data, headers=headers) as r:
assert 200 <= r.status < 300
async def handle(request):
messages = await request.json()
log('got %r', messages)
for msg in messages:
if msg['activity'] != 'message':
continue
assert msg['to'] == ('28:' + os.environ['APP_ID']) # ours
assert msg['from'][:2] == '8:'
skypeid = msg['from'][2:]
assert urllib.parse.quote(skypeid, safe='') == skypeid
text = msg['content'].casefold()
translated = [english2morse, morse2english][is_morse_code(text)](text)
assert {'<', '>', '&'}.isdisjoint(translated)
asyncio.ensure_future(send_message(translated, skypeid)) # don't wait
return aiohttp.web.HTTPCreated() # 201
app = aiohttp.web.Application()
app.router.add_route('POST', '/v1/chat', handle)
if __name__ == "__main__":
logging.basicConfig(format="%(asctime)-15s %(message)s",
datefmt="%F %T",
level=logging.DEBUG)
aiohttp.web.run_app(app,
host='localhost',
ssl_context=None,
port=int(sys.argv[1]) if len(sys.argv) > 1 else None)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.