Code: Select all
GPIO.add_event_detect(4,
GPIO.RISING,
callback=self.my_callback,
bouncetime=500
)
Code: Select all
def my_callback(self, channel):
print "rising detected on GPIO 4"
self.n += 1
Code: Select all
from threading import Thread
from subprocess import Popen
import shlex
from time import sleep
class TimeoutProcess(Thread):
"""New class that calls a subprocess but terminates it if the process has
not completed within the necessary time period.
Takes two parameters:
process: (string) command to be run
timeout: (int) number of seconds until process is killed
"""
def __init__(self, process, timeout):
Thread.__init__(self)
# We use shlex.split to turn the command into something that the
# subprocess module likes to use
self.process = shlex.split(process)
# Make sure the timeout is an integer
self.timeout = int(timeout)
def run(self):
# Create an object that runs the process
self.proc = Popen(self.process)
# Set a flag to show the process is running
running = True
# We'll check the status of the process every second
for _ in range(self.timeout):
# poll will return None if the process hasn't exited (and the
# exit code once it has)
if self.proc.poll() is not None:
# process has exited so change flag
running = False
# No need to keep looping so let's exit
break
# If we're here, process is still running so sleep for 1 second
sleep(1)
# The loop's exited so we've either reached the timeout period
# or the process has already finished.
# Have a look at the flag to see which is true
if running:
# Process is still running so let's kill it
self.proc.terminate()
#print "Process killed"
# To use it we create an instance
t = TimeoutProcess("/path/to/my/command", 5)
# and run it
t.start()
# NB This thread is not daemonised i.e. your main script won't exit until this
# thread finishes (e.g. process completes or it's killed after the timeout)
# If that's not the behaviour you want, call t.daemon = True before starting
# the thread.Code: Select all
#!/usr/bin/python
def pr(n):
print "n = ", n
Code: Select all
Traceback (most recent call last):
File "main_script.py", line 153, in my_callback
t = TimeoutProcess(stamp.pr, 5)
File "main_script.py", line 54, in __init__
self.process = shlex.split(process)
File "/usr/lib/python2.7/shlex.py", line 279, in split
return list(lex)
File "/usr/lib/python2.7/shlex.py", line 269, in next
token = self.get_token()
File "/usr/lib/python2.7/shlex.py", line 96, in get_token
raw = self.read_token()
File "/usr/lib/python2.7/shlex.py", line 124, in read_token
nextchar = self.instream.read(1)
AttributeError: 'function' object has no attribute 'read'
Code: Select all
t = TimeoutProcess("python stamp.py", 5)Code: Select all
t=Timeoutprocess(stamp.pr(self.n),5)
"stamp.py" is run separately, I have an "import stamp" line in the main script.elParaguayo wrote:No.
The main question I have is whether "stamp.py" is run separately or whether you're importing it into your main script (i.e. have you got an "import stamp" line somewhere?)
Ok, but what is the difference between the code that you wrote and the multiprocessing module that you are suggesting me?The code I wrote for you won't work where you are calling python functions that are available after importing the function. It works where you would run a command separately (e.g. "python stamp.py").
If you've imported the function then I might be tempted to use something like the multiprocessing module to run a process in the background and then kill it if necessary.
Code: Select all
# MainApp.py
from random import randint
from glob import glob
from os.path import join, dirname
from kivy.app import App
from kivy.clock import Clock
from kivy.core.window import Window
from kivy.properties import StringProperty
from kivy.uix.carousel import Carousel
from kivy.uix.image import AsyncImage
from kivy.uix.scatter import Scatter
from kivy.uix.floatlayout import FloatLayout
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.label import Label
from kivy.uix.image import Image
import RPi.GPIO as GPIO
# ************************* HERE I IMPORT THE SCRIPT STAMP.PY ***************************************
import stamp.py
class Screen(FloatLayout):
def __init__(self, **kwargs):
super(Screen, self).__init__(**kwargs)
# Here's your GPIO stuff
GPIO.setmode(GPIO.BCM)
GPIO.setup(4, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
self.n = 0
# Add a call function
GPIO.add_event_detect(4,
GPIO.RISING,
callback=self.my_callback,
bouncetime=500)
layout = BoxLayout(orientation='horizontal', spacing=3)
#il parametro spacing determina la distanza tra i due box
# Create a template for your text
self.tx ='[b][size=45][color=ff3333]The value of n is {n}[/color][/size][/b]'
self.lbl = Label(text=self.tx.format(n=self.n),
markup=True,
size_hint=(.4, 1),
pos_hint={"center_x":.5, "center_y":.7})
carousel = Carousel(direction='right', size_hint=(.6, 1))
carousel.loop = True
curdir = dirname(__file__)
for filename in glob(join(curdir, '/home/pi/mufolder/img', '*')):
picture = Image(source=filename,allow_stretch=True)
carousel.add_widget(picture)
Clock.schedule_interval(carousel.load_next, 5)
layout.add_widget(carousel)
layout.add_widget(self.lbl)
self.add_widget(layout)
def my_callback(self, channel):
print "rising detected on GPIO 4"
self.n += 1
# Update our template with the current value of n
tx = self.tx.format(n=self.n)
# Print it
print tx
# Update the text on the label.
self.lbl.text = tx
# *************** HERE I WANT CALL THE NEW PROCESS STAMP.PR PASSING THE VALUE "self.n" *********
class MainApp(App):
def build(self):
#imposto il colore dello sfondo
Window.clearcolor = (255, 255, 255, 1)
#Window.size=(500, 500)
return Screen()
MainApp().run()Code: Select all
# STAMP.PY
#!/usr/bin/python
def pr(n):
print "n = ", n
That's not running separately. It is only run when you call "stamp.pr(n)" from your main script. There's nothing wrong with that at all and I'm sure we can still get it to work.paolojo wrote:"stamp.py" is run separately, I have an "import stamp" line in the main script.
Ok, that's what I want: "stamp.pr(n)" must run when I call it from my main script (see the above post)That's not running separately. It is only run when you call "stamp.pr(n)" from your main script. There's nothing wrong with that at all and I'm sure we can still get it to work.
"stamp.py" is an imported python script that contains a function called "pr".However, rather than giving an example of stamp.py can you say what it's actually going to do i.e. is it a function that you're going to import? do you need to pass parameters to it? do you need to get values back from it?
So, I think that the best way is run the function in a separate process. Do you agree with me?If you're just looking to run a python function in the background, this could be done in a thread however there's no neat way of killing a thread to my knowledge (there are also numerous conversations on StackOverflow about why killing threads is bad).
The multiprocessing module is like threading, but it runs the function in a separate process which does allow it to be killed (it's probably got lots of other benefits which I'm failing to mention too).
Which solution you use depends on what your code is and what you want it to do.
Code: Select all
#PARSE.PY (that substitutes the stamp.py)
import json,httplib
connection = httplib.HTTPSConnection('api.parse.com', 443)
connection.connect()
connection.request('PUT', '/1/classes/GameScore/Ed1nuqPvcm', json.dumps({
"score": 73453
}), {
"X-Parse-Application-Id": "${APPLICATION_ID}",
"X-Parse-REST-API-Key": "${REST_API_KEY}",
"Content-Type": "application/json"
})
result = json.loads(connection.getresponse().read())
print result Code: Select all
#!/usr/bin/env python
from random import randint
from glob import glob
from os.path import join, dirname
from kivy.app import App
from kivy.clock import Clock
from kivy.core.window import Window
from kivy.properties import StringProperty
from kivy.uix.carousel import Carousel
from kivy.uix.image import AsyncImage
from kivy.uix.scatter import Scatter
from kivy.uix.floatlayout import FloatLayout
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.label import Label
from kivy.uix.image import Image
from kivy.uix.button import Button
import RPi.GPIO as GPIO
from webtimeout import WebTimeout
# Some of your web stuff is constant so define it here:
headers = {"X-Parse-Application-Id": "${APPLICATION_ID}",
"X-Parse-REST-API-Key": "${REST_API_KEY}",
"Content-Type": "application/json"
}
address = "https://api.parse.com//1/classes/GameScore/Ed1nuqPvcm"
method = "PUT"
data = None
class Screen(FloatLayout):
def __init__(self, **kwargs):
super(Screen, self).__init__(**kwargs)
# # Here's your GPIO stuff
GPIO.setmode(GPIO.BCM)
GPIO.setup(4, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
self.n = 0
# Add a call function
GPIO.add_event_detect(4,
GPIO.RISING,
callback=self.my_callback,
bouncetime=500)
layout = BoxLayout(orientation='horizontal', spacing=3)
#il parametro spacing determina la distanza tra i due box
# Create a template for your text
self.tx ='[b][size=45][color=ff3333]The value of n is {n}[/color][/size][/b]'
self.lbl = Label(text=self.tx.format(n=self.n),
markup=True,
size_hint=(.4, 1),
pos_hint={"center_x":.5, "center_y":.7})
# This is a blank label for handling the response from the web request
self.callback_lbl = Label(text="", size_hint=(None,None),
size=(200,50), pos=(200,100))
carousel = Carousel(direction='right', size_hint=(.6, 1))
carousel.loop = True
curdir = dirname(__file__)
for filename in glob(join(curdir, '/home/pi/mufolder/img', '*')):
picture = Image(source=filename,allow_stretch=True)
carousel.add_widget(picture)
Clock.schedule_interval(carousel.load_next, 5)
layout.add_widget(carousel)
layout.add_widget(self.lbl)
self.add_widget(layout)
self.add_widget(self.callback_lbl)
def my_callback(self, channel):
print "rising detected on GPIO 4"
self.n += 1
# Update our template with the current value of n
tx = self.tx.format(n=self.n)
# Print it
print tx
# Update the text on the label.
self.lbl.text = tx
# This is where the web request is created
# if you need to submit data, prepare that here (assuming it's not
# constant) e.g.
data = {"score": self.n}
# The example below is set up to run a callback method when the web
# requests ends
wt = WebTimeout(address=address, headers=headers, timeout=10,
method=method, data=data,
callback=self.my_web_callback)
wt.start()
def my_web_callback(self, result):
# This is the callback function
# Result is a tuple of:
# sucess: True if web request submitted successfully
# result: The reponse from the web page
success, response = result
# As an example, I'm just showing a label if the request was submitted
# successfully
if success:
txt = "Web request submitted successfully"
else:
txt = "Error submitting web request"
self.callback_lbl.text = txt
class MainApp(App):
def build(self):
#imposto il colore dello sfondo
Window.clearcolor = (255, 255, 255, 1)
#Window.size=(500, 500)
return Screen()
MainApp().run()
Code: Select all
#!/usr/bin/env python
from threading import Thread
import socket
import requests
class WebTimeout(Thread):
'''An example class to show how web requests can be submitted in a
background thread.
Takes one mandatory parameter:
address: The web address to request
Takes six optional parameters:
timeout: Number of seconds until request timesour
data: Data to be submitted with request (e.g. via POST)
headers: Headers to be added to the request
queue: Queue object for thread-safe communication
method: GET, POST, DELETE, PUT
callback: Function to be called when request ends
'''
def __init__(self, address, timeout=5, data=None, headers=None, queue=None,
method="GET", callback=None):
Thread.__init__(self)
self.q = queue
self.timeout = timeout
self.address = address
self.data = data
self.headers = headers
self.method = method.lower()
self.callback = callback
def _sendRequest(self):
'''Method to send the request. Returns a tuple
(success, result)
where:
success: (Boolean) True if request submitted successfully
result: Python dict/list if response if valid JSON, else text
of response body.
'''
# Check if the requested method (GET, POST etc) is valid
try:
req = getattr(requests, self.method)
except AttributeError:
return (False, "Invalid method")
# Build the request and submit
try:
r = req(self.address,
data=self.data,
headers=self.headers,
timeout=self.timeout)
# Handle a timeout event
except (socket.timeout, requests.Timeout):
return (False, "Timed out")
# Handle a connection error
except requests.ConnectionError:
return (False, "Connection error")
# Catch anything else that goes wrong
# (it's bad practice to do this!)
except:
return (False, "Unknown error")
# Check if the response is successful
if r.status_code != 200:
return (False, "Response code {}".format(r.status_code))
# See if the response is a JSON object
try:
response = r.json()
# If not, try getting the text
except requests.compat.json.JSONDecodeError:
response = r.text
# Naughty...
except:
response = "Some other error"
finally:
return (True, response)
def run(self):
'''This is the method that's run when the start() method is called.'''
# submit the web request
result = self._sendRequest()
# If we've got a queue object, add the response to the queue
# where it can be read via the get() method in the main thread
if self.q:
self.q.put(result)
# If we've got a callback function, call it.
if self.callback:
self.callback(result)
As the Python programming language itself is written in C, then anything Python can do must be possible with C. Note, however, that directly programming in C is likely to be more difficult than Python, especially when implementing communication protocols that require string handling. With all the error checks and comments the posted code looks overwhelming. Still it may be possible to use without full understanding. If you stay focused on one programming language, the meaning of such code should eventually be clear.paolojo wrote:Hi,
I'm trying to understand WebTimeout.py code...but it is quite difficult for me because I don't know very well python language.
In your opinion, is it possibile to implement the same function in C language?
Thank you
