timelapse/timelapse-raspi
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor and enhance timelapse capture system

Browse Source 
- Removed obsolete script `script_SANSDEMARRAGE.sh`.
- Added new `Camera.py` and `Connexion.py` files for camera handling and socket communication.
- Implemented `First_Try.py` for initial camera preview testing.
- Created `Humidity.py` for humidity sensor data acquisition.
- Developed `Send_data_stocked.py` for managing and sending stored data.
- Introduced `Time_Lapse_Connection.py` and `Time_Lapse_NoConnection.py` for connected and offline modes.
- Added `get_from_server.py` for retrieving camera status from the server.
- Updated `sync_offline_data.py` for synchronizing offline data.
- Created `timelapse.service` for managing the timelapse service on Raspberry Pi.
- Established package structure with `__init__.py` and `api_client.py` for API interactions.
- Enhanced `capture.py` for managing image captures and data storage.
- Configured `config.py` for centralized configuration management.
- Developed `sensors.py` for handling environmental sensors and camera operations.
- Implemented `timelapse_offline.py` and `timelapse_online.py` for capturing images in offline and online modes.
This commit is contained in:
Kerboul
2025-04-27 16:45:00 +02:00
parent bebfb8adb3
commit 610d220c3f
20 changed files with 951 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
old_files/Camera.py Normal file
View File

@@ -0,0 +1,25 @@
import picamera2 as pc
import time
cam = pc.Picamera2()
conf = cam.create_preview_configuration(main={"size":(800, 600)})
cam.configure(conf)
cam.start_preview(pc.Preview.QTGL)
cam.start()
time.sleep(10)
image = cam.capture_file("Ma photo.jpg")
print(image)
cam.close()
"""
print("bonjour")
request.save("MyPhotohraph", "AGAGAGAGA.jpg")
request.release()
"""
#cam.capture_image()
#cam.show()
print("ok")

38
old_files/Connexion.py Normal file
View File

@@ -0,0 +1,38 @@
import socket
class MySocket:
"""demonstration class only
- coded for clarity, not efficiency
"""
def __init__(self, sock=None):
if sock is None:
self.sock = socket.socket(
socket.AF_INET, socket.SOCK_STREAM)
else:
self.sock = sock
def connect(self, host, port):
self.sock.connect((host, port))
def mysend(self, msg):
totalsent = 0
while totalsent < MSGLEN:
sent = self.sock.send(msg[totalsent:])
if sent == 0:
raise RuntimeError("socket connection broken")
totalsent = totalsent + sent
def myreceive(self):
chunks = []
bytes_recd = 0
while bytes_recd < MSGLEN:
chunk = self.sock.recv(min(MSGLEN - bytes_recd, 2048))
if chunk == b'':
raise RuntimeError("socket connection broken")
chunks.append(chunk)
bytes_recd = bytes_recd + len(chunk)
return b''.join(chunks)
if __name__ == "__main__":
print("bonjour")
MySocket()

10
old_files/First_Try.py Normal file
View File

@@ -0,0 +1,10 @@
import time
import picamera2 as pc
camera = pc.Picamera2()
try:
camera.strat_preview()
time.sleep(10)
camera.stop_preview()
finally:
camera.close()

53
old_files/Humidity.py Normal file
View File

@@ -0,0 +1,53 @@
import smbus2
import RPi.GPIO as GPIO
import time
"""
>>> bus = smbus2.SMBus(1)
>>> ans = bus.write_byte(0x44, 0xFD)
>>> smbus2.i2c_msg.read(0x44,6)
i2c_msg(68,1,b'\x00\x00\x00\x00\x00\x00')
>>> taa=smbus2.i2c_msg.read(0x44,6)
>>> bus.i2c_rdwr(taa)
>>> data = list(taa)
>>> print(data)
[112, 174, 89, 60, 78, 165]
"""
def main():
i2cbus = smbus2.SMBus(1)
addressSHT40 = 0x44
meas_sht40(i2cbus, 0x44)
def meas_sht40(bus, addressSHT40):
CMD_measure = 0xFD #cmd de mesure
#bus.write_byte(0x00, 0x06)
#time.sleep(1.1)
bus.write_byte(0x44, 0x39)
time.sleep(1.1)
while 1:
time.sleep(1.1)
#data = bus.read_i2c_block_data(0x44, 0, 6) #Lecture des 6 octets
data = []
bus.write_byte(0x44, 0xFD)
byte = bus.read_block_data(0x44,3)
data.append(byte)
print(f"Octet {i}: {byte:#04x}")
time.sleep(1.15)
#print(data)
"""
temp_raw = (data[0] << 8) + data[1]
humidity_raw = (data[3] <<8) + data[4]
"""
temperature = 10000*(data/65535.0)
#temperature = -45 + 175*data/65535
#print(temperature)
if __name__=="__main__":
main()

72
old_files/Send_data_stocked.py Normal file
View File

@@ -0,0 +1,72 @@
import os
import requests
import shutil
import json
class Send_data_stocked:
def __init__(self):
self.url = "https://timelapse.kerboul.me/api/camera/upload"
self.headers = {
'accept': 'application/json'
}
self.project_path = "//home//timelapse//Documents//Time_Lapse//PROJECT//"
def get_list_folder(self,path = "//home//timelapse//Documents//Time_Lapse//PROJECT"):
# Liste des noms de dossiers
noms_dossiers = [nom for nom in os.listdir(path) if os.path.isdir(os.path.join(path, nom))]
print("Noms des dossiers :", noms_dossiers)
return noms_dossiers
def send_ALL_data_stocked(self):
tab_files = self.get_list_folder()
for i in tab_files:
self.send_data_stocked(i)
def send_data_stocked(self,file_name):
file_path = self.project_path + file_name + "//" + file_name
json_file_path = file_path + ".json"
img_file_path = file_path + ".jpg"
datas = self.get_json(json_file_path)
print(datas)
answer_from_server = self.upload_measurement(img_file_path, datas["timestamp"], datas["temperature"], datas["humidity"])
print(answer_from_server)
if (answer_from_server["message"]== "Mesure téléchargée avec succès"):
self.del_Folder(datas["timestamp"])
def get_json(self, file_path):
with open(file_path, "r", encoding='utf-8') as file:
datas = json.load(file)
return datas
return False
def del_Folder(self, name):
shutil.rmtree(self.project_path + name)
def upload_measurement(self, image_path, timestamp, temperature, humidity):
data = {
'timestamp': timestamp,
'temperature': temperature,
'humidity': humidity
}
files = {
'image': (image_path, open(image_path, 'rb'), 'image/jpeg')
}
try:
#print(data)
#print(files)
response = requests.post(self.url, headers=self.headers, data=data, files=files)
print(response)
response.raise_for_status()
response_data = response.json()
return response_data
except requests.exceptions.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
except Exception as err:
print(f"Other error occurred: {err}")
if __name__ == "__main__":
SDS = Send_data_stocked()
#a = SDS.get_list_folder()[0]
SDS.send_ALL_data_stocked()

133
old_files/Time_Lapse_Connection.py Normal file
View File

@@ -0,0 +1,133 @@
import time
from datetime import datetime
import picamera2 as pc
import smbus2
import os
import json
import shutil
import requests
class TimeLapse:
def __init__(self):
self.url = "https://timelapse.kerboul.me/api/camera/upload"
self.headers = {
'accept': 'application/json'
}
self.data = {
"timestamp": "",
"temperature": "",
'humidity': "",
}
self.project_path = "//home//timelapse//Documents//Time_Lapse//PROJECT//"
def create_Folder(self):
filename = self.data["timestamp"]
os.mkdir(self.project_path + filename)
def create_Json(self):
filename = self.project_path + self.data["timestamp"] + "//" + self.data["timestamp"] + ".json"
with open(filename, "w") as file:
json.dump(self.get_Alldic(), file)
def get_Alldic(self):
return self.data
def get_Hygrodata(self):
bus = smbus2.SMBus(1)
ans = bus.write_byte(0x44, 0xFD)
time.sleep(0.1)
ans=smbus2.i2c_msg.read(0x44,6)
bus.i2c_rdwr(ans)
data = list(ans)
t_ticks = data[0]*256 + data[1] #Pas de CRC utilisé
rh_ticks = data[3]*256 + data[4] #Pas de CRC utilisé
self.data["temperature"] = (175*t_ticks)/65535. -45
self.data["humidity"] = (125*rh_ticks)/65535. -6
return self.data
def get_humidity(self):
return self.data["humidity"]
def get_temperature(self):
return self.data["temperature"]
def get_TimeStamp(self): #ATTENTION AU NOMMAGE DES FICHIERS !!! PAS DE : ET ESPACES
now = datetime.now()
#dt_string = now.strftime("%d_%m_%Y_%H_%M_%S")
dt_string = now.strftime("%Y-%m-%d %H:%M:%S")
print("date and time =", str(dt_string))
self.data["timestamp"]= str(dt_string)
return str(dt_string)
def pick_Picture(self):
cam = pc.Picamera2()
conf = cam.create_preview_configuration(main={"size":(800, 600)})
cam.configure(conf)
cam.start_preview(pc.Preview.QTGL)
cam.start()
time.sleep(0.1)
timestamp = self.get_TimeStamp()
path = self.project_path + timestamp + "//" + timestamp+".jpg"
self.create_Folder()
image = cam.capture_file(path)
cam.close()
def del_Folder(self, name):
shutil.rmtree(self.project_path + name)
def Singleroutin(self):
TL.get_Hygrodata()
TL.pick_Picture()
TL.create_Json()
timestamp = str(TL.data["timestamp"])
path_file = self.project_path +timestamp+ "//" + timestamp + ".jpg"
answer_from_server = TL.upload_measurement( path_file, timestamp, TL.get_temperature(), TL.get_humidity())
print(answer_from_server)
try:
if (answer_from_server["message"]== "Measurement uploaded successfully"):
TL.del_Folder(timestamp)
except Exception as e:
print(e)
def upload_measurement(self, image_path, timestamp, temperature, humidity):
data = {
'timestamp': timestamp,
'temperature': temperature,
'humidity': humidity
}
#print(data)
#print(image_path)
files = {
'image': (image_path, open(image_path, 'rb'), 'image/jpeg')
}
try:
response = requests.post(self.url, headers=self.headers, data=data, files=files)
response.raise_for_status()
response_data = response.json()
return response_data
except requests.exceptions.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
except Exception as err:
print(f"Other error occurred: {err}")
if __name__ == "__main__":
print("FICHIER TIMELAPSE")
TL = TimeLapse()
TL.Singleroutin()
time.sleep(1)
""" FONCTIONNEL
TL.get_Hygrodata()
TL.pick_Picture()
TL.create_Json()
try:
TL.del_Folder("13_03_2025_12_07_56")
except:
pass
"""
# print(TL.get_data())
# TL.get_Picture()
# print("Temper/home/timelapse/Documents/Time_Lapse/PROJECTature is : ", TL.get_temperature())
# print("Humidity is : ", TL.get_humidity())
# datas = TL.get_Alldic()
# print(datas)

127
old_files/Time_Lapse_NoConnection.py Normal file
View File

@@ -0,0 +1,127 @@
import time
from datetime import datetime
import picamera2 as pc
import smbus2
import os
import json
import shutil
import requests
class TimeLapse:
def __init__(self):
self.url = "https://timelapse.kerboul.me/api/uploadmeasurement"
self.headers = {
'accept': 'application/json'
}
self.data = {
"timestamp": "",
"temperature": "",
'humidity': "",
}
self.project_path = "//home//timelapse//Documents//Time_Lapse//PROJECT//"
def create_Folder(self):
filename = self.data["timestamp"]
os.mkdir(self.project_path + filename)
def create_Json(self):
filename = self.project_path + self.data["timestamp"] + "//" + self.data["timestamp"] + ".json"
with open(filename, "w") as file:
json.dump(self.get_Alldic(), file)
def get_Alldic(self):
return self.data
def get_Hygrodata(self):
bus = smbus2.SMBus(1)
ans = bus.write_byte(0x44, 0xFD)
time.sleep(0.1)
ans=smbus2.i2c_msg.read(0x44,6)
bus.i2c_rdwr(ans)
data = list(ans)
t_ticks = data[0]*256 + data[1] #Pas de CRC utilisé
rh_ticks = data[3]*256 + data[4] #Pas de CRC utilisé
self.data["temperature"] = (175*t_ticks)/65535. -45
self.data["humidity"] = (125*rh_ticks)/65535. -6
return self.data
def get_humidity(self):
return self.data["humidity"]
def get_temperature(self):
return self.data["temperature"]
def get_TimeStamp(self): #ATTENTION AU NOMMAGE DES FICHIERS !!! PAS DE : ET ESPACES
now = datetime.now()
dt_string = now.strftime("%Y-%m-%d %H:%M:%S")
print("date and time =", str(dt_string))
self.data["timestamp"]= str(dt_string)
return str(dt_string)
def pick_Picture(self):
cam = pc.Picamera2()
conf = cam.create_preview_configuration(main={"size":(800, 600)})
cam.configure(conf)
cam.start_preview(pc.Preview.QTGL)
cam.start()
time.sleep(0.1)
timestamp = self.get_TimeStamp()
path = self.project_path + timestamp + "//" + timestamp+".jpg"
self.create_Folder()
image = cam.capture_file(path)
cam.close()
def del_Folder(self, name):
shutil.rmtree(self.project_path + name)
def Singleroutin(self):
TL.get_Hygrodata()
TL.pick_Picture()
TL.create_Json()
timestamp = str(TL.data["timestamp"])
path_file = self.project_path +timestamp+ "//" + timestamp + ".jpg"
#answer_from_server = TL.upload_measurement( path_file, 1, timestamp, TL.get_temperature(), TL.get_humidity())
#print(answer_from_server)
#TL.del_Folder("PROJECT//" +timestamp)
def upload_measurement(self, image_path, project_id, timestamp, temperature, humidity):
data = {
'projectId': project_id,
'timestamp': timestamp,
'temperature': temperature,
'humidity': humidity
}
files = {
'image': (image_path, open(image_path, 'rb'), 'image/jpeg')
}
try:
response = requests.post(self.url, headers=self.headers, data=data, files=files)
response.raise_for_status()
response_data = response.json()
return response_data
except requests.exceptions.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
except Exception as err:
print(f"Other error occurred: {err}")
if __name__ == "__main__":
print("FICHIER TIMELAPSE")
TL = TimeLapse()
TL.Singleroutin()
time.sleep(1)
""" FONCTIONNEL
TL.get_Hygrodata()
TL.pick_Picture()
TL.create_Json()
"""
try:
TL.del_Folder("13_03_2025_12_07_56")
except:
pass
# print(TL.get_data())
# TL.get_Picture()
# print("Temperature is : ", TL.get_temperature())
# print("Humidity is : ", TL.get_humidity())
# datas = TL.get_Alldic()
# print(datas)

14
old_files/get_from_server.py Normal file
View File

@@ -0,0 +1,14 @@
import requests
url_requete = "https://timelapse.kerboul.me/api/camera/status"
try:
response = requests.get(url_requete)
if response.status_code == 200:
camera_status = response.json()
print(camera_status)
else:
print("mauvais code")
except requests.exceptions.RequestException as e:
print("erreur API ou internet")

3
old_files/script_SANSDEMARRAGE.sh Normal file
View File

@@ -0,0 +1,3 @@
#!/bin/bash
python Time_Lapse_NoConnection.py
fi