import hashlib import requests from requests.auth import HTTPDigestAuth import urllib3 import config import logging import time from datetime import datetime, date, timedelta, timezone, time as dt_time from zoneinfo import ZoneInfo from email.utils import parsedate_to_datetime import re from pathlib import Path import cv2 import numpy as np import sys import logging from urllib.parse import quote import json import pathlib import base64 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) TZ_ALIASES = { "uk": "Europe/London", "gb": "Europe/London", "london": "Europe/London", "belgium": "Europe/Brussels", "be": "Europe/Brussels", "brussels": "Europe/Brussels", "utc": "UTC", "z": "UTC", } def login(base_url, username, password, user_nonce, user_key, client_name="Jupyter Notebook", integration_identifier="", verify_tls=False): url = f"{base_url}/login" authorization_token = generate_authorization_token( user_nonce=user_nonce, user_key=user_key, integration_identifier=integration_identifier, ) payload = { "username": username, "password": password, "clientName": client_name, "authorizationToken": authorization_token, } response = requests.post(url, json=payload, verify=verify_tls, timeout=30) response.raise_for_status() data = response.json() if data.get("status") != "success": raise RuntimeError(f"Login failed: {data}") return data["result"]["session"], data def generate_authorization_token(user_nonce, user_key, integration_identifier=""): timestamp = str(int(time.time())) digest = hashlib.sha256(f"{timestamp}{user_key}".encode("utf-8")).hexdigest().lower() token = f"{user_nonce}:{timestamp}:{digest}" if integration_identifier: token += f":{integration_identifier}" return token def get_cameras(base_url, session, verify_tls=False, verbosity="LOW"): url = f"{base_url}/cameras" params = { "session": session, "verbosity": verbosity, } response = requests.get(url, params=params, verify=verify_tls, timeout=30) response.raise_for_status() data = response.json() # Extract simplified camera list cameras = data.get("result", {}).get("cameras", []) simplified = [ { "id": cam["id"], "name": cam["name"], "location": cam.get("location"), } for cam in cameras if cam.get("active") and cam.get("connectionState") == "CONNECTED" ] return {"raw": data,"cameras": simplified} def normalize_tz(tz_name_or_alias): if tz_name_or_alias is None: return None key = tz_name_or_alias.strip().lower() return TZ_ALIASES.get(key, tz_name_or_alias) def parse_user_time_to_utc( value, *, default_date=None, site_timezone="Europe/London", input_timezone=None, ): site_timezone = normalize_tz(site_timezone) input_timezone = normalize_tz(input_timezone) value = value.strip() # Case 1: time only if re.fullmatch(r"\d{1,2}:\d{2}(:\d{2})?", value): if default_date is None: raise ValueError("default_date is required when value is only HH:MM:SS") if isinstance(default_date, str): try: default_date = date.fromisoformat(default_date) # YYYY-MM-DD except ValueError: default_date = datetime.strptime(default_date, "%d/%m/%Y").date() # DD/MM/YYYY if value.count(":") == 1: value += ":00" local_time = dt_time.fromisoformat(value) naive_dt = datetime.combine(default_date, local_time) tz = ZoneInfo(input_timezone or site_timezone) aware_dt = naive_dt.replace(tzinfo=tz) else: normalized = value.replace(" ", "T") dt = datetime.fromisoformat(normalized) if dt.tzinfo is not None: aware_dt = dt else: tz = ZoneInfo(input_timezone or site_timezone) aware_dt = dt.replace(tzinfo=tz) return aware_dt.astimezone(timezone.utc) def compute_wep_extract_timestamp( value, *, default_date=None, site_timezone="Europe/London", input_timezone=None, wep_offset_seconds=0.0, match_mode="c", ): user_time_utc = parse_user_time_to_utc( value, default_date=default_date, site_timezone=site_timezone, input_timezone=input_timezone, ) corrected_wep_time_utc = user_time_utc + timedelta(seconds=wep_offset_seconds) iso = corrected_wep_time_utc.isoformat(timespec="milliseconds").replace("+00:00", "Z") if match_mode: return f"{iso},{match_mode}" return iso def estimate_wep_offset_from_health(base_url, verify_tls=False): url = f"{base_url.rstrip('/')}/health" t0 = datetime.now(timezone.utc) response = requests.get(url, verify=verify_tls, timeout=15) t1 = datetime.now(timezone.utc) response.raise_for_status() date_header = response.headers.get("Date") if not date_header: raise RuntimeError("No HTTP Date header returned") wep_time_utc = parsedate_to_datetime(date_header).astimezone(timezone.utc) query_midpoint_utc = t0 + (t1 - t0) / 2 offset_seconds = (wep_time_utc - query_midpoint_utc).total_seconds() logging.debug(f"raw_http_date:{date_header}") logging.debug (f'wep_time_utc:{wep_time_utc}') return { "wep_time_utc": wep_time_utc, "query_midpoint_utc": query_midpoint_utc, "offset_seconds": offset_seconds, "rtt_seconds": (t1 - t0).total_seconds(), "raw_http_date": date_header, } def build_backward_chunks(global_start_utc, total_duration_sec, chunk_duration_sec=None): """ Build chunks ending at global_start_utc + total_duration_sec. Returned list is ordered from most recent to oldest. Example: total_duration_sec = 315 chunk_duration_sec = 30 Gives: latest 30s chunk ... oldest 45s chunk So we avoid a final tiny 15s chunk. """ total_duration_sec = int(total_duration_sec) if chunk_duration_sec is None: return [ ( global_start_utc, global_start_utc + timedelta(seconds=total_duration_sec), total_duration_sec, ) ] chunk_duration_sec = int(chunk_duration_sec) if chunk_duration_sec <= 0: raise ValueError("chunk_duration_sec must be a positive integer or None") if total_duration_sec <= chunk_duration_sec: return [ ( global_start_utc, global_start_utc + timedelta(seconds=total_duration_sec), total_duration_sec, ) ] n_full = total_duration_sec // chunk_duration_sec remainder = total_duration_sec % chunk_duration_sec if remainder == 0: durations_chronological = [chunk_duration_sec] * n_full else: # Merge the small remainder into the oldest chunk. # Example: 315s with 30s chunks -> 45s + 9 x 30s durations_chronological = [chunk_duration_sec + remainder] + [ chunk_duration_sec ] * (n_full - 1) intervals = [] current_start = global_start_utc for duration in durations_chronological: current_end = current_start + timedelta(seconds=duration) intervals.append((current_start, current_end, duration)) current_start = current_end # Most recent first return list(reversed(intervals)) def format_wep_timestamp(dt_utc, original_wep_t=None): """ Format a UTC datetime back into the WEP timestamp format. If original_wep_t contains a suffix after the first comma, for example: 2026-06-02T08:30:51.000Z,c the suffix is preserved. """ if dt_utc.tzinfo is None: dt_utc = dt_utc.replace(tzinfo=timezone.utc) dt_utc = dt_utc.astimezone(timezone.utc) iso = dt_utc.isoformat(timespec="milliseconds").replace("+00:00", "Z") if original_wep_t and "," in original_wep_t: suffix = original_wep_t.split(",", 1)[1] return f"{iso},{suffix}" return iso def seconds_to_wep_duration(seconds: int) -> str: seconds = int(seconds) hours = seconds // 3600 seconds %= 3600 minutes = seconds // 60 seconds %= 60 return f"P0Y0M0DT{hours}H{minutes}M{seconds}S" def get_fmp4_media_range_from_timestamp(base_url, session, camera_id,timestamp, duration, verify_tls=False): iso_duration = seconds_to_wep_duration(duration) url = f"{base_url}/media" params = { "session": session, "cameraId": camera_id, "format": "fmp4", "range":iso_duration, "quality":"high", #"size":"720,720", "size":"2048,1536", #"size":"2048,2048", "t": timestamp } response = requests.get(url, params=params, verify=verify_tls, timeout=60) response.raise_for_status() return response.content, response def expand_polygon_from_center( polygon: np.ndarray, expand_percent: float, image_width: int, image_height: int, ) -> np.ndarray: """ Expand a polygon from its centroid while keeping it inside the image. Parameters ---------- polygon: np.array of shape (N, 2), dtype int or float. Example: [[x1, y1], [x2, y2], ...] expand_percent: Percentage expansion. Example: 20 means +20% 50 means +50% 100 means 2x size image_width: Frame width in pixels. image_height: Frame height in pixels. Returns ------- expanded_polygon: np.array of shape (N, 2), dtype np.int32 """ polygon = np.asarray(polygon, dtype=np.float32) if polygon.ndim != 2 or polygon.shape[1] != 2: raise ValueError("polygon must have shape (N, 2)") if polygon.shape[0] < 3: raise ValueError("polygon must have at least 3 vertices") scale = 1.0 + expand_percent / 100.0 # Geometric center of the vertices center = polygon.mean(axis=0) # Expand around center expanded = center + (polygon - center) * scale # Keep inside image boundaries expanded[:, 0] = np.clip(expanded[:, 0], 0, image_width - 1) expanded[:, 1] = np.clip(expanded[:, 1], 0, image_height - 1) return np.round(expanded).astype(np.int32) def submit_video_to_stream_cloud( api_token: str, upload_url: str, camera_id: str | None = None, regions: list[str] | None = None, mmc: bool | None = None, endpoint: str = config.STREAM_CLOUD_URL, timeout: int = 120, ) -> dict: # Multipart form-data request, matching the documented curl -F style headers = { "Authorization": f"Token {api_token}", } files = [ ("upload_url", (None, upload_url)), ] if camera_id: files.append(("camera_id", (None, camera_id))) if regions: for region in regions: files.append(("regions", (None, region))) if mmc is not None: files.append(("mmc", (None, "true" if mmc else "false"))) response = requests.post( endpoint, headers=headers, files=files, timeout=timeout, ) if response.status_code not in (200, 201): raise RuntimeError(f"HTTP {response.status_code}: {response.text}") return response.json() def build_video_url(filename: str, public_base_url: str) -> str: # URL-encode filename in case it contains spaces or special chars return public_base_url.rstrip("/") + "/" + quote(filename) def check_video_url(video_url: str, timeout: int = 20) -> None: # Quick check that Plate Recognizer will be able to download the file r = requests.head(video_url, allow_redirects=True, timeout=timeout) if r.status_code >= 400: raise RuntimeError(f"Video URL not reachable: {r.status_code} - {video_url}") logging.debug("URL OK") logging.debug("Status :" + str(r.status_code)) logging.debug("Content-Type :" + r.headers.get("Content-Type")) logging.debug("Content-Length:" + r.headers.get("Content-Length")) def wait_for_folder_pattern(parent_folder: str, prefix: str, timeout: int = 60, interval: int = 2) -> pathlib.Path | None: """ Periodically checks if a folder starting with `prefix` exists inside parent_folder. :param parent_folder: The directory to watch (e.g., 'folderA') :param prefix: The prefix of the directory to look for (e.g., 'folder_B') :param timeout: Maximum time to wait in seconds (default: 60) :param interval: Seconds to wait between checks (default: 2) :return: Path object of the found directory, or None if timeout reached """ base_path = pathlib.Path(parent_folder) # We look for any directory starting with the prefix: 'prefix*' search_pattern = f"{prefix}*" start_time = time.time() end_time = start_time + timeout logging.debug(f"Watching '{parent_folder}' for a folder starting with '{prefix}'...") logging.debug(f"Timeout set to {timeout} seconds. Checking every {interval} seconds.") while time.time() < end_time: # base_path.glob('folder_B*') returns a generator of matching paths # We filter to make sure we only match actual directories matching_dirs = [path for path in base_path.glob(search_pattern) if path.is_dir()] if matching_dirs: # Found at least one match! Grab the first one found found_folder = matching_dirs[0] elapsed = round(time.time() - start_time, 1) logging.debug(f"\nSuccess! Found '{found_folder.name}' after {elapsed} seconds.") return found_folder # print(".", end="", flush=True) # Progress dot time.sleep(interval) logging.debug(f"\nTimeout reached. No folder starting with '{prefix}' appeared within {timeout} seconds.") return None def convert_image_to_base64(image_path: pathlib.Path) -> str | None: """ Reads an image file and converts it to a base64 encoded string. """ if not image_path.exists(): logging.debug(f"Error: Image not found at {image_path}") return None try: # 'rb' is critical here to read the file in binary mode with open(image_path, "rb") as image_file: binary_data = image_file.read() # Encode to base64 and decode to a UTF-8 string for easy usage/transmission base64_encoded = base64.b64encode(binary_data).decode("utf-8") return base64_encoded except Exception as e: logging.debug(f"Failed to convert image to base64: {e}") return None def build_output_json(folder_path: pathlib.Path, base64_image: str) -> str | None: """ Reads payload_raw.json.txt, extracts the timestamp, and builds the final JSON string. """ payload_file_path = folder_path / "payload_raw.json.txt" if not payload_file_path.exists(): logging.debug(f"Error: Payload file not found at {payload_file_path}") return None try: # 1. Read and parse the payload file with open(payload_file_path, "r", encoding="utf-8") as f: payload_data = json.load(f) # 2. Extract the timestamp (safely fallback to None if it doesn't exist) data_section = payload_data.get("data", {}) timestamp = data_section.get("timestamp") formatted_timestamp = format_timestamp(timestamp) if not timestamp: logging.debug("Warning: 'timestamp' key not found in payload file.") # 3. Build the final dictionary structure output_data = { "scene": base64_image, "payload": payload_data, "TS": formatted_timestamp } # 4. Serialize to a JSON string (indent=4 is optional, make it compact for APIs by removing it) return json.dumps(output_data, indent=4) except json.JSONDecodeError: logging.debug(f"Error: '{payload_file_path.name}' is not a valid JSON file.") return None except Exception as e: logging.debug(f"An unexpected error occurred while building JSON: {e}") return None def format_timestamp(ts_string: str) -> str: """ Converts '2026-07-15T07:53:34.608296Z' to '2026-07-15 07:53:34' """ if not ts_string: return "" # 1. Replace 'Z' with '+00:00' so Python's fromisoformat parser understands it clean_ts = ts_string.replace('Z', '+00:00') # 2. Parse the ISO string into a datetime object dt_obj = datetime.fromisoformat(clean_ts) # 3. Format it to the target string pattern return dt_obj.strftime('%Y-%m-%d %H:%M:%S') def main(argv, logger): if len(sys.argv) < 4: # Return a structured JSON error if arguments are missing logging.debug(json.dumps({"status": "error", "message": "Missing arguments"})) sys.exit(1) element_name = sys.argv[1] timestamp = sys.argv[2] spot_id = sys.argv[3] session = None try: session, login_response = login( base_url=config.BASE_URL, username=config.USERNAME, password=config.PASSWORD, user_nonce=config.USER_NONCE, user_key=config.USER_KEY, client_name=config.CLIENT_NAME, integration_identifier=config.INTEGRATION_IDENTIFIER, verify_tls=config.VERIFY_TLS, ) logging.debug("Login successful") logging.debug("Session: %s", session) full_camera_info=get_cameras(base_url=config.BASE_URL, session=session, verify_tls=config.VERIFY_TLS, verbosity="LOW") CAMERA_ID = full_camera_info["cameras"][0]["id"] # Format: DD/MM/YYYY dt_object = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S") foldername = spot_id + '_' + dt_object.strftime("%Y-%m-%d-%H-%M-%S") # Format: DD/MM/YYYY from standard database format capture_date = dt_object.strftime("%d/%m/%Y") capture_time = dt_object.strftime("%H:%M:%S") # in case we want it hardcoded # capture_date = "01/06/2026" # capture_time = "16:39:05" # This is the value of the snapshot of interest # Duration snapshot_interval = 300 buffer_for_vehicle_movement = 30 capture_duration = snapshot_interval + buffer_for_vehicle_movement # Optional chunk duration # Set to None to keep the old behaviour: one single movie file. movie_chunk_duration = 60 timing_info = estimate_wep_offset_from_health(config.BASE_URL, verify_tls=config.VERIFY_TLS) wep_t = compute_wep_extract_timestamp( capture_time, default_date=capture_date, site_timezone="utc", input_timezone="uk", wep_offset_seconds=timing_info["offset_seconds"] - capture_duration, ) logging.debug("WEP global start:" + str(wep_t)) # Output folder output_dir = Path("/var/www/html/videoplate/video") output_dir.mkdir(parents=True, exist_ok=True) try: wep_start_iso = wep_t.split(",")[0] wep_start_utc = datetime.fromisoformat( wep_start_iso.replace("Z", "+00:00") ) if wep_start_utc.tzinfo is None: wep_start_utc = wep_start_utc.replace(tzinfo=timezone.utc) wep_start_utc = wep_start_utc.astimezone(timezone.utc) wep_end_utc = wep_start_utc + timedelta(seconds=capture_duration) chunks = build_backward_chunks( global_start_utc=wep_start_utc, total_duration_sec=capture_duration, chunk_duration_sec=movie_chunk_duration, ) logging.debug(f"Total duration: {capture_duration}s") logging.debug(f"Number of chunks: " + str(len(chunks))) for chunk_index, (chunk_start_utc, chunk_end_utc, chunk_duration) in enumerate( chunks, start=1 ): chunk_wep_t = format_wep_timestamp( chunk_start_utc, original_wep_t=wep_t, ) logging.debug(f"Chunk " + str(chunk_index) + "/" + str(len(chunks))) logging.debug(" WEP start :" + str(chunk_wep_t)) logging.debug(" duration :" + str(chunk_duration)) stream_bytes, response = get_fmp4_media_range_from_timestamp( base_url=config.BASE_URL, session=session, camera_id=CAMERA_ID, timestamp=chunk_wep_t, duration=chunk_duration, verify_tls=config.VERIFY_TLS, ) logging.debug(" status_code :" + str(response.status_code)) logging.debug(f" content-type: {response.headers.get('Content-Type')}") chunk_start_uk = chunk_start_utc.astimezone(ZoneInfo("Europe/London")) chunk_end_uk = chunk_end_utc.astimezone(ZoneInfo("Europe/London")) filename = ( f"{chunk_start_uk.strftime('%Y%m%d_%H%M%S')}" f"_to_" f"{chunk_end_uk.strftime('%Y%m%d_%H%M%S')}" f"_UK" f"_chunk_{chunk_index:02d}_of_{len(chunks):02d}" f".mp4" ) output_path = output_dir / filename with open(output_path, "wb") as f: f.write(stream_bytes) logging.debug(f" saved: " + str(output_path)) new_output_path = f"{output_path}_masked_fast.mp4" newfilename = f"{filename}_masked_fast.mp4" # Example polygon = area to KEEP if spot_id == "88105": keep_polygon = np.array([[513, 1349], [785, 1034], [1625, 1153], [1546, 1529], [998, 1532]], dtype=np.int32) elif spot_id == "88106": keep_polygon = np.array([[1036, 712], [1671, 696], [1681, 940], [894, 902]], dtype=np.int32) speed_factor = 5.0 # 5 min -> 1 min cap = cv2.VideoCapture(str(output_path)) src_fps = cap.get(cv2.CAP_PROP_FPS) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) keep_polygon_larger = expand_polygon_from_center( polygon=keep_polygon, expand_percent=70, image_width=width, image_height=height, ) target_fps = src_fps * speed_factor fourcc = cv2.VideoWriter_fourcc(*"mp4v") out = cv2.VideoWriter(new_output_path, fourcc, target_fps, (width, height)) # Build mask once mask = np.zeros((height, width), dtype=np.uint8) cv2.fillPoly(mask, [keep_polygon_larger], 255) mask_inv = cv2.bitwise_not(mask) while True: ret, frame = cap.read() if not ret: break # Keep only polygon area, black elsewhere #result = cv2.bitwise_and(frame, frame, mask=mask) blurred_frame = cv2.GaussianBlur(frame, (25, 25), 0) sharp_bg = cv2.bitwise_and(frame, frame, mask=mask) blurred_fg = cv2.bitwise_and(blurred_frame, blurred_frame, mask=mask_inv) result = cv2.add(sharp_bg, blurred_fg) out.write(result) cap.release() out.release() VIDEO_FILENAME = newfilename VIDEO_URL = build_video_url( filename=VIDEO_FILENAME, public_base_url=config.PUBLIC_VIDEO_BASE_URL, ) logging.debug("VIDEO_URL =" + str(VIDEO_URL)) check_video_url(VIDEO_URL) submit_response = submit_video_to_stream_cloud( api_token=config.API_TOKEN, upload_url=VIDEO_URL, camera_id=foldername, regions=config.REGIONS, mmc=config.MMC, ) # To be decided if poll and wait or send all and wait # print(json.dumps(submit_response, indent=2)) # End of for loop # sleep few seconds to wait for plate recognizer response # Replace these with your actual folder paths today_str = datetime.today().strftime('%Y-%m-%d') WATCH_DIR = config.WEBHOOK_BASEFOLDER + today_str PREFIX = foldername # This will block the script for up to 60 seconds looking for folderA/folder_B found_path = wait_for_folder_pattern(WATCH_DIR, PREFIX, timeout=60, interval=3) if found_path: # 3. Construct the path to vehicle.jpg inside the found folder image_path = found_path / "vehicle.jpg" logging.debug(f"Looking for image at: {image_path}") # 4. Convert the image base64_string = convert_image_to_base64(image_path) if base64_string: logging.debug("\nSuccessfully converted 'vehicle.jpg' to base64!") # Print just the first 50 characters so it doesn't flood your terminal logging.debug(f"Base64 String (preview): {base64_string[:50]}...") # Build the final JSON output final_json = build_output_json(found_path, base64_string) print(final_json) except Exception as e: logging.error(f"ERROR for capture {capture_date} {capture_time}: {e}") except Exception as e: logging.error("Login failed: %s", e) if __name__ == "__main__": file = config.PATH_TO_LOGS + config.LOG_NAME + '-' + date.today().strftime("%Y-%m-%d") + '.log' logging.basicConfig(filename=file, level=config.LOG_LEVEL,format='%(asctime)s %(levelname)-8s %(name)-12s %(message)s') logging.debug("INIT TESCO NEW APPROACH SCRIPT") # os.chmod(file, 0o777) logger = logging.getLogger() main(sys.argv[1:], logger)