API Reference

Top level

Assess one image and return {component_name: (raw, scalar)}.

Parameters:

Name	Type	Description	Default
image	'str | object'	path to an image file (str or PathLike), or a BGR uint8 numpy array of shape (H, W, 3).	required

Returns:

Type	Description
dict	Mapping of OFIQ component name to (raw native value, 0-100 scalar). Empty
dict	if no face is detected.

Source code in ofiqpy/__init__.py

def assess(image: "str | object") -> dict:
    """Assess one image and return ``{component_name: (raw, scalar)}``.

    Args:
        image: path to an image file (str or PathLike), or a BGR uint8 numpy
            array of shape (H, W, 3).

    Returns:
        Mapping of OFIQ component name to ``(raw native value, 0-100 scalar)``. Empty
        if no face is detected.
    """
    import numpy as np

    pipe, meas = _lazy()
    if isinstance(image, np.ndarray):
        bgr = image
    else:
        import cv2

        bgr = cv2.imread(str(image))
        if bgr is None:
            raise FileNotFoundError(f"could not read image: {image}")
    session = pipe.process(bgr)
    if session.bbox is None:
        return {}
    return meas.compute(session)

Pipeline

Source code in ofiqpy/pipeline.py

class OFIQPipeline:
    def __init__(self, cfg: OFIQConfig | None = None, enable_pose=True, enable_parsing=True, enable_occlusion=True):
        self.cfg = cfg or OFIQConfig()
        d = self.cfg.detector()
        self.detector = SSDDetector(
            self.cfg.resolve(d["model_path"]),
            self.cfg.resolve(d["prototxt_path"]),
            d["confidence_thr"],
            d["min_rel_face_size"],
            d["padding"],
        )
        self.landmarker = ADNetLandmarker(self.cfg.landmarks_model())
        self._pose = self._parser = self._occ = None
        self.enable_pose = enable_pose
        self.enable_parsing = enable_parsing
        self.enable_occlusion = enable_occlusion

    def _get_pose(self):
        if self._pose is None:
            from .pose.tddfa import TDDFAPose

            self._pose = TDDFAPose(self.cfg.resolve(self.cfg.measure("HeadPose")["model_path"]))
        return self._pose

    def _get_parser(self):
        if self._parser is None:
            from .segmentation.parsing import FaceParser

            self._parser = FaceParser(self.cfg.resolve(self.cfg.measure("FaceParsing")["model_path"]))
        return self._parser

    def _get_occ(self):
        if self._occ is None:
            from .segmentation.occlusion import OcclusionSeg

            self._occ = OcclusionSeg(self.cfg.resolve(self.cfg.measure("FaceOcclusionSegmentation")["model_path"]))
        return self._occ

    def process(self, image: np.ndarray) -> Session:
        s = Session(image=image)
        faces = self.detector.detect(image)
        s.n_faces = len(faces)
        s.face_areas = [f[2] * f[3] for f in faces]
        if not faces:
            return s
        primary = faces[0]
        s.bbox = np.array([primary[0], primary[1], primary[0] + primary[2], primary[1] + primary[3]])
        if self.enable_pose:
            yaw, pitch, roll = self._get_pose().estimate(image, primary)
            s.yaw, s.pitch, s.roll = yaw, pitch, roll
        s.landmarks = self.landmarker.extract(image, primary)
        s.aligned_face, s.aligned_landmarks, s.affine = align(image, s.landmarks)
        if self.enable_parsing:
            s.parsing = self._get_parser().parse(s.aligned_face)
        if self.enable_occlusion:
            s.occlusion_mask = self._get_occ().segment(s.aligned_face)
        s.landmarked_region = landmarked_region(s.aligned_landmarks)
        return s

Source code in ofiqpy/session.py

@dataclass
class Session:
    image: np.ndarray  # original BGR uint8 (H, W, 3)

    # detection
    bbox: np.ndarray | None = None  # (4,) [x1, y1, x2, y2] in original px
    n_faces: int = 0
    face_areas: list[float] = field(default_factory=list)  # descending

    # landmarks (ADNet 98-pt), original-image pixel coords
    landmarks: np.ndarray | None = None  # (98, 2)

    # alignment (616x616 canonical) + transformed landmarks
    aligned_face: np.ndarray | None = None  # (616, 616, 3) BGR uint8
    aligned_landmarks: np.ndarray | None = None  # (98, 2) in aligned space
    affine: np.ndarray | None = None  # (2, 3) transform used

    # pose (degrees)
    yaw: float | None = None
    pitch: float | None = None
    roll: float | None = None

    # segmentation
    parsing: np.ndarray | None = None  # aligned parsing label map
    occlusion_mask: np.ndarray | None = None  # aligned binary occlusion mask

    # derived masks
    landmarked_region: np.ndarray | None = None  # aligned face-region mask

    def has(self, *attrs: str) -> bool:
        return all(getattr(self, a) is not None for a in attrs)

Measures

Source code in ofiqpy/measures/core.py

class Measures:
    def __init__(self, cfg):
        self.cfg = cfg
        self._ssim = None
        self._magface = None
        self._rtree = None
        self._num_trees = None
        self._enet1 = self._enet2 = self._boost = None

    def _load_ml_gz(self, path, loader):
        import gzip
        import tempfile

        data = gzip.decompress(Path(path).read_bytes())
        suffix = ".xml" if str(path).endswith(".xml.gz") else ".yml"
        with tempfile.NamedTemporaryFile("wb", suffix=suffix, delete=False) as f:
            f.write(data)
            tmp = f.name
        return loader(tmp)

    def _rtree_model(self):
        if self._rtree is None:
            p = self.cfg.resolve(self.cfg.measure("Sharpness")["model_path"])
            self._rtree = self._load_ml_gz(p, cv2.ml.RTrees_load)
            self._num_trees = int(self._rtree.getTermCriteria()[1])
        return self._rtree, self._num_trees

    def _expression_models(self):
        if self._enet1 is None:
            m = self.cfg.measure("ExpressionNeutrality")
            self._enet1 = ort.InferenceSession(str(self.cfg.resolve(m["cnn1_model_path"])), providers=["CPUExecutionProvider"])
            self._enet2 = ort.InferenceSession(str(self.cfg.resolve(m["cnn2_model_path"])), providers=["CPUExecutionProvider"])
            self._boost = self._load_ml_gz(self.cfg.resolve(m["adaboost_model_path"]), cv2.ml.Boost_load)
        return self._enet1, self._enet2, self._boost

    def _ssim_sess(self):
        if self._ssim is None:
            p = self.cfg.resolve(self.cfg.measure("CompressionArtifacts")["model_path"])
            self._ssim = ort.InferenceSession(str(p), providers=["CPUExecutionProvider"])
        return self._ssim

    def _magface_sess(self):
        if self._magface is None:
            p = self.cfg.resolve(self.cfg.measure("UnifiedQualityScore")["model_path"])
            self._magface = ort.InferenceSession(str(p), providers=["CPUExecutionProvider"])
        return self._magface

    # --- C03 LuminanceMean (hardcoded double-sigmoid mapping) ---
    def luminance_mean(self, s):
        L = luminance(s.aligned_face)
        hist = cv2.calcHist([L], [0], s.landmarked_region, [256], [0, 256]).flatten()
        hist = hist / hist.sum()
        mean = float(hist @ np.arange(256)) / 255.0
        scalar = _round_half_away(100.0 * _sigmoid(mean, 0.2, 0.05) * (1.0 - _sigmoid(mean, 0.8, 0.05)))
        return "LuminanceMean", max(0.0, min(100.0, scalar)), mean

    # --- C20 HeadSize ---
    def head_size(self, s):
        T = tmetric(s.landmarks)  # ORIGINAL landmarks
        raw = T / s.image.shape[0]  # original height
        cs = abs(raw - 0.45)
        scalar = scalar_conversion(cs, h=200, a=1.0, s=-1.0, x0=0.0, w=0.05, round=True)
        return "HeadSize", scalar, raw

    # --- C17 NoHeadCoverings (custom piecewise mapping) ---
    def no_head_coverings(self, s):
        M = s.parsing  # 400x400
        crop = M[0 : 400 - 204, :]  # top 196 rows
        n = int((crop == CLOTH).sum() + (crop == HAT).sum())
        raw = n / (400 * 196)
        T0, T1, w, x0 = 0.0, 0.95, 0.1, 0.02
        if raw <= T0:
            scalar = 100.0
        elif raw >= T1:
            scalar = 0.0
        else:
            sv = _sigmoid(raw, x0, w)
            s0 = _sigmoid(T0, x0, w)
            s1 = _sigmoid(T1, x0, w)
            scalar = _round_half_away(100.0 * (s1 - sv) / (s1 - s0))
        return "NoHeadCoverings", max(0.0, min(100.0, scalar)), raw

    # --- C09 CompressionArtifacts (reuses OFIQ ssim_248 ONNX) ---
    def compression(self, s):
        crop = s.aligned_face[184 : 616 - 184, 184 : 616 - 184]  # 248x248
        rgb = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB).astype(np.float32)
        mean = np.array([123.7, 116.3, 103.5], np.float32)
        std = np.array([58.4, 57.1, 57.4], np.float32)
        t = (rgb - mean) / std
        blob = np.transpose(t, (2, 0, 1))[None]
        raw = float(self._ssim_sess().run(None, {"input": blob})[0].reshape(-1)[0])
        scalar = scalar_conversion(raw, h=1, a=-0.0278, s=103.0, x0=0.3308, w=0.092, round=True)
        return "CompressionArtifacts", scalar, raw

    # --- UnifiedQualityScore (MagFace magnitude -> sigmoid) ---
    def unified(self, s):
        resized = cv2.resize(s.aligned_face, (192, 192), interpolation=cv2.INTER_LINEAR)
        crop = resized[33 : 192 - 47, 40 : 192 - 40]  # 112x112, BGR
        conv = crop.astype(np.float32) / 255.0
        blob = np.transpose(conv, (2, 0, 1))[None]
        raw = float(self._magface_sess().run(None, {"input": blob})[0].reshape(-1)[0])
        scalar = scalar_conversion(raw, h=100, a=0.0, s=1.0, x0=23.0, w=2.6, round=True)
        return "UnifiedQualityScore", scalar, raw

    # --- HeadPose Yaw/Pitch/Roll (with the OFIQ slot swap) ---
    @staticmethod
    def _cos2(angle_deg):
        c = max(0.0, math.cos(math.radians(angle_deg)))
        return _round_half_away(100.0 * c * c)

    def head_pose(self, s):
        yaw, pitch, roll = s.yaw, s.pitch, s.roll  # geometric
        # slot swap: HeadPoseYaw<-pitch, HeadPosePitch<-yaw, HeadPoseRoll<-roll
        # native value = the (swapped) angle in degrees
        return {
            "HeadPoseYaw": (pitch, self._cos2(pitch)),
            "HeadPosePitch": (yaw, self._cos2(yaw)),
            "HeadPoseRoll": (roll, self._cos2(roll)),
        }

    def compute(self, s) -> dict:
        """Return {component_name: (raw, scalar)} for all implemented measures."""
        from . import geometry as G
        from . import models as M
        from . import pixel as P

        out = {}
        for fn in (self.luminance_mean, self.head_size, self.no_head_coverings, self.compression, self.unified):
            name, scalar, raw = fn(s)
            out[name] = (raw, scalar)
        for fn in (G.inter_eye_distance, G.single_face_present, G.eyes_open, G.mouth_closed):
            name, raw, scalar = fn(s)
            out[name] = (raw, scalar)
        out.update(G.crop_of_face(s))
        # pixel/exposure batch
        for fn in (
            P.background_uniformity,
            P.illumination_uniformity,
            P.luminance_variance,
            P.under_exposure,
            P.over_exposure,
            P.dynamic_range,
            P.natural_colour,
        ):
            name, raw, scalar = fn(s)
            out[name] = (raw, scalar)
        # occlusion measures
        for fn in (M.eyes_visible, M.mouth_occlusion, M.face_occlusion):
            name, raw, scalar = fn(s)
            out[name] = (raw, scalar)
        # model measures
        rtree, nt = self._rtree_model()
        out["Sharpness"] = tuple(M.sharpness(s, rtree, nt)[1:])
        e1, e2, bo = self._expression_models()
        out["ExpressionNeutrality"] = tuple(M.expression_neutrality(s, e1, e2, bo)[1:])
        if s.yaw is not None:
            out.update(self.head_pose(s))
        return out

    def compute_scalars(self, s) -> dict:
        """Return {component_name: scalar} (for the ±1 gate)."""
        return {k: v[1] for k, v in self.compute(s).items()}

compute(s)

Return {component_name: (raw, scalar)} for all implemented measures.

Source code in ofiqpy/measures/core.py

def compute(self, s) -> dict:
    """Return {component_name: (raw, scalar)} for all implemented measures."""
    from . import geometry as G
    from . import models as M
    from . import pixel as P

    out = {}
    for fn in (self.luminance_mean, self.head_size, self.no_head_coverings, self.compression, self.unified):
        name, scalar, raw = fn(s)
        out[name] = (raw, scalar)
    for fn in (G.inter_eye_distance, G.single_face_present, G.eyes_open, G.mouth_closed):
        name, raw, scalar = fn(s)
        out[name] = (raw, scalar)
    out.update(G.crop_of_face(s))
    # pixel/exposure batch
    for fn in (
        P.background_uniformity,
        P.illumination_uniformity,
        P.luminance_variance,
        P.under_exposure,
        P.over_exposure,
        P.dynamic_range,
        P.natural_colour,
    ):
        name, raw, scalar = fn(s)
        out[name] = (raw, scalar)
    # occlusion measures
    for fn in (M.eyes_visible, M.mouth_occlusion, M.face_occlusion):
        name, raw, scalar = fn(s)
        out[name] = (raw, scalar)
    # model measures
    rtree, nt = self._rtree_model()
    out["Sharpness"] = tuple(M.sharpness(s, rtree, nt)[1:])
    e1, e2, bo = self._expression_models()
    out["ExpressionNeutrality"] = tuple(M.expression_neutrality(s, e1, e2, bo)[1:])
    if s.yaw is not None:
        out.update(self.head_pose(s))
    return out

compute_scalars(s)

Return {component_name: scalar} (for the ±1 gate).

Source code in ofiqpy/measures/core.py

def compute_scalars(self, s) -> dict:
    """Return {component_name: scalar} (for the ±1 gate)."""
    return {k: v[1] for k, v in self.compute(s).items()}

Configuration

Parsed OFIQ config with model-path resolution against the OFIQ data root.

Source code in ofiqpy/config.py

class OFIQConfig:
    """Parsed OFIQ config with model-path resolution against the OFIQ data root."""

    def __init__(self, path: Path = OFIQ_CONFIG, data_root: Path = OFIQ_DATA):
        self.cfg = load_config(path)
        self.data_root = Path(data_root)
        self.params = self.cfg["params"]

    def resolve(self, rel_path: str) -> Path:
        p = self.data_root / rel_path
        if not p.exists():
            raise FileNotFoundError(f"OFIQ model not found: {p}")
        return p

    def measure(self, name: str) -> dict:
        return self.params["measures"].get(name, {})

    def sigmoid_params(self, measure: str) -> dict:
        return self.measure(measure).get("Sigmoid", {})

    def detector(self) -> dict:
        return self.params["detector"]["ssd"]

    def landmarks_model(self) -> Path:
        return self.resolve(self.params["landmarks"]["ADNet"]["model_path"])

Output

Source code in ofiqpy/output.py

def header() -> str:
    cols = ["Filename"] + OFIQ_ORDER + [f"{c}.scalar" for c in OFIQ_ORDER] + ["assessment_time_in_ms"]
    return ";".join(cols)

results: {component: (raw, scalar)}. Missing components -> FailureToAssess.

Source code in ofiqpy/output.py

def row(filename: str, results: dict, time_ms: float = 0.0) -> str:
    """results: {component: (raw, scalar)}. Missing components -> FailureToAssess."""
    raws, scalars = [], []
    for c in OFIQ_ORDER:
        raw, scalar = results.get(c, FAILURE)
        raws.append(f"{raw:.6f}" if raw is not None else "nan")
        scalars.append(str(int(scalar)) if scalar is not None else "-1")
    parts = [Path(filename).name] + raws + scalars + [f"{time_ms:.0f}"]
    return ";".join(parts)

Batch

Source code in ofiqpy/batch.py

def run_batch(input_path, output_csv, workers=None, resume=False, progress=True):
    images = discover(Path(input_path))
    out = Path(output_csv)
    done = _already_done(out) if resume else set()
    todo = [im for im in images if im.name not in done]

    write_header = not (resume and out.exists())
    mode = "a" if (resume and out.exists()) else "w"
    n_done = 0
    t0 = time.time()
    with open(out, mode) as fh, ProcessPoolExecutor(max_workers=workers, initializer=_init_worker) as ex:
        if write_header:
            fh.write(header() + "\n")
        for line in ex.map(_assess_one, [str(im) for im in todo], chunksize=1):
            fh.write(line + "\n")
            fh.flush()
            n_done += 1
            if progress and n_done % 25 == 0:
                rate = n_done / max(time.time() - t0, 1e-6)
                print(f"  {n_done}/{len(todo)}  {rate:.1f} img/s", flush=True)
    print(f"done: {n_done} images -> {out} ({len(done)} pre-existing skipped)")
    return out