Codec

def read(
    source: Any, *, allow_remote: bool | None = None, **options: Any
) -> FeatureCollection:
    """Reads a whole document, whatever encoding it turns out to be.

    The encoding is identified from the content, so a document served under no usable name
    - a WFS ``GetFeature`` response, say - is still read correctly.

    Args:
        source: A file path as a string or `os.PathLike`, an ``http(s)`` URL, or a readable
            file-like object.
        allow_remote: Whether an ``http(s)`` source may be retrieved; defaults to the
            ``XMAS_DS_ALLOW_REMOTE`` setting.
        **options: Read options for the encoding; see
            [`gml.decode`][xmas_core.codec.gml.decode] and
            [`jsonfg.decode`][xmas_core.codec.jsonfg.decode].

    Raises:
        DatasourceError: The document could not be identified.
        UnsupportedDatasourceError: The source names a location this package refuses to
            open, or is remote while remote access is disabled.
        ForbiddenDoctypeError: The document is XML and declares a DTD.
        UnsupportedRootElementError: The document is XML, but opens with an element this
            package does not read.
        RemoteFetchError: A remote source could not be retrieved.

    Returns:
        The collection the document holds.
    """
    src = Source(source, allow_remote=allow_remote)
    codec = CODECS.get(src.format) if src.format else None
    if codec is None:
        raise DatasourceError(f"cannot tell what encoding {src!r} is written in")
    return codec.decode(src, **options)

def write(
    collection: FeatureCollection,
    target: Any,
    *,
    format: Format | None = None,
    **options: Any,
) -> None:
    """Writes a collection as a document, replacing whatever is at `target`.

    The encoding comes from the target's name. Where the name does not settle it - a
    file-like object, or an extension this package does not know - pass `format`.

    Args:
        collection: The collection to write.
        target: A file path as a string or `os.PathLike`, or a writable file-like object.
        format: The encoding to write, when the target's name does not give it.
        **options: Write options for the encoding; see
            [`gml.encode`][xmas_core.codec.gml.encode] and
            [`jsonfg.encode`][xmas_core.codec.jsonfg.encode].

    Raises:
        DatasourceError: The encoding could not be determined from the target's name.
        UnsupportedDatasourceError: The target names a location this package refuses to
            write to.
    """
    tgt = Target(target)
    codec = CODECS.get(fmt) if (fmt := format or tgt.format) else None
    if codec is None:
        raise DatasourceError(
            f"cannot tell what encoding to write to {tgt!r}; name it with "
            f"format={'|'.join(CODECS)}"
        )
    codec.encode(collection, tgt, **options)

def fetch_remote(
    url: str,
    *,
    allow_remote: bool | None = None,
    timeout: float | None = None,
    max_bytes: int | None = None,
) -> io.BytesIO:
    """Retrieves a remote document into a buffer.

    The buffer can be handed straight to a `Source`. Retrieval happens here, rather than
    inside `lxml`, so that the parser needs no network access of its own.

    Args:
        url: An ``http`` or ``https`` URL.
        allow_remote: Whether remote retrieval is permitted; defaults to the
            ``XMAS_DS_ALLOW_REMOTE`` setting.
        timeout: Seconds to wait; defaults to the ``XMAS_DS_REMOTE_TIMEOUT`` setting.
        max_bytes: Largest response body accepted; defaults to the
            ``XMAS_DS_REMOTE_MAX_BYTES`` setting.

    Raises:
        UnsupportedDatasourceError: Remote retrieval is disabled, or the URL does not name
            an ``http(s)`` resource.
        RemoteFetchError: The request failed, or the response exceeded `max_bytes`.

    Returns:
        The response body, positioned at the start.
    """
    import httpx2  # deferred so this module does not depend on it being importable

    settings = get_settings()
    if allow_remote is None:
        allow_remote = settings.ds_allow_remote
    if timeout is None:
        timeout = settings.ds_remote_timeout
    if max_bytes is None:
        max_bytes = settings.ds_remote_max_bytes

    if urlsplit(url).scheme.lower() not in _HTTP_SCHEMES:
        raise UnsupportedDatasourceError(
            f"only http(s) datasources can be retrieved, got {url!r}"
        )
    if not allow_remote:
        raise UnsupportedDatasourceError(
            f"remote datasources are disabled, refusing to retrieve {_mask(url)!r}; set "
            "XMAS_DS_ALLOW_REMOTE=1 or pass --allow-remote to enable them"
        )

    buffer = io.BytesIO()
    size = 0
    try:
        with (
            httpx2.Client(
                timeout=timeout, follow_redirects=True, max_redirects=5
            ) as client,
            client.stream("GET", url) as response,
        ):
            response.raise_for_status()
            for chunk in response.iter_bytes():
                size += len(chunk)
                if size > max_bytes:
                    raise RemoteFetchError(
                        f"{url!r} exceeds the {max_bytes} byte limit; raise "
                        "XMAS_DS_REMOTE_MAX_BYTES to allow it"
                    )
                buffer.write(chunk)
            final_url = str(response.url)
    except httpx2.HTTPStatusError as e:
        raise RemoteFetchError(f"{url!r} returned HTTP {e.response.status_code}") from e
    except httpx2.HTTPError as e:
        raise RemoteFetchError(f"could not retrieve {url!r}: {e}") from e

    logger.info(f"retrieved {size} bytes from {final_url}")
    buffer.seek(0)
    return buffer

def decode(
    source: Source,
    *,
    always_generate_ids: bool = False,
    context: dict[str, Any] | None = None,
    **_: Any,
) -> FeatureCollection:
    """Reads a whole GML document.

    Args:
        source: The document to read.
        always_generate_ids: Generate new feature ids even where the ``gml:id`` parses
            as a UUID.
        context: Pydantic validation context for the collection. Setting
            `DROP_INVALID_REFS` in it makes unresolvable and external references be
            dropped rather than raise, and the dropped `InvalidReference`s are
            collected in the same dict under `INVALID_REFS` for the caller to read
            back. It does not cover a malformed ``xlink:href`` - one that is neither
            intra-document (``#GML_<uuid>``, ``urn:uuid:<uuid>``) nor an absolute URI -
            which fails while the feature itself is read and so never reaches the
            collection.
        **_: Options for another encoding, ignored here.

    Raises:
        AppschemaNotFoundError: No supported application schema was declared.
        SRSNotFoundError: The document declares no SRS.
        DuplicateGmlIdError: Two features share a ``gml:id``.

    Returns:
        The collection the document holds.
    """
    appschema = _appschema_from_root(source.root_info)
    root: etree._Element = parse_hardened(source.content).getroot()
    srs = _document_srs(root)

    # every id is settled before the first feature is read, so the same elements are
    # walked twice
    feature_elements = list(_feature_elements(root))
    _settle_ids(feature_elements, root, always_generate_ids)

    collection = {}
    for feature in feature_elements:
        model = decode_feature(
            appschema.model_factory(etree.QName(feature).localname, FeatureType),
            feature,
            srs,
        )
        collection[model.id] = model
    return FeatureCollection.from_features(
        collection, srs.srid, appschema, context=context
    )

def decode_feature[F: FeatureType](
    cls: type[F], element: etree._Element, srs: SRS
) -> F:
    """Decodes one feature's GML element as `cls`.

    Args:
        cls: The feature type the element names.
        element: The feature's element.
        srs: The document's SRS, for a geometry that names none of its own.

    Returns:
        The validated feature.
    """
    return cls.model_validate(_dump_from_element(cls, element, srs))

def encode(
    collection: FeatureCollection,
    target: Target,
    *,
    feature_srs: bool = True,
    **_: Any,
) -> None:
    """Writes a collection as a GML document.

    Args:
        collection: The collection to write.
        target: Where to write it.
        feature_srs: Write ``srsName`` on each feature's geometry.
        **_: Options for another encoding, ignored here.

    Raises:
        AppschemaNotFoundError: The collection's application schema has no GML
            encoding in this package.
    """
    appschema = collection.appschema
    try:
        ns_prefixes, root_tag, schema_location = _GML_PROFILES[appschema.prefix]
    except KeyError:
        raise AppschemaNotFoundError(
            f"cannot write GML for appschema prefix {appschema.prefix!r}"
        ) from None

    namespace = str(appschema.namespace_uri).rstrip("/")
    nsmap: dict[str | None, str] = {None: namespace, **dict(ns_prefixes)}
    root = etree.Element(
        root_tag,
        attrib={
            f"{{{_XSI_NS}}}schemaLocation": schema_location.format(
                ns=namespace, version=appschema.version
            ),
            # every other id written is GML_<uuid>, so the root's own name is unique
            _GML_ID: etree.QName(root_tag).localname,
        },
        # lxml-stubs leave out the None key a default namespace is bound under
        nsmap=nsmap,  # type: ignore[arg-type]  # ty: ignore[invalid-argument-type]
    )

    is_xplan = appschema.prefix == "xplan"
    if is_xplan:
        bounds = etree.SubElement(root, f"{{{_GML_NS}}}boundedBy")

    member_tag = f"{{{_GML_NS if is_xplan else _SF_NS}}}featureMember"
    geoms = []
    for feature in collection.get_features():
        if feature.get_name() in appschema.top_level_featuretypes and (
            wkt := feature.get_geom_wkt()
        ):
            geoms.append(wkt)
        etree.SubElement(root, member_tag).append(
            encode_feature(feature, feature_srs=feature_srs)
        )

    if is_xplan:
        bbox = get_envelope(geoms)
        envelope = etree.SubElement(
            bounds,
            f"{{{_GML_NS}}}Envelope",
            attrib={"srsName": format_srs(collection.srid, "short")},
        )
        etree.SubElement(
            envelope, f"{{{_GML_NS}}}lowerCorner"
        ).text = f"{bbox[0]} {bbox[2]}"
        etree.SubElement(
            envelope, f"{{{_GML_NS}}}upperCorner"
        ).text = f"{bbox[1]} {bbox[3]}"

    target.write_bytes(
        etree.tostring(
            etree.ElementTree(root),
            pretty_print=True,
            xml_declaration=True,
            encoding="UTF-8",
        )
    )

def encode_feature(feature: FeatureType, *, feature_srs: bool = True) -> etree._Element:
    """Encodes one feature as its GML element, the content of a ``gml:featureMember``.

    Args:
        feature: The feature to encode.
        feature_srs: Write ``srsName`` on its geometry.

    Returns:
        The feature's element.
    """
    # dumped once, nested values included; a value in a role admitting several data
    # types names its class under the discriminator
    data = feature.model_dump(
        mode="json",
        exclude_unset=True,
        exclude_none=True,
        context={POLYMORPHIC_DISCRIMINATOR: True},
    )
    return _element_from_dump(type(feature), data, feature_srs)

def decode(
    source: Source,
    *,
    context: dict[str, Any] | None = None,
    **_: Any,
) -> FeatureCollection:
    """Reads a whole JSON-FG document.

    Args:
        source: The document to read.
        context: Pydantic validation context for the collection. Setting
            `DROP_INVALID_REFS` in it makes unresolvable and external references be
            dropped rather than raise, and the dropped `InvalidReference`s are
            collected in the same dict under `INVALID_REFS` for the caller to read
            back.
        **_: Options for another encoding, ignored here.

    Raises:
        JsonFGParseError: The document is not readable JSON-FG.
        AppschemaNotFoundError: No supported application schema was declared.

    Returns:
        The collection the document holds.
    """
    content = _parse(source)
    appschema = _appschema_from_links(content)
    srs = parse_srs(content.get("coordRefSys", None))
    _settle_ids(content, appschema)

    collection = {}
    for feature in content["features"]:
        if not srs.srid:
            # an undeclared JSON-FG CRS is CRS84 - lon/lat, the order a geometry is
            # stored in, so naming it that way orders no swap
            srs = parse_srs(feature.get("coordRefSys", "OGC:CRS84"))
        model = decode_feature(
            appschema.model_factory(_featuretype(feature, content), FeatureType),
            feature,
            srs,
        )
        collection[model.id] = model
    return FeatureCollection.from_features(
        collection, srs.srid, appschema, context=context
    )

def decode_feature[F: FeatureType](
    cls: type[F], feature: dict[str, Any], srs: SRS
) -> F:
    """Decodes one JSON-FG feature object as `cls`.

    Args:
        cls: The feature type the object holds.
        feature: The feature object.
        srs: The document's SRS, for a ``place`` whose feature declares none.

    Returns:
        The validated feature.
    """
    # `properties` is null for a feature carrying none of its own - a presentation
    # object's geometry travels in `place` - and `id` is read the same way so that
    # validating a document's feature directly reports the missing id as the model's
    # own error rather than as a KeyError from here
    data = (feature.get("properties") or {}) | {"id": feature.get("id")}
    if geom := feature.get("place"):
        ogr_geometry = ogr.CreateGeometryFromJson(to_json(geom).decode())
        # a JSON-FG 1.0 document declares the CRS on its root object only; a pre-1.0
        # one put it on the feature
        if raw := feature.get("coordRefSys", None):
            srs = parse_srs(raw)
        if srs.swap_axes:
            # recurses into curve members
            ogr_geometry.SwapXY()
        data[cls.get_geom_field()] = {
            "srid": srs.srid,
            "wkt": ogr_geometry.ExportToWkt(),
        }
    elif geom := feature.get("geometry"):
        ogr_geometry = ogr.CreateGeometryFromJson(to_json(geom).decode())
        data[cls.get_geom_field()] = {
            "srid": 4326,
            "wkt": ogr_geometry.ExportToWkt(),
        }
    return cls.model_validate(data)

def encode(
    collection: FeatureCollection,
    target: Target,
    *,
    single_collection: bool = True,
    write_geometry: bool = True,
    write_bbox: bool = True,
    **_: Any,
) -> None:
    """Writes a collection as one JSON-FG document, or one per feature type.

    Args:
        collection: The collection to write.
        target: Where to write it. With `single_collection=False` it names the family
            rather than a single document, so it has to be a file path.
        single_collection: Write everything as one document. `False` writes one
            document per feature type, each beside the target with the type appended
            to its stem.
        write_geometry: See [`encode_feature`][xmas_core.codec.jsonfg.encode_feature].
        write_bbox: See [`encode_feature`][xmas_core.codec.jsonfg.encode_feature].
        **_: Options for another encoding, ignored here.

    Raises:
        DatasourceError: A fan-out was asked for and the target is not a file path.
    """
    appschema = collection.appschema
    options = {"write_geometry": write_geometry, "write_bbox": write_bbox}
    if single_collection:
        document = _collection_template(appschema, collection.srid)
        document["features"].extend(
            encode_feature(feature, **options)
            for feature in collection.get_features()
            if feature
        )
        _write(document, target)
        return

    # one document per feature type, so the target names a family of files rather than
    # a single one - which only a path can do
    featuretypes: dict[str, list[FeatureType]] = {}
    for feature in collection.get_features():
        featuretypes.setdefault(feature.get_name(), []).append(feature)
    for featuretype, typed_features in featuretypes.items():
        document = _collection_template(appschema, collection.srid, featuretype)
        document["features"].extend(
            encode_feature(feature, **options, write_featuretype=False)
            for feature in typed_features
        )
        _write(document, target.sibling(featuretype))

def encode_feature(
    feature: FeatureType,
    *,
    write_featuretype: bool = True,
    write_geometry: bool = True,
    write_bbox: bool = True,
) -> dict[str, Any]:
    """Encodes one feature as a JSON-FG feature object.

    Args:
        feature: The feature to encode.
        write_featuretype: Write ``featureType``; a document holding one feature type
            declares it on the collection instead.
        write_geometry: Write the WGS 84 ``geometry`` beside ``place``. It is always
            written when there is no ``place``.
        write_bbox: Write ``bbox`` with ``geometry``.

    Returns:
        The feature object.
    """
    properties = feature.model_dump(mode="json", exclude={"id"}, exclude_none=True)
    geom_field = feature.get_geom_field()
    geometry = properties.pop(geom_field, None) if geom_field else None
    data: dict[str, Any] = {
        "type": "Feature",
        "id": feature.id,
        "featureType": feature.get_name(),
        "properties": properties,
        "geometry": None,
        "place": None,
    }
    if not write_featuretype:
        data.pop("featureType")
    if geometry is not None:
        srid = feature.get_geom_srid()
        srs = osr.SpatialReference()
        srs.ImportFromEPSG(srid)
        # the stored eWKT is x/y - easting/longitude first - whatever the authority orders
        srs.SetAxisMappingStrategy(osr.OAMS_TRADITIONAL_GIS_ORDER)
        ogr_geom = ogr.CreateGeometryFromWkt(feature.get_geom_wkt())
        ogr_geom.AssignSpatialReference(srs)
        # `place` carries the geometry as authored, arcs included, in the CRS's own axis
        # order. A linear geometry in WGS 84 lon/lat is forbidden there (JSON-FG 1.0
        # req 11) and `geometry` already holds it, so at 4326 only a curve earns a place.
        if srid != 4326 or ogr_geom.HasCurveGeometry():
            data["place"] = from_json(
                ogr_geom.ExportToJson(
                    options=[
                        "ALLOW_CURVE=YES",
                        "COORDINATE_ORDER=AUTHORITY_COMPLIANT",
                    ]
                )
            )
        # `geometry` is always WGS 84 lon/lat and always linear; it is the only carrier
        # when there is no `place`, so `write_geometry` cannot drop it then.
        if data["place"] is None or write_geometry:
            wgs_geom = ogr_geom.GetLinearGeometry()
            if srid != 4326:
                wgs = osr.SpatialReference()
                wgs.ImportFromEPSG(4326)
                wgs.SetAxisMappingStrategy(osr.OAMS_TRADITIONAL_GIS_ORDER)
                wgs_geom.TransformTo(wgs)
            data["geometry"] = from_json(wgs_geom.ExportToJson())
            if write_bbox:
                min_x, max_x, min_y, max_y = wgs_geom.GetEnvelope()
                data["bbox"] = (min_x, min_y, max_x, max_y)
    return data