Building a Webcam on a Raspberry Pi - the Code

Wed, Jun 28, 2023 5-minute read

The code is all in Python is not that complicated. If you followed my previous guide on beginning web scraping with Python then there will be nothing here to catch you out.

The flow is reasonably simple:

  1. Open a connection to the camera
  2. Grab an image
  3. Store a copy of it (possibly on an external USB stick)
  4. Watermark that image with something
  5. Upload it to Azure, always with the same name, so that the website displaying it doesn’t need to know the name of the latest file

The solution (moreorless) consists of 2 files:

  • picam.py - where the bulk of the code is
  • config.py - where the… config… is.

The Azure part uses the example from the docs if you want to see more detail.

Config.py

This should be fairly self-explanatory.

The only additional thing here is the import of a file called ‘sensitive.py’ - this should contain the connection string (storage_conn_str) to your blob store, which you get from the Azure Portal. I separate it out like this so that it’s simple to Git ignore the file so it’s not stored in source control.

## config
import datetime as dt
import os
import sensitive

# general stuff
latest_filepath = '/path/to/where/images/should/be/stored'
latest_filename = 'capture.jpg'
latest_filename_fullpath = os.path.join(latest_filepath, latest_filename)
free_space_threshold = 500

watermark_image_filename = 'logo_small.png'
watermark_image_filename_fullpath = os.path.join(latest_filepath, watermark_image_filename)

# watermarked image (for website)
latest_watermarked_filename = 'latest.jpg'
latest_watermarked_filename_fullpath = os.path.join(latest_filepath, latest_watermarked_filename)

# file handling
archive_filename_fullpath = '/path/to/where/images/should/be/stored/pic-%s.jpg' % dt.datetime.now().strftime('%Y-%m-%d-%H%M%S')
overlay_text = "some words to overlay"

# azure stuff
storage_account_name = 'storagename'
storage_container_name = 'containername'
storage_connection_string = sensitive.storage_conn_str

Picam.py (or whatever you want to call it)

The code should work moreorless OK.

Points to note:

If you dont want to copy to an external USB drive, comment out the section there. The mount point will typically be /media/* which you can set in the config file. This is why there’s an additional housekeeping section to monitor the USB drive filling up.

The reason I use an external USB drive as opposed to the main SD card is in the event the card does fill up, you don’t accidentally brick your whole device.

#!/usr/bin/python3
import config
import watermark

from picamera import PiCamera
from time import sleep
from shutil import copyfile, disk_usage
from azure.storage.blob import BlobServiceClient, PublicAccess

import logging

def ProcessError(msg):
    logging.error(msg)
    exit

def ProcessWarning(msg):
    logging.warning(msg)
    
## setup logging
logging.basicConfig(handlers=[logging.FileHandler('picam.log'),logging.StreamHandler()], format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.DEBUG)
logging.info('Starting')

## grab an image
logging.debug('Initialising camera')
camera = PiCamera()
camera.resolution = (2592,1944)
camera.rotation = 0
camera.exposure_mode = 'auto'
camera.annotate_text = config.overlay_text
logging.debug('Initialising camera done')

sleep(5)

logging.info('Capturing image')
try:
    camera.capture(config.latest_filename_fullpath)
except Exception as e:
    ProcessError('Error capturing image' + str(e))    

## copy it to USB drive
logging.info('Copying image to USB')
try:
    copyfile(config.latest_filename_fullpath, config.archive_filename_fullpath)
except Exception as e:
    ProcessError('Error copying image to USB' + str(e))

## watermark it
logging.info('Watermarking image')
try:
    watermark.AddWatermarks(config.latest_filename_fullpath, config.watermark_image_filename_fullpath, config.latest_watermarked_filename_fullpath)
except Exception as e:
    ProcessError('Error watermarking image' + str(e))
    
## send latest to Azure
logging.info('Sending to Azure')
try:
    logging.info('Creating block blob service')
    blob_service_client = BlobServiceClient.from_connection_string(config.storage_connection_string)
    
    logging.info('Creating blob')  
    blob_client = blob_service_client.get_blob_client(container=config.storage_container_name, blob=config.latest_watermarked_filename)
    
    logging.info('Uploading blob')  
    #block_blob_service.create_blob_from_path(config.storage_container_name, config.latest_watermarked_filename, config.latest_watermarked_filename_fullpath)
    with open(file=config.latest_watermarked_filename_fullpath, mode="rb") as data:
        blob_client.upload_blob(data, overwrite=True)

except Exception as e:
    ProcessError('Error doing Azure: ' + str(e))

## check if USB drive is full
logging.info('Housekeeping USB drive')
try:
    total, used, free = disk_usage("/media/usb")
    freeMb = free // (1024.0**2)

    if freeMb < config.free_space_threshold:
        ProcessWarning('USB drive is starting to get a little full. Only {} mb left.'.format(freeMb))
except:
    ProcessError('Error housekeeping')

logging.info('Done')

If you want to use the watermarking, you will also need watermark.py…

watermark.py

## watermark.py
from PIL import Image
from PIL import ImageDraw
from PIL import ImageFont
import datetime as dt
import logging

def AddWatermarks(input_image_path, watermark_image_path, output_image_path):
    
    logging.info(input_image_path)
    logging.info(watermark_image_path)
    logging.info(output_image_path)

    # open images
    base_image = Image.open(input_image_path)
    base_width, base_height = base_image.size

    watermark = Image.open(watermark_image_path)
    wmark_width, wmark_height = watermark.size
    
    #add logo image
    transparent = Image.new('RGBA', (base_width, base_height), (0,0,0,0))
    transparent.paste(base_image, (0,0))
    transparent.paste(watermark, (base_width - wmark_width - 10, base_height - wmark_height - 10), mask=watermark)
        
    # make the image editable
    drawing = ImageDraw.Draw(transparent)

    # get date
    overlay_text = "%s" % dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    # add text
    black = (255, 255, 255)
    font = ImageFont.truetype("DejaVuSans.ttf", 30)
    drawing.text((10,base_height - 40), overlay_text, fill=black, font=font)   

    # show it and finish
    #transparent.show()
    transparent = transparent.convert('RGB')
    transparent.save(output_image_path)

With the above written you should be able to run it, assuming you’re in the right directory

python picam.py

which should create an image in the output directory.

From there, the last thing to do is to schedule it to run at a suitable frequency, which can be done with cron

crontab -e

If you want it to run e.g., every 10 mins, your cron will look like this

*/10 * * * * /usr/bin/python /path/to/your/picam.py

Maybe you only want it to run during daylight (ish) hours? You could do e.g.,

*/10 9-21 * * * /usr/bin/python /path/to/your/picam.py

which will mean it will run every 10 mins but only between 9am and 9pm.

And that’s about it! Hope this is useful. Leave a comment below with any comments, thoughts or issues.

Streaming?

If you want to add streaming to your install on the local network, there is an excellent tutorial with code here.

Remote editing

I use VS Code for stuff like this and after struggling with the whole ‘make change / push / pull’ or SCP-ing files to the device, finally sorted it and found this great article (where the opening paragraph literally describes it!) on how to enable remote editing on a Raspberry PI in VS Code. The article is step by step, but in brief find an extension for VS Code called “Remote - SSH Extension”. Excellent.

Posts in this Series