opensfm/io.py

import json
import logging
import os
import shutil
from abc import ABC, abstractmethod
from typing import Dict, Any, Iterable, List, IO, Tuple, TextIO, Optional

import cv2
import numpy as np
import pyproj
from opensfm import context, features, geo, pygeometry, pymap, types
from PIL import Image

logger = logging.getLogger(__name__)


def camera_from_json(key: str, obj: Dict[str, Any]) -> pygeometry.Camera:
    """
    Read camera from a json object
    """
    camera = None
    pt = obj.get("projection_type", "perspective")
    if pt == "perspective":
        camera = pygeometry.Camera.create_perspective(
            obj["focal"], obj.get("k1", 0.0), obj.get("k2", 0.0)
        )
    elif pt == "brown":
        camera = pygeometry.Camera.create_brown(
            obj["focal_x"],
            obj["focal_y"] / obj["focal_x"],
            np.array([obj.get("c_x", 0.0), obj.get("c_y", 0.0)]),
            np.array(
                [
                    obj.get("k1", 0.0),
                    obj.get("k2", 0.0),
                    obj.get("k3", 0.0),
                    obj.get("p1", 0.0),
                    obj.get("p2", 0.0),
                ]
            ),
        )
    elif pt == "fisheye":
        camera = pygeometry.Camera.create_fisheye(
            obj["focal"], obj.get("k1", 0.0), obj.get("k2", 0.0)
        )
    elif pt == "fisheye_opencv":
        camera = pygeometry.Camera.create_fisheye_opencv(
            obj["focal_x"],
            obj["focal_y"] / obj["focal_x"],
            np.array([obj.get("c_x", 0.0), obj.get("c_y", 0.0)]),
            np.array(
                [
                    obj.get("k1", 0.0),
                    obj.get("k2", 0.0),
                    obj.get("k3", 0.0),
                    obj.get("k4", 0.0),
                ]
            ),
        )
    elif pt == "fisheye62":
        camera = pygeometry.Camera.create_fisheye62(
            obj["focal_x"],
            obj["focal_y"] / obj["focal_x"],
            np.array([obj.get("c_x", 0.0), obj.get("c_y", 0.0)]),
            np.array(
                [
                    obj.get("k1", 0.0),
                    obj.get("k2", 0.0),
                    obj.get("k3", 0.0),
                    obj.get("k4", 0.0),
                    obj.get("k5", 0.0),
                    obj.get("k6", 0.0),
                    obj.get("p1", 0.0),
                    obj.get("p2", 0.0),
                ]
            ),
        )
    elif pt == "fisheye624":
        camera = pygeometry.Camera.create_fisheye624(
            obj["focal_x"],
            obj["focal_y"] / obj["focal_x"],
            np.array([obj.get("c_x", 0.0), obj.get("c_y", 0.0)]),
            np.array(
                [
                    obj.get("k1", 0.0),
                    obj.get("k2", 0.0),
                    obj.get("k3", 0.0),
                    obj.get("k4", 0.0),
                    obj.get("k5", 0.0),
                    obj.get("k6", 0.0),
                    obj.get("p1", 0.0),
                    obj.get("p2", 0.0),
                    obj.get("s0", 0.0),
                    obj.get("s1", 0.0),
                    obj.get("s2", 0.0),
                    obj.get("s3", 0.0),
                ]
            ),
        )
    elif pt == "radial":
        camera = pygeometry.Camera.create_radial(
            obj["focal_x"],
            obj["focal_y"] / obj["focal_x"],
            np.array([obj.get("c_x", 0.0), obj.get("c_y", 0.0)]),
            np.array(
                [
                    obj.get("k1", 0.0),
                    obj.get("k2", 0.0),
                ]
            ),
        )
    elif pt == "simple_radial":
        camera = pygeometry.Camera.create_simple_radial(
            obj["focal_x"],
            obj["focal_y"] / obj["focal_x"],
            np.array([obj.get("c_x", 0.0), obj.get("c_y", 0.0)]),
            obj.get("k1", 0.0),
        )
    elif pt == "dual":
        camera = pygeometry.Camera.create_dual(
            obj.get("transition", 0.5),
            obj["focal"],
            obj.get("k1", 0.0),
            obj.get("k2", 0.0),
        )
    elif pygeometry.Camera.is_panorama(pt):
        camera = pygeometry.Camera.create_spherical()
    else:
        raise NotImplementedError
    camera.id = key
    camera.width = int(obj.get("width", 0))
    camera.height = int(obj.get("height", 0))
    return camera


def pose_from_json(obj: Dict[str, Any]) -> pygeometry.Pose:
    pose = pygeometry.Pose()
    pose.rotation = obj["rotation"]
    if "translation" in obj:
        pose.translation = obj["translation"]
    return pose


def bias_from_json(obj: Dict[str, Any]) -> pygeometry.Similarity:
    return pygeometry.Similarity(obj["rotation"], obj["translation"], obj["scale"])


def assign_shot_attributes(obj: Dict[str, Any], shot: pymap.Shot) -> None:
    shot.metadata = json_to_pymap_metadata(obj)
    if "scale" in obj:
        shot.scale = obj["scale"]
    if "covariance" in obj:
        shot.covariance = np.array(obj["covariance"])
    if "merge_cc" in obj:
        shot.merge_cc = obj["merge_cc"]
    if "vertices" in obj and "faces" in obj:
        shot.mesh.vertices = obj["vertices"]
        shot.mesh.faces = obj["faces"]


def shot_in_reconstruction_from_json(
    reconstruction: types.Reconstruction,
    key: str,
    obj: Dict[str, Any],
    rig_instance_id: Optional[str] = None,
    rig_camera_id: Optional[str] = None,
    is_pano_shot: bool = False,
) -> pymap.Shot:
    """
    Read shot from a json object and append it to a reconstruction
    """
    pose = pose_from_json(obj)

    if is_pano_shot:
        shot = reconstruction.create_pano_shot(key, obj["camera"], pose)
    else:
        shot = reconstruction.create_shot(
            key, obj["camera"], pose, rig_camera_id, rig_instance_id
        )
    assign_shot_attributes(obj, shot)
    return shot


def single_shot_from_json(
    key: str, obj: Dict[str, Any], camera: pygeometry.Camera
) -> pymap.Shot:
    """
    Read shot from a json object
    """
    pose = pose_from_json(obj)
    shot = pymap.Shot(key, camera, pose)
    assign_shot_attributes(obj, shot)
    return shot


def point_from_json(
    reconstruction: types.Reconstruction, key: str, obj: Dict[str, Any]
) -> pymap.Landmark:
    """
    Read a point from a json object
    """
    point = reconstruction.create_point(key, obj["coordinates"])
    point.color = obj["color"]
    return point


def rig_camera_from_json(key: str, obj: Dict[str, Any]) -> pymap.RigCamera:
    """
    Read a rig cameras from a json object
    """
    pose = pygeometry.Pose()
    pose.rotation = obj["rotation"]
    pose.translation = obj["translation"]
    rig_camera = pymap.RigCamera(pose, key)
    return rig_camera


def rig_cameras_from_json(obj: Dict[str, Any]) -> Dict[str, pymap.RigCamera]:
    """
    Read rig cameras from a json object
    """
    rig_cameras = {}
    for key, value in obj.items():
        rig_cameras[key] = rig_camera_from_json(key, value)
    return rig_cameras


def rig_instance_from_json(
    reconstruction: types.Reconstruction, instance_id: str, obj: Dict[str, Any]
) -> None:
    """
    Read any rig instance from a json shot object
    """
    reconstruction.add_rig_instance(pymap.RigInstance(instance_id))

    pose = pygeometry.Pose()
    pose.rotation = obj["rotation"]
    pose.translation = obj["translation"]
    reconstruction.rig_instances[instance_id].pose = pose


def rig_instance_camera_per_shot(obj: Dict[str, Any]) -> Dict[str, Tuple[str, str]]:
    """
    Given JSON root data, return (rig_instance_id, rig_camera_id) per shot.
    """
    panoshots = set(obj["pano_shots"].keys()) if "pano_shots" in obj else {}
    rig_shots = {}
    if "rig_instances" in obj:
        rig_shots = {
            s_key: (i_key, c_key)
            for i_key, ri in obj["rig_instances"].items()
            for s_key, c_key in ri["rig_camera_ids"].items()
            if s_key not in panoshots
        }
    return rig_shots


def reconstruction_from_json(obj: Dict[str, Any]) -> types.Reconstruction:
    """
    Read a reconstruction from a json object
    """
    reconstruction = types.Reconstruction()

    # Extract cameras
    for key, value in obj["cameras"].items():
        camera = camera_from_json(key, value)
        reconstruction.add_camera(camera)

    # Extract camera biases
    if "biases" in obj:
        for key, value in obj["biases"].items():
            transform = bias_from_json(value)
            reconstruction.set_bias(key, transform)

    # Extract rig models
    if "rig_cameras" in obj:
        for key, value in obj["rig_cameras"].items():
            reconstruction.add_rig_camera(rig_camera_from_json(key, value))

    # Extract rig instances from shots
    if "rig_instances" in obj:
        for key, value in obj["rig_instances"].items():
            rig_instance_from_json(reconstruction, key, value)

    # Extract shots
    rig_shots = rig_instance_camera_per_shot(obj)
    for key, value in obj["shots"].items():
        shot_in_reconstruction_from_json(
            reconstruction,
            key,
            value,
            rig_camera_id=rig_shots[key][1] if key in rig_shots else None,
            rig_instance_id=rig_shots[key][0] if key in rig_shots else None,
            is_pano_shot=False,
        )

    # Extract points
    if "points" in obj:
        for key, value in obj["points"].items():
            point_from_json(reconstruction, key, value)

    # Extract pano_shots
    if "pano_shots" in obj:
        for key, value in obj["pano_shots"].items():
            shot_in_reconstruction_from_json(
                reconstruction, key, value, is_pano_shot=True
            )

    # Extract reference topocentric frame
    if "reference_lla" in obj:
        lla = obj["reference_lla"]
        reconstruction.reference = geo.TopocentricConverter(
            lla["latitude"], lla["longitude"], lla["altitude"]
        )

    return reconstruction


def reconstructions_from_json(obj: List[Dict[str, Any]]) -> List[types.Reconstruction]:
    """
    Read all reconstructions from a json object
    """
    return [reconstruction_from_json(i) for i in obj]


def cameras_from_json(obj: Dict[str, Any]) -> Dict[str, pygeometry.Camera]:
    """
    Read cameras from a json object
    """
    cameras = {}
    for key, value in obj.items():
        cameras[key] = camera_from_json(key, value)
    return cameras


def camera_to_json(camera) -> Dict[str, Any]:
    """
    Write camera to a json object
    """
    if camera.projection_type == "perspective":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal": camera.focal,
            "k1": camera.k1,
            "k2": camera.k2,
        }
    elif camera.projection_type == "brown":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal_x": camera.focal,
            "focal_y": camera.focal * camera.aspect_ratio,
            "c_x": camera.principal_point[0],
            "c_y": camera.principal_point[1],
            "k1": camera.k1,
            "k2": camera.k2,
            "p1": camera.p1,
            "p2": camera.p2,
            "k3": camera.k3,
        }
    elif camera.projection_type == "fisheye":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal": camera.focal,
            "k1": camera.k1,
            "k2": camera.k2,
        }
    elif camera.projection_type == "fisheye_opencv":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal_x": camera.focal,
            "focal_y": camera.focal * camera.aspect_ratio,
            "c_x": camera.principal_point[0],
            "c_y": camera.principal_point[1],
            "k1": camera.k1,
            "k2": camera.k2,
            "k3": camera.k3,
            "k4": camera.k4,
        }
    elif camera.projection_type == "fisheye62":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal_x": camera.focal,
            "focal_y": camera.focal * camera.aspect_ratio,
            "c_x": camera.principal_point[0],
            "c_y": camera.principal_point[1],
            "k1": camera.k1,
            "k2": camera.k2,
            "k3": camera.k3,
            "k4": camera.k4,
            "k5": camera.k5,
            "k6": camera.k6,
            "p1": camera.p1,
            "p2": camera.p2,
        }
    elif camera.projection_type == "fisheye624":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal_x": camera.focal,
            "focal_y": camera.focal * camera.aspect_ratio,
            "c_x": camera.principal_point[0],
            "c_y": camera.principal_point[1],
            "k1": camera.k1,
            "k2": camera.k2,
            "k3": camera.k3,
            "k4": camera.k4,
            "k5": camera.k5,
            "k6": camera.k6,
            "p1": camera.p1,
            "p2": camera.p2,
            "s0": camera.s0,
            "s1": camera.s1,
            "s2": camera.s2,
            "s3": camera.s3,
        }
    elif camera.projection_type == "simple_radial":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal_x": camera.focal,
            "focal_y": camera.focal * camera.aspect_ratio,
            "c_x": camera.principal_point[0],
            "c_y": camera.principal_point[1],
            "k1": camera.k1,
        }
    elif camera.projection_type == "radial":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal_x": camera.focal,
            "focal_y": camera.focal * camera.aspect_ratio,
            "c_x": camera.principal_point[0],
            "c_y": camera.principal_point[1],
            "k1": camera.k1,
            "k2": camera.k2,
        }
    elif camera.projection_type == "dual":
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
            "focal": camera.focal,
            "k1": camera.k1,
            "k2": camera.k2,
            "transition": camera.transition,
        }
    elif pygeometry.Camera.is_panorama(camera.projection_type):
        return {
            "projection_type": camera.projection_type,
            "width": camera.width,
            "height": camera.height,
        }
    else:
        raise NotImplementedError


def shot_to_json(shot: pymap.Shot) -> Dict[str, Any]:
    """
    Write shot to a json object
    """
    obj: Dict[str, Any] = {
        "rotation": list(shot.pose.rotation),
        "translation": list(shot.pose.translation),
        "camera": shot.camera.id,
    }

    if shot.metadata is not None:
        obj.update(pymap_metadata_to_json(shot.metadata))
    if shot.mesh is not None:
        obj["vertices"] = [list(vertice) for vertice in shot.mesh.vertices]
        obj["faces"] = [list(face) for face in shot.mesh.faces]
    if hasattr(shot, "scale"):
        obj["scale"] = shot.scale
    if hasattr(shot, "covariance"):
        obj["covariance"] = shot.covariance.tolist()
    if hasattr(shot, "merge_cc"):
        obj["merge_cc"] = shot.merge_cc
    return obj


def rig_instance_to_json(rig_instance: pymap.RigInstance) -> Dict[str, Any]:
    """
    Write a rig instance to a json object
    """
    return {
        "translation": list(rig_instance.pose.translation),
        "rotation": list(rig_instance.pose.rotation),
        "rig_camera_ids": rig_instance.rig_camera_ids,
    }


def rig_camera_to_json(rig_camera: pymap.RigCamera) -> Dict[str, Any]:
    """
    Write a rig camera to a json object
    """
    obj = {
        "rotation": list(rig_camera.pose.rotation),
        "translation": list(rig_camera.pose.translation),
    }
    return obj


def pymap_metadata_to_json(metadata: pymap.ShotMeasurements) -> Dict[str, Any]:
    obj = {}
    if metadata.orientation.has_value:
        obj["orientation"] = metadata.orientation.value
    if metadata.capture_time.has_value:
        obj["capture_time"] = metadata.capture_time.value
    if metadata.gps_accuracy.has_value:
        obj["gps_dop"] = metadata.gps_accuracy.value
    if metadata.gps_position.has_value:
        obj["gps_position"] = list(metadata.gps_position.value)
    if metadata.accelerometer.has_value:
        obj["accelerometer"] = list(metadata.accelerometer.value)
    if metadata.compass_angle.has_value and metadata.compass_accuracy.has_value:
        obj["compass"] = {
            "angle": metadata.compass_angle.value,
            "accuracy": metadata.compass_accuracy.value,
        }
    else:
        if metadata.compass_angle.has_value:
            obj["compass"] = {"angle": metadata.compass_angle.value}
        elif metadata.compass_accuracy.has_value:
            obj["compass"] = {"accuracy": metadata.compass_accuracy.value}
    if metadata.sequence_key.has_value:
        obj["skey"] = metadata.sequence_key.value
    return obj


def json_to_pymap_metadata(obj: Dict[str, Any]) -> pymap.ShotMeasurements:
    metadata = pymap.ShotMeasurements()
    if obj.get("orientation") is not None:
        metadata.orientation.value = obj.get("orientation")
    if obj.get("capture_time") is not None:
        metadata.capture_time.value = obj.get("capture_time")
    if obj.get("gps_dop") is not None:
        metadata.gps_accuracy.value = obj.get("gps_dop")
    if obj.get("gps_position") is not None:
        metadata.gps_position.value = obj.get("gps_position")
    if obj.get("skey") is not None:
        metadata.sequence_key.value = obj.get("skey")
    if obj.get("accelerometer") is not None:
        metadata.accelerometer.value = obj.get("accelerometer")
    if obj.get("compass") is not None:
        compass = obj.get("compass")
        if "angle" in compass:
            metadata.compass_angle.value = compass["angle"]
        if "accuracy" in compass:
            metadata.compass_accuracy.value = compass["accuracy"]
    return metadata


def point_to_json(point: pymap.Landmark) -> Dict[str, Any]:
    """
    Write a point to a json object
    """
    return {
        "color": list(point.color.astype(float)),
        "coordinates": list(point.coordinates),
    }


def reconstruction_to_json(reconstruction: types.Reconstruction) -> Dict[str, Any]:
    """
    Write a reconstruction to a json object
    """
    obj = {"cameras": {}, "shots": {}, "points": {}, "biases": {}}

    # Extract cameras
    for camera in reconstruction.cameras.values():
        obj["cameras"][camera.id] = camera_to_json(camera)

    # Extract cameras biases
    for camera_id, bias in reconstruction.biases.items():
        obj["biases"][camera_id] = bias_to_json(bias)

    # Extract rig models
    if len(reconstruction.rig_cameras):
        obj["rig_cameras"] = {}
    for rig_camera in reconstruction.rig_cameras.values():
        obj["rig_cameras"][rig_camera.id] = rig_camera_to_json(rig_camera)
    if len(reconstruction.rig_instances):
        obj["rig_instances"] = {}
    for rig_instance in reconstruction.rig_instances.values():
        obj["rig_instances"][rig_instance.id] = rig_instance_to_json(rig_instance)

    # Extract shots
    for shot in reconstruction.shots.values():
        obj["shots"][shot.id] = shot_to_json(shot)

    # Extract points
    for point in reconstruction.points.values():
        obj["points"][point.id] = point_to_json(point)

    # Extract pano_shots
    if hasattr(reconstruction, "pano_shots"):
        if len(reconstruction.pano_shots) > 0:
            obj["pano_shots"] = {}
            for shot in reconstruction.pano_shots.values():
                obj["pano_shots"][shot.id] = shot_to_json(shot)

    # Extract reference topocentric frame
    if reconstruction.reference:
        ref = reconstruction.reference
        obj["reference_lla"] = {
            "latitude": ref.lat,
            "longitude": ref.lon,
            "altitude": ref.alt,
        }

    return obj


def reconstructions_to_json(
    reconstructions: Iterable[types.Reconstruction],
) -> List[Dict[str, Any]]:
    """
    Write all reconstructions to a json object
    """
    return [reconstruction_to_json(i) for i in reconstructions]


def cameras_to_json(cameras: Dict[str, pygeometry.Camera]) -> Dict[str, Dict[str, Any]]:
    """
    Write cameras to a json object
    """
    obj = {}
    for camera in cameras.values():
        obj[camera.id] = camera_to_json(camera)
    return obj


def bias_to_json(bias: pygeometry.Similarity) -> Dict[str, Any]:
    return {
        "rotation": list(bias.rotation),
        "translation": list(bias.translation),
        "scale": bias.scale,
    }


def rig_cameras_to_json(
    rig_cameras: Dict[str, pymap.RigCamera]
) -> Dict[str, Dict[str, Any]]:
    """
    Write rig cameras to a json object
    """
    obj = {}
    for rig_camera in rig_cameras.values():
        obj[rig_camera.id] = rig_camera_to_json(rig_camera)
    return obj


def camera_from_vector(
    camera_id: str,
    width: int,
    height: int,
    projection_type: str,
    parameters: List[float],
) -> pygeometry.Camera:
    """Build a camera from a serialized vector of parameters."""
    if projection_type == "perspective":
        focal, k1, k2 = parameters
        camera = pygeometry.Camera.create_perspective(focal, k1, k2)
    elif projection_type == "brown":
        fx, fy, cx, cy, k1, k2, p1, p2, k3 = parameters
        camera = pygeometry.Camera.create_brown(
            fx, fy / fx, np.array([cx, cy]), np.array([k1, k2, k3, p1, p2])
        )
    elif projection_type == "fisheye":
        focal, k1, k2 = parameters
        camera = pygeometry.Camera.create_fisheye(focal, k1, k2)
    elif projection_type == "fisheye_opencv":
        fx, fy, cx, cy, k1, k2, k3, k4 = parameters
        camera = pygeometry.Camera.create_fisheye_opencv(
            fx, fy / fx, np.array([cx, cy]), np.array([k1, k2, k3, k4])
        )
    elif projection_type == "fisheye62":
        fx, fy, cx, cy, k1, k2, k3, k4, k5, k6, p1, p2 = parameters
        camera = pygeometry.Camera.create_fisheye62(
            fx, fy / fx, np.array([cx, cy]), np.array([k1, k2, k3, k4, k5, k6, p1, p2])
        )
    elif projection_type == "fisheye624":
        fx, fy, cx, cy, k1, k2, k3, k4, k5, k6, p1, p2, s0, s1, s2, s3 = parameters
        camera = pygeometry.Camera.create_fisheye624(
            fx,
            fy / fx,
            np.array([cx, cy]),
            np.array([k1, k2, k3, k4, k5, k6, p1, p2, s0, s1, s2, s3]),
        )
    elif projection_type == "radial":
        fx, fy, cx, cy, k1, k2 = parameters
        camera = pygeometry.Camera.create_radial(
            fx, fy / fx, np.array([cx, cy]), np.array([k1, k2])
        )
    elif projection_type == "simple_radial":
        fx, fy, cx, cy, k1 = parameters
        camera = pygeometry.Camera.create_simple_radial(
            fx, fy / fx, np.array([cx, cy]), k1
        )
    elif projection_type == "dual":
        focal, k1, k2, transition = parameters
        camera = pygeometry.Camera.create_dual(transition, focal, k1, k2)
    elif pygeometry.Camera.is_panorama(projection_type):
        camera = pygeometry.Camera.create_spherical()
    else:
        raise NotImplementedError
    camera.id = camera_id
    camera.width = width
    camera.height = height
    return camera


def camera_to_vector(camera: pygeometry.Camera) -> List[float]:
    """Serialize camera parameters to a vector of floats."""
    if camera.projection_type == "perspective":
        parameters = [camera.focal, camera.k1, camera.k2]
    elif camera.projection_type == "brown":
        parameters = [
            camera.focal,
            camera.focal * camera.aspect_ratio,
            camera.principal_point[0],
            camera.principal_point[1],
            camera.k1,
            camera.k2,
            camera.p1,
            camera.p2,
            camera.k3,
        ]
    elif camera.projection_type == "fisheye":
        parameters = [camera.focal, camera.k1, camera.k2]
    elif camera.projection_type == "fisheye_opencv":
        parameters = [
            camera.focal,
            camera.focal * camera.aspect_ratio,
            camera.principal_point[0],
            camera.principal_point[1],
            camera.k1,
            camera.k2,
            camera.k3,
            camera.k4,
        ]
    elif camera.projection_type == "fisheye62":
        parameters = [
            camera.focal,
            camera.focal * camera.aspect_ratio,
            camera.principal_point[0],
            camera.principal_point[1],
            camera.k1,
            camera.k2,
            camera.k3,
            camera.k4,
            camera.k5,
            camera.k6,
            camera.p1,
            camera.p2,
        ]
    elif camera.projection_type == "fisheye624":
        parameters = [
            camera.focal,
            camera.focal * camera.aspect_ratio,
            camera.principal_point[0],
            camera.principal_point[1],
            camera.k1,
            camera.k2,
            camera.k3,
            camera.k4,
            camera.k5,
            camera.k6,
            camera.p1,
            camera.p2,
            camera.s0,
            camera.s1,
            camera.s2,
            camera.s3,
        ]
    elif camera.projection_type == "radial":
        parameters = [
            camera.focal,
            camera.focal * camera.aspect_ratio,
            camera.principal_point[0],
            camera.principal_point[1],
            camera.k1,
            camera.k2,
        ]
    elif camera.projection_type == "simple_radial":
        parameters = [
            camera.focal,
            camera.focal * camera.aspect_ratio,
            camera.principal_point[0],
            camera.principal_point[1],
            camera.k1,
        ]
    elif camera.projection_type == "dual":
        parameters = [
            camera.focal,
            camera.k1,
            camera.k2,
            camera.transition,
        ]
    elif pygeometry.Camera.is_panorama(camera.projection_type):
        parameters = []
    else:
        raise NotImplementedError

    return parameters


def _read_gcp_list_lines(
    lines: Iterable[str],
    projection,
    exifs: Dict[str, Dict[str, Any]],
) -> List[pymap.GroundControlPoint]:
    points = {}
    for line in lines:
        words = line.split(None, 5)
        easting, northing, alt, pixel_x, pixel_y = map(float, words[:5])
        key = (easting, northing, alt)

        shot_tokens = words[5].split(None)
        shot_id = shot_tokens[0]
        if shot_id not in exifs:
            continue


        if key in points:
            point = points[key]
        else:
            # Convert 3D coordinates
            if np.isnan(alt):
                alt = 0
                has_altitude = False
            else:
                has_altitude = True
            if projection is not None:
                lat, lon = projection.transform(easting, northing)
            else:
                lon, lat = easting, northing

            point = pymap.GroundControlPoint()
            point.id = "unnamed-%d" % len(points)
            point.lla = {"latitude": lat, "longitude": lon, "altitude": alt}
            point.has_altitude = has_altitude

            points[key] = point

        # Convert 2D coordinates
        d = exifs[shot_id]
        coordinates = features.normalized_image_coordinates(
            np.array([[pixel_x, pixel_y]]), d["width"], d["height"]
        )[0]

        o = pymap.GroundControlPointObservation()
        o.shot_id = shot_id
        o.projection = coordinates
        point.add_observation(o)

    return list(points.values())


def _parse_utm_projection_string(line: str) -> str:
    """Convert strings like 'WGS84 UTM 32N' to a proj4 definition."""
    words = line.lower().split()
    assert len(words) == 3
    zone = line.split()[2].upper()
    if zone[-1] == "N":
        zone_number = int(zone[:-1])
        zone_hemisphere = "north"
    elif zone[-1] == "S":
        zone_number = int(zone[:-1])
        zone_hemisphere = "south"
    else:
        zone_number = int(zone)
        zone_hemisphere = "north"
    s = "+proj=utm +zone={} +{} +ellps=WGS84 +datum=WGS84 +units=m +no_defs"
    return s.format(zone_number, zone_hemisphere)


def _parse_projection(line: str):
    """Build a proj4 from the GCP format line."""
    crs_4326 = pyproj.CRS.from_epsg(4326)
    if line.strip() == "WGS84":
        return None
    elif line.upper().startswith("WGS84 UTM"):
        return pyproj.Transformer.from_proj(pyproj.CRS(_parse_utm_projection_string(line)), crs_4326)
    elif "+proj" in line:
        return pyproj.Transformer.from_proj(pyproj.CRS(line), crs_4326)
    elif line.upper().startswith("EPSG:"):
        return pyproj.Transformer.from_proj(pyproj.CRS.from_epsg(int(line.split(":")[1])), crs_4326)
    else:
        raise ValueError("Un-supported geo system definition: {}".format(line))


def _valid_gcp_line(line: str) -> bool:
    stripped = line.strip()
    return stripped != "" and stripped[0] != "#"


def read_gcp_list(fileobj, exif: Dict[str, Any]) -> List[pymap.GroundControlPoint]:
    """Read a ground control points from a gcp_list.txt file.

    It requires the points to be in the WGS84 lat, lon, alt format.
    If reference is None, topocentric data won't be initialized.
    """
    all_lines = fileobj.readlines()
    lines = iter(filter(_valid_gcp_line, all_lines))
    projection = _parse_projection(next(lines))
    points = _read_gcp_list_lines(lines, projection, exif)
    return points


def read_ground_control_points(fileobj: IO) -> List[pymap.GroundControlPoint]:
    """Read ground control points from json file"""
    obj = json_load(fileobj)

    points = []
    for point_dict in obj["points"]:
        point = pymap.GroundControlPoint()
        point.id = point_dict["id"]
        lla = point_dict.get("position")
        if lla:
            point.lla = lla
            point.has_altitude = "altitude" in point.lla

        observations = []
        observing_images = set()
        for o_dict in point_dict["observations"]:
            o = pymap.GroundControlPointObservation()
            o.shot_id = o_dict["shot_id"]
            if o.shot_id in observing_images:
                logger.warning(
                    "GCP {} has multiple observations in image {}".format(
                        point.id, o.shot_id
                    )
                )
            observing_images.add(o.shot_id)
            if "projection" in o_dict:
                o.projection = np.array(o_dict["projection"])
            observations.append(o)
        point.observations = observations
        points.append(point)
    return points


def write_ground_control_points(
    gcp: List[pymap.GroundControlPoint],
    fileobj: IO,
) -> None:
    """Write ground control points to json file."""
    obj = {"points": []}

    for point in gcp:
        point_obj = {}
        point_obj["id"] = point.id
        if point.lla:
            point_obj["position"] = {
                "latitude": point.lla["latitude"],
                "longitude": point.lla["longitude"],
            }
            if point.has_altitude:
                point_obj["position"]["altitude"] = point.lla["altitude"]

        point_obj["observations"] = []
        for observation in point.observations:
            point_obj["observations"].append(
                {
                    "shot_id": observation.shot_id,
                    "projection": tuple(observation.projection),
                }
            )

        obj["points"].append(point_obj)

    json_dump(obj, fileobj)


def json_dump_kwargs(minify: bool = False) -> Dict[str, Any]:
    if minify:
        indent, separators = None, (",", ":")
    else:
        indent, separators = 4, None
    return {"indent": indent, "ensure_ascii": False, "separators": separators}


def json_dump(data, fout, minify=False):
    kwargs = json_dump_kwargs(minify)
    return json.dump(data, fout, **kwargs)


def json_dumps(data, minify=False):
    kwargs = json_dump_kwargs(minify)
    return json.dumps(data, **kwargs)


def json_load(fp):
    return json.load(fp)


def json_loads(text):
    return json.loads(text)


# PLY


def ply_header(
    count_vertices: int, with_normals: bool = False, point_num_views: bool = False
) -> List[str]:
    if with_normals:
        header = [
            "ply",
            "format ascii 1.0",
            "element vertex {}".format(count_vertices),
            "property float x",
            "property float y",
            "property float z",
            "property float nx",
            "property float ny",
            "property float nz",
            "property uchar diffuse_red",
            "property uchar diffuse_green",
            "property uchar diffuse_blue",
        ]
    else:
        header = [
            "ply",
            "format ascii 1.0",
            "element vertex {}".format(count_vertices),
            "property float x",
            "property float y",
            "property float z",
            "property uchar diffuse_red",
            "property uchar diffuse_green",
            "property uchar diffuse_blue",
        ]

    if point_num_views:
        header += ["property uchar views"]

    header += ["end_header"]

    return header


def points_to_ply_string(vertices: List[str], point_num_views: bool = False):
    header = ply_header(len(vertices), point_num_views=point_num_views)
    return "\n".join(header + vertices + [""])


def reconstruction_to_ply(
    reconstruction: types.Reconstruction,
    tracks_manager: Optional[pymap.TracksManager] = None,
    no_cameras: bool = False,
    no_points: bool = False,
    point_num_views: bool = False,
):
    """Export reconstruction points as a PLY string."""
    vertices = []

    if not no_points:
        for point in reconstruction.points.values():
            p, c = point.coordinates, point.color
            s = "{} {} {} {} {} {}".format(
                p[0], p[1], p[2], int(c[0]), int(c[1]), int(c[2])
            )

            if point_num_views and tracks_manager:
                obs_count = point.number_of_observations()
                if obs_count == 0:
                    obs_count = len(tracks_manager.get_track_observations(point.id))
                s += " {}".format(obs_count)

            vertices.append(s)

    if not no_cameras:
        for shot in reconstruction.shots.values():
            o = shot.pose.get_origin()
            R = shot.pose.get_rotation_matrix()
            for axis in range(3):
                c = 255 * np.eye(3)[axis]
                for depth in np.linspace(0, 2, 10):
                    p = o + depth * R[axis]
                    s = "{} {} {} {} {} {}".format(
                        p[0], p[1], p[2], int(c[0]), int(c[1]), int(c[2])
                    )
                    if point_num_views:
                        s += " 0"
                    vertices.append(s)
    return points_to_ply_string(vertices, point_num_views)


def point_cloud_from_ply(
    fp: TextIO,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
    """Load point cloud from a PLY file."""
    all_lines = fp.read().splitlines()
    start = all_lines.index("end_header") + 1
    lines = all_lines[start:]
    n = len(lines)

    points = np.zeros((n, 3), dtype=np.float32)
    normals = np.zeros((n, 3), dtype=np.float32)
    colors = np.zeros((n, 3), dtype=np.uint8)
    labels = np.zeros((n,), dtype=np.uint8)

    for i, row in enumerate(lines):
        words = row.split()
        label = int(words[9])
        points[i] = list(map(float, words[0:3]))
        normals[i] = list(map(float, words[3:6]))
        colors[i] = list(map(int, words[6:9]))
        labels[i] = label

    return points, normals, colors, labels


def point_cloud_to_ply(
    points: np.ndarray,
    normals: np.ndarray,
    colors: np.ndarray,
    labels: np.ndarray,
    fp: TextIO,
):
    fp.write("ply\n")
    fp.write("format ascii 1.0\n")
    fp.write("element vertex {}\n".format(len(points)))
    fp.write("property float x\n")
    fp.write("property float y\n")
    fp.write("property float z\n")
    fp.write("property float nx\n")
    fp.write("property float ny\n")
    fp.write("property float nz\n")
    fp.write("property uchar diffuse_red\n")
    fp.write("property uchar diffuse_green\n")
    fp.write("property uchar diffuse_blue\n")
    fp.write("property uchar class\n")
    fp.write("end_header\n")

    template = "{:.4f} {:.4f} {:.4f} {:.3f} {:.3f} {:.3f} {} {} {} {}\n"
    for i in range(len(points)):
        p, n, c, l = points[i], normals[i], colors[i], labels[i]
        fp.write(
            template.format(
                p[0],
                p[1],
                p[2],
                n[0],
                n[1],
                n[2],
                int(c[0]),
                int(c[1]),
                int(c[2]),
                int(l),
            )
        )


# Filesystem interaction methods
def mkdir_p(path: str) -> None:
    """Make a directory including parent directories."""
    os.makedirs(path, exist_ok=True)


def open_wt(path: str) -> IO[Any]:
    """Open a file in text mode for writing utf-8."""
    return open(path, "w", encoding="utf-8")


def open_rt(path: str) -> IO[Any]:
    """Open a file in text mode for reading utf-8."""
    return open(path, "r", encoding="utf-8")


def imread(
    path: str, grayscale: bool = False, unchanged: bool = False, anydepth: bool = False
):
    with open(path, "rb") as fb:
        return imread_from_fileobject(fb, grayscale, unchanged, anydepth)


def imread_from_fileobject(
    fb, grayscale: bool = False, unchanged: bool = False, anydepth: bool = False
) -> np.ndarray:
    """Load image as an array ignoring EXIF orientation."""
    if context.OPENCV3:
        if grayscale:
            flags = cv2.IMREAD_GRAYSCALE
        elif unchanged:
            flags = cv2.IMREAD_UNCHANGED
        else:
            flags = cv2.IMREAD_COLOR

        try:
            flags |= cv2.IMREAD_IGNORE_ORIENTATION
        except AttributeError:
            logger.warning(
                "OpenCV version {} does not support loading images without "
                "rotating them according to EXIF. Please upgrade OpenCV to "
                "version 3.2 or newer.".format(cv2.__version__)
            )

        if anydepth:
            flags |= cv2.IMREAD_ANYDEPTH
    else:
        if grayscale:
            flags = cv2.CV_LOAD_IMAGE_GRAYSCALE
        elif unchanged:
            flags = cv2.CV_LOAD_IMAGE_UNCHANGED
        else:
            flags = cv2.CV_LOAD_IMAGE_COLOR

        if anydepth:
            flags |= cv2.CV_LOAD_IMAGE_ANYDEPTH

    im_buffer = np.asarray(bytearray(fb.read()), dtype=np.uint8)
    image = cv2.imdecode(im_buffer, flags)

    if image is None:
        raise IOError("Unable to load image")

    if len(image.shape) == 3:
        image[:, :, :3] = image[:, :, [2, 1, 0]]  # Turn BGR to RGB (or BGRA to RGBA)
    return image

    @classmethod
    def imwrite(cls, path: str, image: np.ndarray) -> None:
        with cls.open(path, "wb") as fwb:
            imwrite(fwb, image, path)


def imwrite(path: str, image: np.ndarray):
    with open(path, "wb") as fwb:
        return imwrite_from_fileobject(fwb, image, path)


def imwrite_from_fileobject(fwb, image: np.ndarray, ext: str) -> None:
    """Write an image to a file object"""
    if len(image.shape) == 3:
        image[:, :, :3] = image[:, :, [2, 1, 0]]  # Turn RGB to BGR (or RGBA to BGRA)
    _, im_buffer = cv2.imencode(ext, image)
    fwb.write(im_buffer)


def image_size_from_fileobject(fb):
    """Height and width of an image."""
    try:
        with Image.open(fb) as img:
            width, height = img.size
            return height, width
    except Exception:
        # Slower fallback
        image = imread(fb.name)
        return image.shape[:2]


def image_size(path: str) -> Tuple[int, int]:
    """Height and width of an image."""
    with open(path, "rb") as fb:
        return image_size_from_fileobject(fb)


# IO Filesystem
class IoFilesystemBase(ABC):
    @classmethod
    @abstractmethod
    def exists(cls, path: str):
        pass

    @classmethod
    def ls(cls, path: str):
        pass

    @classmethod
    @abstractmethod
    def isfile(cls, path: str):
        pass

    @classmethod
    @abstractmethod
    def isdir(cls, path: str):
        pass

    @classmethod
    def rm_if_exist(cls, filename: str):
        pass

    @classmethod
    def symlink(cls, src_path: str, dst_path: str, **kwargs):
        pass

    @classmethod
    @abstractmethod
    def open(cls, *args, **kwargs):
        pass

    @classmethod
    @abstractmethod
    def open_wt(cls, path: str):
        pass

    @classmethod
    @abstractmethod
    def open_rt(cls, path: str):
        pass

    @classmethod
    @abstractmethod
    def mkdir_p(cls, path: str):
        pass

    @classmethod
    @abstractmethod
    def imwrite(cls, path: str, image):
        pass

    @classmethod
    @abstractmethod
    def imread(cls, path: str, grayscale=False, unchanged=False, anydepth=False):
        pass

    @classmethod
    @abstractmethod
    def image_size(cls, path: str):
        pass

    @classmethod
    @abstractmethod
    def timestamp(cls, path: str):
        pass


class IoFilesystemDefault(IoFilesystemBase):
    def __init__(self):
        self.type = "default"

    @classmethod
    def exists(cls, path: str) -> str:
        return os.path.exists(path)

    @classmethod
    def ls(cls, path: str) -> List[str]:
        return os.listdir(path)

    @classmethod
    def isfile(cls, path: str) -> str:
        return os.path.isfile(path)

    @classmethod
    def isdir(cls, path: str) -> str:
        return os.path.isdir(path)

    @classmethod
    def rm_if_exist(cls, filename: str) -> None:
        if os.path.islink(filename):
            os.unlink(filename)
        if os.path.exists(filename):
            if os.path.isdir(filename):
                shutil.rmtree(filename)
            else:
                os.remove(filename)

    @classmethod
    def symlink(cls, src_path: str, dst_path: str, **kwargs):
        os.symlink(src_path, dst_path, **kwargs)

    @classmethod
    def open(cls, *args, **kwargs):
        return open(*args, **kwargs)

    @classmethod
    def open_wt(cls, path: str):
        return cls.open(path, "w", encoding="utf-8")

    @classmethod
    def open_rt(cls, path: str):
        return cls.open(path, "r", encoding="utf-8")

    @classmethod
    def mkdir_p(cls, path: str):
        return os.makedirs(path, exist_ok=True)

    @classmethod
    def imread(
        cls,
        path: str,
        grayscale: bool = False,
        unchanged: bool = False,
        anydepth: bool = False,
    ):
        with cls.open(path, "rb") as fb:
            return imread_from_fileobject(fb, grayscale, unchanged, anydepth)

    @classmethod
    def imwrite(cls, path: str, image) -> None:
        with cls.open(path, "wb") as fwb:
            imwrite_from_fileobject(fwb, image, path)

    @classmethod
    def image_size(cls, path: str) -> Tuple[int, int]:
        with cls.open(path, "rb") as fb:
            return image_size_from_fileobject(fb)

    @classmethod
    def timestamp(cls, path: str) -> str:
        return os.path.getmtime(path)