139 lines
4.6 KiB
Python
Executable File
139 lines
4.6 KiB
Python
Executable File
#! /usr/bin/env python3
|
|
# Installation: pip install exif asyncio blinkpy
|
|
import argparse
|
|
import sys
|
|
import logging
|
|
import getpass
|
|
import asyncio
|
|
import exif
|
|
from datetime import datetime
|
|
from blinkpy.blinkpy import Blink
|
|
from blinkpy.auth import Auth
|
|
from blinkpy.helpers.util import json_load
|
|
from pathlib import Path
|
|
from PIL import Image, ImageDraw
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def decdeg2dms(dd):
|
|
mult = -1 if dd < 0 else 1
|
|
mnt, sec = divmod(abs(dd) * 3600, 60)
|
|
deg, mnt = divmod(mnt, 60)
|
|
return mult * deg, mult * mnt, mult * sec
|
|
|
|
|
|
async def main() -> int:
|
|
# Handle program arguments
|
|
ap = argparse.ArgumentParser(
|
|
prog="blink-fetch", description="Fetch an image from a Blink camera"
|
|
)
|
|
ap.add_argument("--username", help="Blink username", required=True)
|
|
ap.add_argument("--password", help="Blink password")
|
|
ap.add_argument("--camera-id", help="Camera ID", default="155295")
|
|
ap.add_argument("--output-dir", help="Output directory", default="~/Pictures/blink")
|
|
ap.add_argument(
|
|
"--copy-latest", help="Copies the latest frame to this path", type=Path
|
|
)
|
|
ap.add_argument(
|
|
"--no-2fa", help="Don't try to get 2FA credentials", action="store_true"
|
|
)
|
|
ap.add_argument("--no-exif", help="Don't write EXIF data", action="store_true")
|
|
ap.add_argument("--exif-camera", help="Camera name", default="Blink Mini")
|
|
ap.add_argument(
|
|
"--exif-latitude", "--exif-lat", help="Camera latitude (Decimal Degrees)"
|
|
)
|
|
ap.add_argument(
|
|
"--exif-longitude", "--exif-lng", help="Camera longitude (Decimal Degrees)"
|
|
)
|
|
ap.add_argument(
|
|
"-v", "--verbose", help="Enable verbose logging", action="store_true"
|
|
)
|
|
args = ap.parse_args()
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(levelname)s: %(message)s",
|
|
)
|
|
|
|
# Ask for the password if it wasn't provided
|
|
if args.password is None:
|
|
args.password = getpass.getpass(prompt="Blink Password: ")
|
|
|
|
# Authenticate with Blink servers
|
|
auth = Auth(
|
|
{"username": args.username, "password": args.password}, no_prompt=args.no_2fa
|
|
)
|
|
blink = Blink()
|
|
blink.auth = auth
|
|
await blink.start()
|
|
|
|
# Find the requested camera
|
|
for name, camera in blink.cameras.items():
|
|
logger.debug(f"Found camera: {name} ({camera.attributes['camera_id']})")
|
|
if camera.attributes["camera_id"] == args.camera_id:
|
|
logger.info("Found requested camera")
|
|
break
|
|
else:
|
|
logger.error("Could not find requested camera")
|
|
return 1
|
|
|
|
# Fetch the image
|
|
logger.info("Fetching image")
|
|
await camera.snap_picture()
|
|
await blink.refresh()
|
|
|
|
# Create the output directory if it doesn't exist
|
|
now = datetime.now()
|
|
out_file = (
|
|
Path(args.output_dir).expanduser()
|
|
/ f"camera_{args.camera_id}.{now.strftime('%Y%m%d_%H%M%S')}.jpg"
|
|
)
|
|
out_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
logger.info(f"Writing image to: {out_file}")
|
|
await camera.image_to_file(str(out_file))
|
|
|
|
# Draw the timestamp on the image in the bottom left corner
|
|
image = Image.open(out_file)
|
|
draw = ImageDraw.Draw(image)
|
|
draw.text((0, image.height - 10), now.strftime("%Y-%m-%d %H:%M:%S"), fill=(255, 255, 255), stroke_width=2, stroke_fill=(0, 0, 0))
|
|
image.save(out_file)
|
|
|
|
# Handle EXIF data
|
|
if not args.no_exif:
|
|
logger.info("Re-reading image to inject EXIF data")
|
|
with open(out_file, "rb") as f:
|
|
image = exif.Image(f)
|
|
|
|
# Set the camera type
|
|
image.model = args.exif_camera
|
|
|
|
# If the user provided a latitude and longitude, set it
|
|
# if args.exif_latitude and args.exif_longitude:
|
|
# image.gps_latitude = decdeg2dms(float(args.exif_latitude))
|
|
# image.gps_longitude = decdeg2dms(float(args.exif_longitude))
|
|
# image.gps_latitude_ref = "N"
|
|
# image.gps_longitude_ref = "W"
|
|
|
|
# Set the timestamp
|
|
image.datetime_original = now.strftime(exif.DATETIME_STR_FORMAT)
|
|
|
|
# Write the EXIF data back to the file
|
|
logger.info("Writing EXIF data")
|
|
with open(out_file, "wb") as f:
|
|
f.write(image.get_file())
|
|
|
|
# If we were asked to copy the latest frame, do so
|
|
if args.copy_latest:
|
|
logger.info(f"Copying latest frame to: {args.copy_latest}")
|
|
args.copy_latest.parent.mkdir(parents=True, exist_ok=True)
|
|
args.copy_latest.write_bytes(out_file.read_bytes())
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(asyncio.run(main()))
|