105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
|
|
import logging
|
|
import os
|
|
from pydbus import SystemBus
|
|
from gi.repository import GLib
|
|
|
|
|
|
# Setup logging
|
|
logfile = "/opt/sms/sms.log"
|
|
FORMAT = '%(asctime)-15s %(message)s'
|
|
logging.basicConfig(format=FORMAT,filename=logfile,level=10)
|
|
logger = logging.getLogger('observiumsms')
|
|
|
|
class ModemManagerError(Exception):
|
|
def __init__(self, value):
|
|
self.value = value
|
|
def __str__(self):
|
|
return repr(self.value)
|
|
|
|
class SIMError(ModemManagerError):
|
|
def __init__(self, value):
|
|
self.value = value
|
|
def __str__(self):
|
|
return repr(self.value)
|
|
|
|
def assemble_sms():
|
|
str_list = []
|
|
str_list.append("Observium {}\n".format(os.environ.get('OBSERVIUM_TITLE')))
|
|
str_list.append("{} [{}]\n".format(os.environ.get('OBSERVIUM_ENTITY_NAME'), os.environ.get('OBSERVIUM_ENTITY_DESCRIPTION')))
|
|
str_list.append("{}".format(os.environ.get('OBSERVIUM_DURATION')))
|
|
return ''.join(str_list)
|
|
|
|
def get_modem():
|
|
# Contact ModemManager, unlock SIM and send SMS to emergency numbers
|
|
with SystemBus() as bus:
|
|
mm = bus.get('.ModemManager1')
|
|
modems = mm.GetManagedObjects()
|
|
if not modems or len(modems) == 0:
|
|
raise ModemManagerError('No modems found')
|
|
modem_path = list(modems.keys())[0]
|
|
|
|
logger.info("Going with %s", modem_path)
|
|
modem = bus.get(".ModemManager1", modem_path)
|
|
return modem
|
|
|
|
def unlock_sim(modem):
|
|
sim_path = modem.Sim
|
|
if not sim_path:
|
|
raise SIMError('No SIM card found')
|
|
|
|
# unlock SIM, if not alreadY
|
|
props = modem.GetStatus()
|
|
print(props)
|
|
if props and props['state'] == 2:
|
|
logger.info("Trying to unlock %s", sim_path)
|
|
with SystemBus() as bus:
|
|
sim = bus.get('.ModemManager1', sim_path)
|
|
sim.SendPin("1234", timeout=10000)
|
|
props = modem.GetStatus()
|
|
if props and props['state'] == 2:
|
|
remaining_attempts = modem.UnlockRetries
|
|
print(remaining_attempts)
|
|
raise SIMError('Wrong PIN!')
|
|
|
|
def send_sms(recipients, smsstring):
|
|
|
|
# Assemble SMS
|
|
logger.info("Sending SMS: %s", smsstring)
|
|
|
|
# Get modem, unlock SIM
|
|
modem = get_modem()
|
|
unlock_sim(modem)
|
|
|
|
# enable the modem
|
|
modem.Enable(True)
|
|
|
|
# test sending SMS
|
|
modem_msg = modem['org.freedesktop.ModemManager1.Modem.Messaging']
|
|
smses = modem_msg.List()
|
|
|
|
print(modem.GetStatus()['signal-quality'])
|
|
|
|
for line in recipients:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
logger.info("Creating new SMS for %s", line)
|
|
smstext = GLib.Variant('s', smsstring)
|
|
smsnumber = GLib.Variant('s', line)
|
|
|
|
dictsms = { 'text': smstext, 'number': smsnumber }
|
|
|
|
msgpath = modem_msg.Create(dictsms)
|
|
logger.info("Created new SMS %s", msgpath)
|
|
|
|
logger.info("Sending SMS %s", msgpath)
|
|
with SystemBus() as bus:
|
|
sms = bus.get(".ModemManager1", msgpath)
|
|
sms.Send(timeout=30000)
|
|
|
|
logger.info("SMS %s state %s", msgpath, sms.State)
|
|
|
|
|