| | # photofresh.py - adaptive, per-photo touch-up for phone photos (no AI, nothing invented) | | | # | | | # Watches an inbox folder, gently improves each new JPEG and writes the result to an | | | # output folder (e.g. an Immich external library), sorted into year/month subfolders. | | | # | | | # Folders (inside the container): | | | # /data/inbox <- new photos arrive here (e.g. via Syncthing) | | | # /data/processed <- originals after processing (auto-deleted after KEEP_DAYS) | | | # /data/failed <- photos that could not be processed (never deleted) | | | # /data/redo <- put an original here to process it again and overwrite the old result | | | # /output <- processed photos, in year/month subfolders | | | # | | | # Per photo it MEASURES first and only then adjusts: | | | # 1. white balance: corrects half of the measured colour cast (max 12%) | | | # 2. contrast range: stretched only if the histogram is compressed (haze, flat light) | | | # 3. midtone lift: only for photos that are too dark | | | # 4. local contrast (CLAHE), mixed in at 40% | | | # 5. noise reduction: only if measured noise is high, and only on flat areas (walls, sky) | | | # "Night mode": high ISO + dark photo -> keep the darkness, only clean up noise. | | | # All EXIF data (date, GPS, orientation) is copied unchanged. | | | import os, time, shutil, subprocess, traceback | | | import cv2, numpy as np | | | | | | DATA = "/data" | | | INBOX, PROCESSED, FAILED, REDO = (f"{DATA}/{d}" for d in ("inbox", "processed", "failed", "redo")) | | | OUTPUT = "/output" |
| | LEDGER, LOG = f"{DATA}/processed.list", f"{DATA}/photofresh.log" |
| | KEEP_DAYS = float(os.environ.get("KEEP_DAYS", "30")) # 0 = delete originals right after processing |
| | SHARPEN = float(os.environ.get("SHARPEN", "0")) # 0 = off (default), 1 = light sharpening |
| | NIGHT_ISO = float(os.environ.get("NIGHT_ISO", "1000")) # ISO from which a dark photo counts as a night shot (0 = off) |
| | PUID, PGID = int(os.environ.get("PUID", "99")), int(os.environ.get("PGID", "100")) # 99/100 = Unraid default |
| | JPEG_QUALITY = int(os.environ.get("JPEG_QUALITY", "95")) |
| | INTERVAL = 120 # seconds between scans | | | MIN_AGE = 30 # a file must be unchanged for this long (still syncing otherwise) | | | EXT = (".jpg", ".jpeg") | | | |
| | def log(msg): |
| | line = time.strftime("%Y-%m-%d %H:%M:%S ") + msg |
| | print(line, flush=True) |
| | with open(LOG, "a") as f: f.write(line + "\n") |
| | | | | def set_owner(path, is_dir=False): | | | try: | | | os.chown(path, PUID, PGID) | | | os.chmod(path, 0o777 if is_dir else 0o666) | | | except Exception: pass | | | |
| | def make_dir(path): |
| | if not os.path.isdir(path): |
| | os.makedirs(path, exist_ok=True) |
| | set_owner(path, True) |
| | |
| | # ---------- the algorithm ---------- |
| | def noise_sigma(gray): |
| | """Estimate noise only in the 15% flattest blocks, so texture (grass, hair) doesn't count as noise.""" |
| | g = gray.astype(np.float32) |
| | hp = g - cv2.GaussianBlur(g, (0, 0), 1.5) |
| | grad = cv2.GaussianBlur(np.abs(cv2.Sobel(cv2.GaussianBlur(g, (0, 0), 3), cv2.CV_32F, 1, 1)), (0, 0), 3) |
| | B = 32; vals = []; flat = [] |
| | for y in range(0, g.shape[0] - B, B): |
| | for x in range(0, g.shape[1] - B, B): |
| | m = g[y:y+B, x:x+B].mean() |
| | if 25 < m < 235: |
| | flat.append(grad[y:y+B, x:x+B].mean()) |
| | vals.append(np.median(np.abs(hp[y:y+B, x:x+B])) * 1.4826) |
| | if not vals: return 0.0 |
| | flat = np.array(flat); vals = np.array(vals) |
| | return float(np.median(vals[flat <= np.percentile(flat, 15)])) |
| | |
| | def enhance(bgr, iso=None): |
| | info = {} |
| | img = bgr.astype(np.float32) / 255.0 |
| | # night shot? (high ISO and dark image): keep the darkness, it is the mood |
| | mean = float(np.mean(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY))) / 255 |
| | night = bool(NIGHT_ISO > 0 and iso is not None and iso >= NIGHT_ISO and mean < 0.40) |
| | info["night"] = night |
| | # 1. white balance: measure the cast on near-grey pixels, correct 50% (25% at night), max 12% |
| | hsv = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV) |
| | v, s = hsv[..., 2], hsv[..., 1] |
| | mask = (s < 60) & (v > 40) & (v < 240) |
| | if mask.sum() < 0.02 * mask.size: mask = (v > 20) & (v < 245) |
| | means = np.array([img[..., c][mask].mean() for c in range(3)]) |
| | strength = 0.25 if night else 0.5 |
| | gains = np.clip(1 + strength * (means.mean() / means - 1), 0.88, 1.12) |
| | img = np.clip(img * gains, 0, 1) |
| | info["wb"] = [round(float(x), 3) for x in gains] |
| | # 2. stretch the contrast range (may clip a tiny bit) |
| | lab = cv2.cvtColor((img * 255).astype(np.uint8), cv2.COLOR_BGR2LAB) |
| | L = lab[..., 0].astype(np.float32) |
| | lo, hi = np.percentile(L, 0.5), np.percentile(L, 99.5) |
| | lo_new = lo - 0.7 * max(lo - 6, 0) |
| | hi_new = hi + 0.8 * max(252 - hi, 0) |
| | if hi - lo > 10 and not night: |
| | L = (L - lo) / (hi - lo) * (hi_new - lo_new) + lo_new |
| | # 3. photo too dark: gentle midtone lift | | | meanL = float(np.mean(np.clip(L, 0, 255))) / 255 | | | if meanL < 0.42 and not night: |
| | gamma = max(0.85, np.log(0.45) / np.log(max(meanL, 1e-3))) |
| | L = 255 * np.power(np.clip(L / 255, 0, 1), gamma) |
| | # 4. adaptive local contrast (CLAHE), mixed in at 40%; skipped at night (it lifts shadows and grain) | | | L8 = np.clip(L, 0, 255).astype(np.uint8) | | | if not night: |
| | L8 = cv2.addWeighted(L8, 0.6, cv2.createCLAHE(clipLimit=1.6, tileGridSize=(8, 8)).apply(L8), 0.4, 0) |
| | lab[..., 0] = L8 |
| | out = cv2.cvtColor(lab, cv2.COLOR_LAB2BGR) |
| | # 5. noise reduction only when needed, and only on flat areas |
| | sig = noise_sigma(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)) |
| | info["noise"] = round(sig, 2) |
| | s_ref = max(sig, 0.8) |
| | g = cv2.cvtColor(out, cv2.COLOR_BGR2GRAY).astype(np.float32) |
| | tex = cv2.GaussianBlur(np.abs(g - cv2.GaussianBlur(g, (0, 0), 2)), (0, 0), 4) |
| | texture = np.clip((tex - s_ref) / (3.0 * s_ref), 0, 1) # 0 = flat, 1 = texture/edge |
| | if sig > 1.2: |
| | # luminance noise (grain): gentle; colour noise (speckles): firm |
| | h_lum = float(np.clip((sig - 1.2) * 0.8 + 2.5, 2.5, 5.0)) |
| | h_col = float(np.clip((sig - 1.2) * 2.0 + 6.0, 6.0, 14.0)) |
| | den = cv2.fastNlMeansDenoisingColored(out, None, h_lum, h_col, 5, 15) |
| | w = (1 - texture)[..., None] |
| | out = (den * w + out * (1 - w)).astype(np.uint8) |
| | # 6. optional light sharpening (OFF by default): luminance only, textured areas only, | | | # ignores differences below the noise level, clamped to avoid halos |
| | if SHARPEN > 0: |
| | lab = cv2.cvtColor(out, cv2.COLOR_BGR2LAB) |
| | Lf = lab[..., 0].astype(np.float32) |
| | detail = Lf - cv2.GaussianBlur(Lf, (0, 0), 1.0) |
| | thr = max(1.5, 1.2 * sig) |
| | detail = np.sign(detail) * np.maximum(np.abs(detail) - thr, 0) |
| | detail = np.clip(detail, -12, 12) |
| | amount = (0.7 if sig < 3 else 0.5) * SHARPEN |
| | lab[..., 0] = np.clip(Lf + amount * detail * texture, 0, 255).astype(np.uint8) |
| | out = cv2.cvtColor(lab, cv2.COLOR_LAB2BGR) |
| | return out, info | | | |
| | # ---------- file handling ---------- |
| | def load_ledger(): |
| | if not os.path.exists(LEDGER): return set() |
| | with open(LEDGER) as f: return set(l.strip() for l in f if l.strip()) |
| | | | | def read_iso(src): | | | try: |
| | r = subprocess.run(["exiftool", "-s3", "-n", "-ISO", src], capture_output=True, text=True, timeout=30) |
| | return float(r.stdout.strip().split()[0]) |
| | except Exception: | | | return None | | | | | | def target_subdir(src): | | | try: |
| | r = subprocess.run(["exiftool", "-s3", "-d", "%Y/%m", "-DateTimeOriginal", src], |
| | capture_output=True, text=True, timeout=30) |
| | sub = r.stdout.strip() |
| | if len(sub) == 7 and sub[4] == "/": return sub |
| | except Exception: pass | | | return time.strftime("%Y/%m", time.localtime(os.path.getmtime(src))) | | | |
| | def unique(path): |
| | if not os.path.exists(path): return path |
| | base, ext = os.path.splitext(path); i = 1 |
| | while os.path.exists(f"{base}_{i}{ext}"): i += 1 |
| | return f"{base}_{i}{ext}" |
| | |
| | def process(name, ledger, folder=INBOX, redo=False): |
| | src = os.path.join(folder, name) |
| | key = f"{name}\|{os.path.getsize(src)}" |
| | if key in ledger and not redo: | | | # already done (e.g. re-synced): don't duplicate it in the library, but don't delete it either |
| | shutil.move(src, unique(os.path.join(PROCESSED, name))) |
| | log(f"skipped (already processed), moved to 'processed': {name} -> put it in 'redo' to process it again"); return |
| | # IGNORE_ORIENTATION: don't rotate pixels, so the original Orientation tag stays correct | | | bgr = cv2.imread(src, cv2.IMREAD_COLOR | cv2.IMREAD_IGNORE_ORIENTATION) | | | if bgr is None: raise RuntimeError("could not read photo") |
| | iso = read_iso(src) |
| | out, info = enhance(bgr, iso) |
| | sub = target_subdir(src) |
| | make_dir(os.path.join(OUTPUT, sub[:4])); make_dir(os.path.join(OUTPUT, sub)) |
| | dst = os.path.join(OUTPUT, sub, name) |
| | if not redo: dst = unique(dst) # in 'redo' the old result is overwritten | | | tmp = dst + ".tmp.jpg" |
| | if not cv2.imwrite(tmp, out, [cv2.IMWRITE_JPEG_QUALITY, JPEG_QUALITY]): raise RuntimeError("write failed") |
| | r = subprocess.run(["exiftool", "-q", "-overwrite_original", "-TagsFromFile", src, "-all:all", tmp], |
| | capture_output=True, text=True, timeout=60) |
| | if r.returncode != 0: raise RuntimeError("copying metadata failed: " + r.stderr.strip()) |
| | if cv2.imread(tmp) is None: raise RuntimeError("result check failed") |
| | if not redo: # in 'redo' the file gets the current time so the library notices the change |
| | st = os.stat(src); os.utime(tmp, (st.st_atime, st.st_mtime)) |
| | os.rename(tmp, dst); set_owner(dst) |
| | if key not in ledger: |
| | with open(LEDGER, "a") as f: f.write(key + "\n") |
| | ledger.add(key) |
| | if KEEP_DAYS > 0: |
| | keep = unique(os.path.join(PROCESSED, name)) |
| | shutil.move(src, keep) |
| | os.utime(keep, None) # retention clock starts now, not at the capture date | | | else: |
| | os.remove(src) |
| | log(f"{'redo ' if redo else ''}done: {name} -> {sub}/ iso={iso} night={'yes' if info['night'] else 'no'} wb={info['wb']} noise={info['noise']}") |
| | |
| | def cleanup(): |
| | limit = time.time() - KEEP_DAYS * 86400 |
| | for n in os.listdir(PROCESSED): |
| | p = os.path.join(PROCESSED, n) |
| | if os.path.isfile(p) and os.path.getmtime(p) < limit: |
| | os.remove(p); log(f"original deleted after {KEEP_DAYS:g} days: {n}") |
| | | | | REPORTED = set() # so each skipped file is logged only once | | | |
| | def report_once(name, reason): |
| | if (name, reason) not in REPORTED: |
| | REPORTED.add((name, reason)); log(f"skipped: {name} ({reason})") |
| | |
| | def scan(ledger): |
| | for name in sorted(os.listdir(INBOX)): |
| | p = os.path.join(INBOX, name) |
| | if name.startswith(".") or not os.path.isfile(p): continue |
| | if not name.lower().endswith(EXT): |
| | report_once(name, "not a .jpg/.jpeg file"); continue |
| | age = time.time() - os.path.getmtime(p) |
| | if age < -60: |
| | # file date in the future: would otherwise never be processed | | | report_once(name, "file date is in the future, processing anyway") | | | elif age < MIN_AGE: | | | continue | | | try: | | | process(name, ledger) | | | except Exception as e: |
| | log(f"ERROR with {name}: {e}") |
| | try: shutil.move(p, unique(os.path.join(FAILED, name))) |
| | except Exception: log(traceback.format_exc()) |
| | for name in sorted(os.listdir(REDO)): |
| | p = os.path.join(REDO, name) |
| | if name.startswith(".") or not os.path.isfile(p): continue |
| | if not name.lower().endswith(EXT): |
| | report_once(name, "not a .jpg/.jpeg file (redo folder)"); continue |
| | if 0 <= time.time() - os.path.getmtime(p) < 10: continue # still being copied |
| | try: | | | process(name, ledger, REDO, redo=True) | | | except Exception as e: |
| | log(f"ERROR with {name} (redo): {e}") |
| | try: shutil.move(p, unique(os.path.join(FAILED, name))) |
| | except Exception: log(traceback.format_exc()) |
| | |
| | if __name__ == "__main__": |
| | for d in (INBOX, PROCESSED, FAILED, REDO): make_dir(d) |
| | log(f"photofresh started (keeping originals: {KEEP_DAYS:g} days, scanning every {INTERVAL} s)") |
| | log(f"inbox currently contains: {sorted(os.listdir(INBOX)) or 'nothing'}") |
| | ledger = load_ledger() |
| | while True: | | | try: | | | scan(ledger); cleanup() | | | except Exception: |
| | log(traceback.format_exc()) |
| | time.sleep(INTERVAL) |