vincent/gsmHat
1
0
Fork 0
mirror of https://github.com/Civlo85/gsmHat.git synced 2026-01-12 02:20:02 +00:00
Code Issues Projects Releases Wiki Activity

initial commit

Browse Source
This commit is contained in:
Tarek
2020-09-02 11:52:18 +02:00
commit 58630ea4db
4 changed files with 445 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2020 Tarek Tounsi
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

115
REAMDE.md Normal file
View File

@@ -0,0 +1,115 @@
# gsmHat - A non-blocking software to use the Waveshare GSM/GPRS/GNSS HAT for Raspberry Pi with Python
With gsmHat, you can easily use the functionality of the Waveshare GSM/GPRS/GNSS HAT for Raspberry Pi ([Link to HAT](https://www.waveshare.com/gsm-gprs-gnss-hat.htm)). On this module a SIM868 Controller is doing the job too connect your Raspberry Pi with the world just by using a sim card.
## Overview
gsmHat was written for Python 3. It provides the following features
- Non-blocking receiving and sending SMS in background
## Usage
In the following paragraphs, I am going to describe how you can get and use gsmHat for your own projects.
### Getting it
To download scrapeasy, either fork this github repo or simply use Pypi via pip.
```sh
$ pip3 install gsmHat
```
### Using it
1. Install your sim card in your module, connect the GSM antenna and mount the module on the pin headers of your Raspberry Pi
Make sure, that you **do not** need to enter Pin Code to use your card
2. Enable the Uart Interface in your Raspberry Pi
1. Start raspi-config: `sudo raspi-config`.
2. Select option 5 - interfacing options.
3. Select option P6 - serial.
4. At the prompt `Would you like a login shell to be accessible over serial?` answer 'No'
5. At the prompt `Would you like the serial port hardware to be enabled?` answer 'Yes'
6. Exit raspi-config and reboot the Pi for changes to take effect.
3. Import gsmHat to your project
```Python
from gsmHat import GSMHat, SMS
```
4. Init gsmHat
```Python
gsm = GSMHat('/dev/ttyS0', 115200)
```
5. Check, if new SMS are available in your main loop
```Python
# Check, if new SMS is available
if gsm.SMS_available() > 0:
# Get new SMS
newSMS = gsm.SMS_read()
# Do something with it
```
6. Do something with your newly received SMS
```Python
# Get new SMS
newSMS = gsm.SMS_read()
print('Got new SMS from number %s' % newSMS.Sender)
print('It was received at %s' % newSMS.Date)
print('The message is: %s' % newSMS.Message)
```
7. You can also write SMS
```Python
Number = '+491601234567'
Message = 'Hello mobile world'
# Send SMS
gsm.SMS_write(Number, Message)
```
## On which platform was gsmHat built and tested?
# Hardware:
* [Raspberry Pi 4, Model B](https://www.raspberrypi.org/products/raspberry-pi-4-model-b/)
* [GSM/GPRS/GNSS/Bluetooth HAT for Raspberry Pi](https://www.waveshare.com/gsm-gprs-gnss-hat.htm), **later version that allows to power on/off the module by controlling GPIO 4**
# Software:
* Raspbian (Codename: buster, Release: 10)
* Kernel: Linux 5.4.51-v7l+
* Python: 3.7.3
License
----
MIT License
Copyright (c) 2020 Tarek Tounsi
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
contact me: <software@tounsi.de>

1
gsmHat/__init__.py Executable file
View File

@@ -0,0 +1 @@
from gsmHat.gsmHat import GSMHat, SMS

308
gsmHat/gsmHat.py Executable file
View File

@@ -0,0 +1,308 @@
#!/usr/bin/python3
# Filename: gsmHat.py
import logging
import serial
import threading
import time
import re
from datetime import datetime
import RPi.GPIO as GPIO
class SMS:
def __init__(self):
self.Message = ''
self.Sender = ''
self.Receiver = ''
self.Date = ''
class GSMHat:
"""GSM Hat Backend with SMS Functionality (for now)"""
regexGetSingleValue = r'([+][a-zA-Z\ ]+(:\ ))([\d]+)'
regexGetAllValues = r'([+][a-zA-Z:\s]+)([\w\",\s+\/:]+)'
timeoutSerial = 5
def __init__(self, SerialPort, Baudrate):
self.__baudrate = Baudrate
self.__port = SerialPort
self.__logger = logging.getLogger(__name__)
self.__logger.setLevel(logging.DEBUG)
self.__loggerFileHandle = logging.FileHandler('gsmHat.log')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
self.__loggerFileHandle.setFormatter(formatter)
self.__loggerFileHandle.setLevel(logging.DEBUG)
self.__logger.addHandler(self.__loggerFileHandle)
self.__connect()
self.__startWorking()
def __connect(self):
self.__ser = serial.Serial(self.__port, self.__baudrate)
self.__ser.flushInput()
self.__serData = ''
self.__writeLock = False
self.__logger.info('Serial connection to '+self.__port+' established')
def __disconnect(self):
self.__ser.close()
def __startWorking(self):
self.__working = True
self.__state = 1
self.__smsToRead = 0
self.__init = False
self.__readRAW = False
self.__smsToBuild = None
self.__smsList = []
self.__smsSendList = []
self.__workerThread = threading.Thread(target=self.__workerThread, daemon=True)
self.__workerThread.start()
def __stopWorking(self):
self.__working = False
self.__workerThread.join(10.0) # Timeout = 10.0 Seconds
def __sendToHat(self, string):
if self.__writeLock == False:
self.__lastCommand = string
string = string + '\n'
self.__ser.write(string.encode('iso-8859-1'))
self.__writeLock = True
self.__sentTimeout = int(round(time.time())) + self.timeoutSerial
self.__logger.debug('Sent to hat: %s' % string)
return True
else:
self.__logger.debug('Wait for Lock...')
time.sleep(1)
return False
def __pressPowerKey(self):
GPIO.setmode(GPIO.BOARD)
GPIO.setup(7, GPIO.OUT)
while True:
GPIO.output(7, GPIO.LOW)
time.sleep(4)
GPIO.output(7, GPIO.HIGH)
break
GPIO.cleanup()
time.sleep(10)
def SMS_available(self):
return len(self.__smsList)
def SMS_read(self):
if self.SMS_available() > 0:
retSMS = self.__smsList[0]
del self.__smsList[0]
return retSMS
return None
def SMS_write(self, NumberReceiver, Message):
newSMS = SMS()
newSMS.Receiver = NumberReceiver
newSMS.Message = Message
self.__smsSendList.append(newSMS)
def close(self):
self.__disconnect()
self.__logger.info('Serial connection to '+self.__port+' closed')
self.__stopWorking()
def __processData(self):
if self.__serData != '':
if self.__readRAW:
self.__logger.debug('Received Raw Data: %s' % self.__serData)
if self.__serData == 'OK\r\n':
self.__smsToBuild.Message = self.__smsToBuild.Message.rstrip('\r\n')
self.__smsList.append(self.__smsToBuild)
self.__readRAW = False
self.__writeLock = False
else:
#self.__smsToBuild.Message = bytearray.fromhex(self.__serData).decode()
self.__smsToBuild.Message = self.__smsToBuild.Message + self.__serData
else:
self.__logger.debug('Received Data: %s' % self.__serData)
if 'OK' in self.__serData:
self.__writeLock = False
self.__logger.debug('Lock Off')
elif '+CME ERROR:' in self.__serData:
self.__writeLock = False
match = re.findall(self.regexGetSingleValue, self.__serData)
self.__cmeErr = int(match[0][1])
self.__logger.info('Got CME ERROR: %s' % match[0][1])
elif '+CMS ERROR:' in self.__serData:
self.__writeLock = False
match = re.findall(self.regexGetSingleValue, self.__serData)
self.__cmsErr = int(match[0][1])
self.__logger.info('Got CMS ERROR: %s' % match[0][1])
elif '+CPMS:' in self.__serData:
match = re.findall(self.regexGetAllValues, self.__serData)
rawData = match[0][1].split(',')
self.__masSMSSpace = int(rawData[1])
numSMS = int(rawData[0])
if numSMS > 0:
self.__smsToRead = 1
elif '+CMGR:' in self.__serData:
# read SMS content
match = re.findall(self.regexGetAllValues, self.__serData)
rawData = match[0][1].split('","')
self.__readRAW = True
self.__smsToBuild = SMS()
#self.__smsToBuild.Sender = bytearray.fromhex(rawData[1]).decode()
self.__smsToBuild.Sender = rawData[1]
self.__smsToBuild.Date = rawData[3].replace('"', '')
self.__smsToBuild.Date = datetime.strptime(rawData[3].replace('"', '')[:-3], '%y/%m/%d,%H:%M:%S')
self.__smsToBuild.Message = ''
# unannounced data reception below (e.g. new SMS oder phone call)
elif '+CMTI:' in self.__serData:
self.__logger.info('Received new SMS')
match = re.findall(self.regexGetAllValues, self.__serData)
rawData = match[0][1].split(',')
storage = rawData[0]
numSMS = int(rawData[1])
self.__logger.debug('New SMS in memory ' + storage + ' at position ' + str(numSMS))
self.__smsToRead = numSMS
self.__serData = ''
def __waitForUnlock(self):
actTime = int(round(time.time()))
if self.__sentTimeout > 0 and actTime > self.__sentTimeout:
# Timeout
self.__logger.error('Timeout during data reception')
if self.__state == 2:
# It might be that the gsm module is not powered on
# So let's try
self.__logger.error('Try to restart gsm module')
self.__pressPowerKey()
self.__state = 1
self.__writeLock = False
self.__sentTimeout = 0
return False
else:
raise 'Unhandled timeout during data reception'
if self.__writeLock:
return False
else:
self.__sentTimeout = 0
return True
def __workerThread(self):
self.__logger.info('Worker started')
self.__waitTime = 0
while self.__working:
# Check for incoming chars
while self.__ser.inWaiting() > 0:
newChar = self.__ser.read().decode('iso-8859-1')
if newChar == '\n':
if self.__readRAW == True:
self.__serData += newChar
self.__processData()
else:
if newChar == '\r':
if self.__readRAW == True:
self.__serData += newChar
else:
self.__serData += newChar
# Statemachine
actTime = int(round(time.time() * 1000))
if self.__state == 1:
if self.__sendToHat('AT+CMGF=1'):
self.__state = 2
elif self.__state == 2:
if self.__waitForUnlock():
if self.__sendToHat('AT+CPMS="SM"'):
self.__state = 3
elif self.__state == 3:
if self.__waitForUnlock():
self.__state = 97
self.__nextState = 2
self.__waitTime = actTime + 5000
elif self.__state == 20:
# Read SMS
if self.__sendToHat('AT+CMGR='+str(self.__smsToRead)):
self.__state = 21
elif self.__state == 21:
if self.__waitForUnlock():
if self.__smsToBuild == None:
# An der Stelle self.__smsToRead gab es keine SMS zu lesen
pass
else:
# Es gab eine neue SMS
self.__smsToBuild = None
# Lösche die behandelte SMS an der Stelle
if self.__sendToHat('AT+CMGD='+str(self.__smsToRead)):
self.__state = 22
elif self.__state == 22:
if self.__waitForUnlock():
if(self.__smsToRead == 20):
self.__smsToRead = 0
else:
self.__smsToRead = self.__smsToRead + 1
self.__state = 97
self.__nextState = 2
self.__waitTime = actTime + 5000
elif self.__state == 30:
# SMS versenden
retSMS = self.__smsSendList[0]
messageString = 'AT+CMGS="' + retSMS.Receiver + '"\n' + retSMS.Message + '\x1A'
self.timeoutSerial = 30
if self.__sendToHat(messageString):
self.__state = 31
elif self.__state == 31:
if self.__waitForUnlock():
retSMS = self.__smsSendList[0]
self.__logger.info('Message to ' + retSMS.Receiver + ' successfully sent')
del self.__smsSendList[0]
self.timeoutSerial = 5
self.__state = 97
self.__nextState = 2
self.__waitTime = actTime + 5000
elif self.__state == 97:
# Check if new SMS to send is there
if len(self.__smsSendList) > 0:
self.__state = 30
# Check if new SMS is there
elif self.__smsToRead > 0:
self.__state = 20
# Wait x Seconds
elif actTime > self.__waitTime:
self.__state = self.__nextState
elif self.__state == 98:
#Check if alive
self.__logger.debug('Check if alive 98')
if self.__sendToHat('AT'):
self.__state = 99
elif self.__state == 99:
#Check if alive
if self.__waitForUnlock():
self.__state = 97
self.__nextState = 98
self.__waitTime = actTime + 5000
# Let other Threads also do their job
time.sleep(0.1)
self.__logger.info('Worker ended')