#! /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()))