brammp
f6c2b9dd5b
* Ability to use full path for image path * Ability to configure the wait time between images * Added protection from escaping the image directory
325 lines
9.9 KiB
Python
325 lines
9.9 KiB
Python
import http.server
|
|
import socketserver
|
|
import os
|
|
import json
|
|
import requests
|
|
from PIL import Image
|
|
from PIL.ExifTags import TAGS
|
|
import urllib.parse
|
|
from datetime import datetime
|
|
import mimetypes
|
|
from pathlib import Path
|
|
|
|
# --
|
|
# Configuration
|
|
# --
|
|
with open("config.json") as f:
|
|
cfg = json.load(f)
|
|
|
|
#--
|
|
# Web Router
|
|
#--
|
|
class SlideshowHandler(http.server.SimpleHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path == "/":
|
|
self.send_slideshow_page()
|
|
elif self.path == "/config.js":
|
|
self.send_clientconf()
|
|
elif self.path == "/images.json":
|
|
self.send_image_list()
|
|
elif self.path == "/weather":
|
|
if cfg['WeatherProvider'] == "HomeAssistant":
|
|
self.get_HASSweather()
|
|
else:
|
|
self.get_METNOweather()
|
|
elif self.path.startswith("/exif"):
|
|
self.send_exif_data()
|
|
elif self.path.startswith("/img"):
|
|
self.send_image()
|
|
else:
|
|
self.send_error(404, "Not found")
|
|
|
|
#--
|
|
# Main Page
|
|
#--
|
|
def send_slideshow_page(self):
|
|
html_path = "show.html"
|
|
|
|
if os.path.exists(html_path):
|
|
with open(html_path, "rb") as f:
|
|
content = f.read()
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "text/html")
|
|
self.send_header("Content-length", len(content))
|
|
self.end_headers()
|
|
self.wfile.write(content)
|
|
|
|
#--
|
|
# Image depot
|
|
#--
|
|
def send_image(self):
|
|
parsed = urllib.parse.urlparse(self.path)
|
|
params = urllib.parse.parse_qs(parsed.query)
|
|
filename = params.get("file", [None])[0]
|
|
|
|
if not filename:
|
|
self.send_error(400, "Missing ?file= parameter")
|
|
return
|
|
|
|
base_dir = Path(cfg["IMAGE_DIR"]).resolve()
|
|
print(base_dir)
|
|
file_path = (base_dir / filename).resolve()
|
|
try:
|
|
file_path.relative_to(base_dir)
|
|
except ValueError:
|
|
self.send_error(403, "Forbidden")
|
|
return
|
|
|
|
content_type, _ = mimetypes.guess_type(filename)
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", content_type or "application/octet-stream")
|
|
self.end_headers()
|
|
with open(cfg['IMAGE_DIR'] + "/" + filename, 'rb') as file:
|
|
self.wfile.write(file.read())
|
|
|
|
#--
|
|
# Client config
|
|
#--
|
|
def send_clientconf(self):
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "text/javascript")
|
|
self.end_headers()
|
|
response = f"""const ImgDuration = "{cfg["IMAGE_TIMEOUT"]}";"""
|
|
self.wfile.write(response.encode("utf-8"))
|
|
|
|
#--
|
|
# Image list
|
|
#--
|
|
def send_image_list(self):
|
|
files = os.listdir(cfg['IMAGE_DIR'])
|
|
images = [
|
|
f"{f}"
|
|
for f in files
|
|
if f.lower().endswith((".jpg", ".jpeg", ".png", ".gif", ".webp"))
|
|
]
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(images).encode("utf-8"))
|
|
print(self)
|
|
#--
|
|
# Weather - From HASS
|
|
#--
|
|
def get_HASSweather(self):
|
|
try:
|
|
url = f"{cfg['HA_URL']}/api/states/{cfg['HA_WEATHER_ENTITY']}"
|
|
headers = {
|
|
"Authorization": f"Bearer {cfg['HA_TOKEN']}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
ha_response = requests.get(url, headers=headers, timeout=5)
|
|
ha_response.raise_for_status()
|
|
data = ha_response.json()
|
|
|
|
icon_map = {
|
|
"sunny": "clearsky_day",
|
|
"clear-night": "clearsky_night",
|
|
"cloudy": "cloudy",
|
|
"partlycloudy": "fair_day",
|
|
"rainy": "rain",
|
|
"pouring": "heavyrain",
|
|
"lightning": "lightrainandthunder",
|
|
"lightning-rainy": "rainandthunder",
|
|
"snowy": "snow",
|
|
"snowy-rainy": "sleet",
|
|
"fog": "fog"
|
|
}
|
|
|
|
symbol = icon_map.get(data.get("state"), "cloudy")
|
|
|
|
result = {
|
|
#"state": data.get("state"),
|
|
"temperature": data["attributes"].get("temperature"),
|
|
"symbol": 'https://raw.githubusercontent.com/metno/weathericons/refs/heads/main/weather/svg/'+ symbol + '.svg'
|
|
#"humidity": data["attributes"].get("humidity"),
|
|
#"pressure": data["attributes"].get("pressure"),
|
|
#"wind_speed": data["attributes"].get("wind_speed"),
|
|
#"forecast": data["attributes"].get("forecast"),
|
|
}
|
|
|
|
self._send_json(result)
|
|
|
|
except Exception as e:
|
|
self._send_json({"error": str(e)}, status=500)
|
|
|
|
#--
|
|
# Weather - From Net.no
|
|
#--
|
|
def get_METNOweather(self):
|
|
try:
|
|
url = f"https://api.met.no/weatherapi/locationforecast/2.0/compact?lat={cfg['METNO_LAT']}&lon={cfg['METNO_LON']}"
|
|
headers = {
|
|
"User-Agent": "AmbientDisplay/1.0"
|
|
}
|
|
|
|
metno_response = requests.get(url, headers=headers, timeout=5)
|
|
metno_response.raise_for_status()
|
|
data = metno_response.json()
|
|
result = {
|
|
"temperature": data.get("properties", {}).get("timeseries", [{}])[0].get("data", {}).get("instant", {}).get("details", {}).get("air_temperature"),
|
|
"symbol": 'https://raw.githubusercontent.com/metno/weathericons/refs/heads/main/weather/svg/'+ data.get("properties", {}).get("timeseries", [{}])[0].get("data", {}).get("next_1_hours", {}).get("summary", {}).get("symbol_code") + '.svg'
|
|
}
|
|
|
|
self._send_json(result)
|
|
|
|
except Exception as e:
|
|
self._send_json({"error": str(e)}, status=500)
|
|
|
|
#--
|
|
# Weather - Json answer
|
|
#--
|
|
def _send_json(self, obj, status=200):
|
|
payload = json.dumps(obj).encode("utf-8")
|
|
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
#--
|
|
# Image info
|
|
#--
|
|
def send_exif_data(self):
|
|
parsed = urllib.parse.urlparse(self.path)
|
|
params = urllib.parse.parse_qs(parsed.query)
|
|
filename = params.get("file", [None])[0]
|
|
|
|
if not filename:
|
|
self.send_error(400, "Missing ?file= parameter")
|
|
return
|
|
|
|
base_dir = Path(cfg["IMAGE_DIR"]).resolve()
|
|
print(base_dir)
|
|
file_path = (base_dir / filename).resolve()
|
|
try:
|
|
file_path.relative_to(base_dir)
|
|
except ValueError:
|
|
self.send_error(403, "Forbidden")
|
|
return
|
|
|
|
|
|
filepath = cfg['IMAGE_DIR'] +"/"+ filename
|
|
|
|
if not os.path.exists(filepath):
|
|
self.send_error(404, "File not found")
|
|
return
|
|
|
|
capture_date = self.get_capture_date(filepath)
|
|
GeoData = self.get_GeoData(filepath)
|
|
|
|
response = {
|
|
"capture_date": capture_date,
|
|
"geo_data": GeoData
|
|
}
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(response).encode("utf-8"))
|
|
|
|
#--
|
|
# Image info -- Date
|
|
#--
|
|
def get_capture_date(self, path):
|
|
try:
|
|
img = Image.open(path)
|
|
exif = img._getexif()
|
|
if not exif:
|
|
return None
|
|
|
|
for tag_id, value in exif.items():
|
|
tag = TAGS.get(tag_id, tag_id)
|
|
if tag in ("DateTimeOriginal", "DateTime"):
|
|
try:
|
|
dt = datetime.strptime(value, "%Y:%m:%d %H:%M:%S")
|
|
return dt.isoformat()
|
|
except Exception:
|
|
return None
|
|
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
#--
|
|
# Image info -- GeoData/GPS
|
|
#--
|
|
def get_GeoData(self, path):
|
|
|
|
def convert_to_degrees(value):
|
|
def to_float(x):
|
|
return x[0] / x[1] if isinstance(x, tuple) else float(x)
|
|
|
|
d = to_float(value[0])
|
|
m = to_float(value[1])
|
|
s = to_float(value[2])
|
|
return d + (m / 60.0) + (s / 3600.0)
|
|
|
|
def get_gecoded(lon, lat):
|
|
url = f"{cfg['PhotonAPI']}?lang=en&lat={lat}&lon={lon}"
|
|
headers = {
|
|
"X-Api-Key": cfg['PhotonAPI_TOKEN']
|
|
}
|
|
|
|
geoapi_response = requests.get(url, headers=headers, timeout=30)
|
|
geoapi_response.raise_for_status()
|
|
return geoapi_response.json()
|
|
|
|
|
|
try:
|
|
img = Image.open(path)
|
|
exif = img._getexif()
|
|
if not exif:
|
|
return None
|
|
|
|
for tag_id, value in exif.items():
|
|
tag = TAGS.get(tag_id, tag_id)
|
|
if tag in ("GPSInfo"):
|
|
try:
|
|
lat = convert_to_degrees(value[2])
|
|
lat_ref = value[1]
|
|
if lat_ref == "S":
|
|
lat = -lat
|
|
|
|
lon = convert_to_degrees(value[4])
|
|
lon_ref = value[3]
|
|
if lon_ref == "W":
|
|
lon = -lon
|
|
|
|
GeoDATA = get_gecoded(lon, lat)
|
|
return {
|
|
"GeoData": GeoDATA,
|
|
"latitude": lat,
|
|
"longitude": lon
|
|
}
|
|
|
|
except Exception:
|
|
return None
|
|
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
#--
|
|
# WebServer
|
|
#--
|
|
if __name__ == "__main__":
|
|
os.chdir(os.path.dirname(os.path.abspath(__file__)))
|
|
print(f"Serving slideshow on http://0.0.0.0:{cfg['PORT']}")
|
|
socketserver.TCPServer.allow_reuse_address = True
|
|
with socketserver.TCPServer(("", cfg['PORT']), SlideshowHandler) as httpd:
|
|
httpd.serve_forever()
|