Automating PhotoMesh Projects with Python.
A Step-by-Step Guide.
PhotoMesh, a photogrammetry software from Skyline Software Systems, includes a local REST API that makes it possible to automate key parts of the photogrammetry workflow. For example, creating projects, submitting them to the PhotoMesh Project Queue, and triggering various processing steps.
This is an example of a simple, yet effective automation solution. For example: The operator copies a folder of aerial images into a watched directory, and the system will automatically detect the input data, build the project configuration files, and submit the project for processing using the PhotoMesh API.
This article will break down a Python script which performs the following:
The full script is explained line by line so you can modify it or extend it for your own use cases. While the workflow shown here is simple, it demonstrates the automatic processing of aerial photogrammetry data, using the PhotoMesh API and Python scripting.
Script Breakdown.
🔧 1. Imports and Setup.
import os
import time
import json
import requests
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from sseclient import SSEClient
2. Configuration.
# === Configuration ===
WATCH_FOLDER = r"D:\00Automation\Data"
PROJECT_ROOT = r"D:\00Automation\Projects"
JSON_FOLDER = r"D:\00Automation\JSON"
WORKING_FOLDER = r"C:\WorkingFolder"
QUEUE_API_URL = "http://localhost:8087/ProjectQueue/"
QUEUE_SSE_URL = "http://localhost:8087/ProjectQueue/events"
FOLDER_STABILITY_WAIT = 10 # Seconds between stability checks
The configuration section defines folder paths and timing:
3. Ensuring Folders Exist.
# === Ensure required folders exist ===
os.makedirs(PROJECT_ROOT, exist_ok=True)
os.makedirs(JSON_FOLDER, exist_ok=True)
os.makedirs(WORKING_FOLDER, exist_ok=True)
Makes sure directories exist before processing begins.
4. Detecting Folder Stability.
# === Folder stability check: filenames + sizes ===
def folder_is_stable(folder_path, wait=FOLDER_STABILITY_WAIT):
This helper function waits and checks whether file names and sizes inside the folder have stopped changing. This is useful for detecting when a large copy operation has finished.
5. Watching for New Folders.
# === File system event handler ===
class ImageFolderHandler(FileSystemEventHandler):
def on_created(self, event):
This class is called by watchdog whenever a new folder appears. Once a folder is detected, it begins the stabilisation process.
6. Waiting for Folder Copy to Finish.
def wait_and_process(self, folder_path):
...
Waits repeatedly until folder_is_stable() returns True. This avoids starting to build the project before the data transfer is complete.
Recommended by LinkedIn
7. Processing a Ready Folder.
def process(self, folder_path):
...
Once the folder is stable and data is no longer being transferred:
8. Building and Submitting JSON.
# === Build JSON payload ===
data = [
{
"comment": f"Auto project: {folder_name}",
"action": 0,
"projectPath": project_path,
"buildFrom": 1,
"buildUntil": 6,
"inheritBuild": "",
"preset": "CROptimised",
"workingFolder": WORKING_FOLDER,
"MaxLocalFusers": 10,
"MaxAWSFusers": 0,
"AWSFuserStartupScript": "script",
"AWSBuildConfigurationName": "",
"AWSBuildConfigurationJsonPath": "",
"sourceType": 0,
"sourcePath": source_path
}
]
This builds a properly structured JSON file for the PhotoMesh /project/add endpoint.
<?xml version="1.0" encoding="utf-8"?>
<BuildParametersPreset xmlns:i="http://www.w3.org/2001/XMLSchema-instance">
<SerializableVersion>8.0.4.50513</SerializableVersion>
<Version xmlns:d2p1="http://schemas.datacontract.org/2004/07/System">
<d2p1:_Build>4</d2p1:_Build>
<d2p1:_Major>8</d2p1:_Major>
<d2p1:_Minor>0</d2p1:_Minor>
<d2p1:_Revision>50513</d2p1:_Revision>
</Version>
<BuildParameters>
<SerializableVersion>8.0.4.50513</SerializableVersion>
<Version xmlns:d3p1="http://schemas.datacontract.org/2004/07/System">
<d3p1:_Build>4</d3p1:_Build>
<d3p1:_Major>8</d3p1:_Major>
<d3p1:_Minor>0</d3p1:_Minor>
<d3p1:_Revision>50513</d3p1:_Revision>
</Version>
<AddWalls>false</AddWalls>
<BuildATFlags>-bn_gm 1 @alignmultitiles[AlwaysCreateTilesBelowSparse=1] -m_tf 100 @Match2[num_cameras_per_group=10] @Match2[min_connected_to_camera=30] @Match2[num_features_per_collection=300] @Match2[total_num_features=900] @sfmb2[est_type=2] @sfmb2[bundle_adjustment_max_num_iterations_last_ba=500] @dpc[AddSeeds1=0]</BuildATFlags>
<BuildFlags></BuildFlags>
<ColorTone>1.05</ColorTone>
<DsmSettings />
<EdgeEnhance>false</EdgeEnhance>
<FillInGround>false</FillInGround>
<FocalLengthAccuracy>-1</FocalLengthAccuracy>
<HorizontalAccuracyFactor>0.1</HorizontalAccuracyFactor>
<IgnoreOrientation>false</IgnoreOrientation>
<MeshResolution>7.5</MeshResolution>
<OrthoSettings />
<OutputCoordinateSystem xmlns:d3p1="http://www.skylineglobe.com/schema-3dml">
<d3p1:OriginalWKT>COMPD_CS["UTM zone 55, Southern Hemisphere + 5773",PROJCS["UTM zone 55, Southern Hemisphere",GEOGCS["WGS 84",DATUM["WGS_1984",SPHEROID["WGS 84",6378137,298.257223563,AUTHORITY["EPSG","7030"]],AUTHORITY["EPSG","6326"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4326"]],PROJECTION["Transverse_Mercator"],PARAMETER["latitude_of_origin",0],PARAMETER["central_meridian",147],PARAMETER["scale_factor",0.9996],PARAMETER["false_easting",500000],PARAMETER["false_northing",10000000],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["Easting",EAST],AXIS["Northing",NORTH],AUTHORITY["EPSG","32755"]],VERT_CS["EGM96 geoid height",VERT_DATUM["EGM96 geoid",2005,AUTHORITY["EPSG","5171"],EXTENSION["PROJ4_GRIDS","egm96_15.gtx"]],UNIT["m",1.0],AXIS["Up",UP],AUTHORITY["EPSG","5773"]]]</d3p1:OriginalWKT>
<d3p1:WKT>COMPD_CS["UTM zone 55, Southern Hemisphere + 5773",PROJCS["UTM zone 55, Southern Hemisphere",GEOGCS["WGS 84",DATUM["WGS_1984",SPHEROID["WGS 84",6378137,298.257223563,AUTHORITY["EPSG","7030"]],AUTHORITY["EPSG","6326"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4326"]],PROJECTION["Transverse_Mercator"],PARAMETER["latitude_of_origin",0],PARAMETER["central_meridian",147],PARAMETER["scale_factor",0.9996],PARAMETER["false_easting",500000],PARAMETER["false_northing",10000000],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["Easting",EAST],AXIS["Northing",NORTH],AUTHORITY["EPSG","32755"]],VERT_CS["EGM96 geoid height",VERT_DATUM["EGM96 geoid",2005,AUTHORITY["EPSG","5171"],EXTENSION["PROJ4_GRIDS","egm96_15.gtx"]],UNIT["m",1.0],AXIS["Up",UP],AUTHORITY["EPSG","5773"]]]</d3p1:WKT>
</OutputCoordinateSystem>
<PointCloudFormat>LAS</PointCloudFormat>
<PointCloudQuality>4</PointCloudQuality>
<PrincipalPointAccuracy>-1</PrincipalPointAccuracy>
<RadialAccuracy>false</RadialAccuracy>
<SmoothSurface>false</SmoothSurface>
<TangentialAccuracy>false</TangentialAccuracy>
<TileSplitMethod>Dynamic</TileSplitMethod>
<VerticalAccuracyFactor>0.1</VerticalAccuracyFactor>
<VerticalBias>false</VerticalBias>
</BuildParameters>
<Description>CROptimised</Description>
<IsDefault>false</IsDefault>
<IsLastUsed>false</IsLastUsed>
<IsSystem>false</IsSystem>
<IsSystemDefault>false</IsSystemDefault>
<PresetFileName i:nil="true" />
<PresetName>CROptimised</PresetName>
</BuildParametersPreset>
The CROptimised preset file provides all the build parameters that is normally set by the operator manually. Different PhotoMesh preset files can be created when needing to change the specific build parameters. For example:
with open(json_path, 'w') as f:
json.dump(data, f, indent=2)
Once the JSON file is created, it is written to the disk.
response = requests.post(f"{QUEUE_API_URL}project/add", json=data)
Then the JSON file is submitted to the local PhotoMesh Project Queue.
9. Starting the Build.
response = requests.get(f"{QUEUE_API_URL}Build/Start")
This API call tells PhotoMesh to begin processing the current queued project.
The JSON file that is submitted to the Project Queue, tells PhotoMesh everything it needs to know to process the specific dataset.
10. Real-Time Build Monitoring.
for event in SSEClient(QUEUE_SSE_URL):
...
Listens for build status updates via Server-Sent Events, and prints it to the command prompt. It logs when the project finishes and when the entire queue is complete.
11. Waiting for New Projects to Arrive.
if __name__ == "__main__":
...
Initializes the Observer from watchdog, starts watching the folder, and loops forever—waiting for new projects to arrive.
12. Putting it all Together.
import os
import time
import json
import requests # type: ignore
from watchdog.observers import Observer # type: ignore
from watchdog.events import FileSystemEventHandler # type: ignore
from sseclient import SSEClient # type: ignore
print("⚠️ IMPORTANT: Make sure PhotoMesh is running as Administrator before continuing.")
input("Press Enter to continue...\n")
# === Helpers for default folders ===
def get_default_documents_folder(subdir):
return os.path.join(os.path.expanduser("~"), "Documents", subdir)
# === Folder input with optional default and auto-create ===
def get_folder_input(prompt_text, default_path=None, create_if_missing=False):
while True:
path = input(prompt_text).strip('"')
if not path and default_path:
path = default_path
print(f"No input given. Using default: {path}")
if os.path.isdir(path):
return path
elif create_if_missing:
try:
os.makedirs(path, exist_ok=True)
return path
except Exception as e:
print(f"Failed to create directory: {e}")
else:
print("Folder does not exist. Try again.")
# === Prompt user for folder paths ===
WATCH_FOLDER = get_folder_input(
"Enter folder to WATCH for new projects (leave blank for default): ",
default_path=get_default_documents_folder("PhotoMeshWatch"),
create_if_missing=True
)
PROJECT_ROOT = get_folder_input(
"Enter PROJECT ROOT directory (where to create projects, leave blank for default): ",
default_path=get_default_documents_folder("PhotoMeshProjects"),
create_if_missing=True
)
# === Prompt user for PhotoMesh preset ===
preset_choice = input("Enter PhotoMesh preset name (default is 'PhotoMesh Default'): ").strip()
if not preset_choice:
preset_choice = "PhotoMesh Default"
# === Static Configuration ===
JSON_FOLDER = r"D:\\00Automation\\JSON"
WORKING_FOLDER = r"C:\\WorkingFolder"
QUEUE_API_URL = "http://localhost:8087/ProjectQueue/"
QUEUE_SSE_URL = "http://localhost:8087/ProjectQueue/events"
FOLDER_STABILITY_WAIT = 10 # Seconds between stability checks
# === Ensure required folders exist ===
os.makedirs(JSON_FOLDER, exist_ok=True)
os.makedirs(WORKING_FOLDER, exist_ok=True)
# === Folder stability check ===
def folder_is_stable(folder_path, wait=FOLDER_STABILITY_WAIT):
def get_folder_signature(path):
signature = set()
for root, _, files in os.walk(path):
for f in files:
fp = os.path.join(root, f)
try:
size = os.path.getsize(fp)
signature.add((fp, size))
except:
continue
return signature
try:
snapshot = get_folder_signature(folder_path)
time.sleep(wait)
return snapshot == get_folder_signature(folder_path)
except Exception:
return False
# === File system event handler ===
class ImageFolderHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
print(f"\nNew folder detected: {event.src_path}")
self.wait_and_process(event.src_path)
def wait_and_process(self, folder_path):
print("Waiting for folder to stabilize...")
attempt = 1
while True:
if folder_is_stable(folder_path):
print("Folder is stable.")
self.process(folder_path)
return
print(f"Still copying... attempt {attempt}")
time.sleep(FOLDER_STABILITY_WAIT)
attempt += 1
def process(self, folder_path):
folder_name = os.path.basename(folder_path)
print(f"Preparing project: {folder_name}")
# Create project-specific folder
project_dir = os.path.join(PROJECT_ROOT, folder_name)
os.makedirs(project_dir, exist_ok=True)
project_path = os.path.join(project_dir, f"{folder_name}.PhotoMeshXML")
json_path = os.path.join(JSON_FOLDER, f"{folder_name}.json")
# === Detect subfolders (collections) ===
subfolders = [
name for name in os.listdir(folder_path)
if os.path.isdir(os.path.join(folder_path, name))
]
if not subfolders:
print("No sub-collections found. Using folder itself.")
source_path = [
{
"name": "RGB",
"path": folder_path,
"properties": ""
}
]
else:
source_path = []
for subfolder in subfolders:
subfolder_path = os.path.join(folder_path, subfolder)
has_images = any(
f.lower().endswith(('.jpg', '.jpeg'))
for f in os.listdir(subfolder_path)
if os.path.isfile(os.path.join(subfolder_path, f))
)
if has_images:
source_path.append({
"name": subfolder,
"path": subfolder_path,
"properties": ""
})
if not source_path:
print("No valid image collections found. Skipping.")
return
# === Build JSON payload ===
data = [
{
"comment": f"Auto project: {folder_name}",
"action": 0,
"projectPath": project_path,
"buildFrom": 1,
"buildUntil": 6,
"inheritBuild": "",
"preset": preset_choice,
"workingFolder": WORKING_FOLDER,
"MaxLocalFusers": 10,
"MaxAWSFusers": 0,
"AWSFuserStartupScript": "script",
"AWSBuildConfigurationName": "",
"AWSBuildConfigurationJsonPath": "",
"sourceType": 0,
"sourcePath": source_path
}
]
# Write JSON to file
with open(json_path, 'w') as f:
json.dump(data, f, indent=2)
print(f"JSON created: {json_path}")
# Submit to PhotoMesh queue
response = requests.post(f"{QUEUE_API_URL}project/add", json=data)
if response.status_code == 200:
print(f"Submitted project: {folder_name}")
self.start_build(folder_name)
else:
print(f"Submission failed with status {response.status_code}")
print(f"Response: {response.text}")
def start_build(self, folder_name):
response = requests.get(f"{QUEUE_API_URL}Build/Start")
if response.status_code == 200:
print(f"Build started: {folder_name}")
self.watch_queue()
else:
print("Failed to start build.")
def watch_queue(self):
try:
for event in SSEClient(QUEUE_SSE_URL):
if event.event == "Finished":
data = json.loads(event.data)
print(f"Build complete: {data.get('comment', '')}")
elif event.event == "QueueFinished":
print("All projects in queue complete.")
break
except Exception as e:
print(f"Error watching queue: {e}")
# === Start watching ===
if __name__ == "__main__":
print(f"\nWatching folder: {WATCH_FOLDER}")
observer = Observer()
handler = ImageFolderHandler()
observer.schedule(handler, path=WATCH_FOLDER, recursive=False)
observer.start()
try:
while True:
time.sleep(5)
except KeyboardInterrupt:
observer.stop()
observer.join()
13. Summary & Conclusion.
With this basic example, you can already:
Because PhotoMesh exposes an accessible API, extending this example into more advanced workflows is entirely possible.
