class BinaryDecoder:
    """Decoder for the avro binary format.

    NOTE: All attributes and methods on this class should be considered
    private.

    Parameters
    ----------
    fo: file-like
        Input stream

    """

    def __init__(self, fo):
        self.fo = fo

    def read_null(self):
        """null is written as zero bytes."""
        return None

    def read_boolean(self):
        """A boolean is written as a single byte whose value is either 0
        (false) or 1 (true).
        """

        # technically 0x01 == true and 0x00 == false, but many languages will
        # cast anything other than 0 to True and only 0 to False
        return unpack("B", self.fo.read(1))[0] != 0

    def read_long(self):
        """int and long values are written using variable-length, zig-zag
        coding."""
        c = self.fo.read(1)

        # We do EOF checking only here, since most reader start here
        if not c:
            raise StopIteration

        b = ord(c)
        n = b & 0x7F
        shift = 7

        while (b & 0x80) != 0:
            b = ord(self.fo.read(1))
            n |= (b & 0x7F) << shift
            shift += 7

        return (n >> 1) ^ -(n & 1)

    read_int = read_long

    def read_float(self):
        """A float is written as 4 bytes.

        The float is converted into a 32-bit integer using a method equivalent
        to Java's floatToIntBits and then encoded in little-endian format.
        """
        return unpack("<f", self.fo.read(4))[0]

    def read_double(self):
        """A double is written as 8 bytes.

        The double is converted into a 64-bit integer using a method equivalent
        to Java's doubleToLongBits and then encoded in little-endian format.
        """
        return unpack("<d", self.fo.read(8))[0]

    def read_bytes(self):
        """Bytes are encoded as a long followed by that many bytes of data."""
        size = self.read_long()
        return self.fo.read(size)

    def read_utf8(self):
        """A string is encoded as a long followed by that many bytes of UTF-8
        encoded character data.
        """
        return self.read_bytes().decode()

    def read_fixed(self, size):
        """Fixed instances are encoded using the number of bytes declared in the
        schema."""
        return self.fo.read(size)

    def read_enum(self):
        """An enum is encoded by a int, representing the zero-based position of the
        symbol in the schema.
        """
        return self.read_long()

    def read_array_start(self):
        """Arrays are encoded as a series of blocks."""
        self._block_count = self.read_long()

    def read_array_end(self):
        pass

    def _iter_array_or_map(self):
        """Each block consists of a long count value, followed by that many
        array items. A block with count zero indicates the end of the array.
        Each item is encoded per the array's item schema.

        If a block's count is negative, then the count is followed immediately
        by a long block size, indicating the number of bytes in the block.
        The actual count in this case is the absolute value of the count
        written.
        """
        while self._block_count != 0:
            if self._block_count < 0:
                self._block_count = -self._block_count
                # Read block size, unused
                self.read_long()

            for i in range(self._block_count):
                yield
            self._block_count = self.read_long()

    iter_array = _iter_array_or_map
    iter_map = _iter_array_or_map

    def read_map_start(self):
        """Maps are encoded as a series of blocks."""
        self._block_count = self.read_long()

    def read_map_end(self):
        pass

    def read_index(self):
        """A union is encoded by first writing a long value indicating the
        zero-based position within the union of the schema of its value.

        The value is then encoded per the indicated schema within the union.
        """
        return self.read_long() class AvroJSONDecoder:
    """Decoder for the avro JSON format.

    NOTE: All attributes and methods on this class should be considered
    private.

    Parameters
    ----------
    fo: file-like
        Input stream

    """

    def __init__(self, fo):
        self._fo = fo
        self._stack = []
        self._json_data = [json.loads(line.strip()) for line in fo]
        if self._json_data:
            self._current = self._json_data.pop(0)
            self.done = False
        else:
            self.done = True
        self._key = None

    def read_value(self, symbol):
        if isinstance(self._current, dict):
            if self._key not in self._current:
                # Use the default value
                return symbol.get_default()
            else:
                return self._current[self._key]
        else:
            # If we aren't in a dict or a list then this must be a schema which
            # just has a single basic type
            return self._current

    def _push(self, symbol=None):
        self._stack.append((self._current, self._key))
        if isinstance(self._current, dict) and self._key is not None:
            if self._key not in self._current:
                self._current = symbol.get_default()
            else:
                # self._current = self._current.pop(self._key)
                self._current = self._current[self._key]

    def _pop(self):
        self._current, self._key = self._stack.pop()

    def configure(self, schema, named_schemas):
        self._parser = Parser(schema, named_schemas, self.do_action)

    def do_action(self, action):
        if isinstance(action, RecordStart):
            self._push(action)
        elif isinstance(action, RecordEnd):
            self._pop()
        elif isinstance(action, FieldStart):
            self.read_object_key(action.field_name)
        elif isinstance(action, FieldEnd) or isinstance(action, UnionEnd):
            # TODO: Do we need a FieldEnd and UnionEnd symbol?
            pass
        else:
            raise Exception(f"cannot handle: {action}")

    def drain(self):
        self._parser.drain_actions()
        if self._json_data:
            self._current = self._json_data.pop(0)
            self._key = None
        else:
            self.done = True

    def read_null(self):
        symbol = self._parser.advance(Null())
        return self.read_value(symbol)

    def read_boolean(self):
        symbol = self._parser.advance(Boolean())
        return self.read_value(symbol)

    def read_utf8(self):
        symbol = self._parser.advance(String())
        if self._parser.stack[-1] == MapKeyMarker():
            self._parser.advance(MapKeyMarker())
            for key in self._current:
                self._key = key
                break
            return self._key
        else:
            return self.read_value(symbol)

    def read_bytes(self):
        symbol = self._parser.advance(Bytes())
        return self.read_value(symbol).encode("iso-8859-1")

    def read_int(self):
        symbol = self._parser.advance(Int())
        return self.read_value(symbol)

    def read_long(self):
        symbol = self._parser.advance(Long())
        return self.read_value(symbol)

    def read_float(self):
        symbol = self._parser.advance(Float())
        return self.read_value(symbol)

    def read_double(self):
        symbol = self._parser.advance(Double())
        return self.read_value(symbol)

    def read_enum(self):
        symbol = self._parser.advance(Enum())
        enum_labels = self._parser.pop_symbol()  # pop the enumlabels
        # TODO: Should we verify the value is one of the symbols?
        label = self.read_value(symbol)
        return enum_labels.labels.index(label)

    def read_fixed(self, size):
        symbol = self._parser.advance(Fixed())
        return self.read_value(symbol).encode("iso-8859-1")

    def read_map_start(self):
        symbol = self._parser.advance(MapStart())
        self._push(symbol)

    def read_object_key(self, key):
        self._key = key

    def iter_map(self):
        while len(self._current) > 0:
            yield
            del self._current[self._key]

    def read_map_end(self):
        self._parser.advance(MapEnd())
        self._pop()

    def read_array_start(self):
        symbol = self._parser.advance(ArrayStart())
        self._push(symbol)
        self._key = None

    def read_array_end(self):
        self._parser.advance(ArrayEnd())
        self._pop()

    def iter_array(self):
        while len(self._current) > 0:
            self._push()
            self._current = self._current.pop(0)
            yield
            self._pop()
            self._parser.advance(ItemEnd())

    def read_index(self):
        self._parser.advance(Union())
        alternative_symbol = self._parser.pop_symbol()

        # TODO: Try to clean this up.
        # A JSON union is encoded like this: {"union_field": {int: 32}} and so
        # what we are doing is trying to change that into {"union_field": 32}
        # before eventually reading the value of "union_field"
        if self._key is None:
            # If self._key is None, self._current is an item in an array
            if self._current is None:
                label = "null"
            else:
                label, data = self._current.popitem()
                self._current = data
                # TODO: Do we need to do this?
                self._parser.push_symbol(UnionEnd())
        else:
            # self._current is a JSON object and self._key should be the name
            # of the union field
            if self._key not in self._current:
                self._current[self._key] = {
                    alternative_symbol.labels[0]: alternative_symbol.get_default()
                }

            if self._current[self._key] is None:
                label = "null"
            else:
                label, data = self._current[self._key].popitem()
                self._current[self._key] = data
                # TODO: Do we need to do this?
                self._parser.push_symbol(UnionEnd())

        index = alternative_symbol.labels.index(label)
        symbol = alternative_symbol.get_symbol(index)
        self._parser.push_symbol(symbol)
        return index FINGERPRINT_ALGORITHMS = _schema.FINGERPRINT_ALGORITHMS class SchemaResolutionError(Exception):
    pass MAGIC = b"Obj" + chr(VERSION).encode() SYNC_SIZE = 16 HEADER_SCHEMA = {
    "type": "record",
    "name": "org.apache.avro.file.Header",
    "fields": [
        {
            "name": "magic",
            "type": {"type": "fixed", "name": "magic", "size": len(MAGIC)},
        },
        {"name": "meta", "type": {"type": "map", "values": "bytes"}},
        {"name": "sync", "type": {"type": "fixed", "name": "sync", "size": SYNC_SIZE}},
    ],
} def missing_codec_lib(codec, library):
    def missing(fo):
        raise ValueError(
            f"{codec} codec is supported but you need to install {library}"
        )

    return missing MCS_PER_HOUR = MCS_PER_MINUTE * 60 MCS_PER_MINUTE = MCS_PER_SECOND * 60 MCS_PER_SECOND = 1000000 MLS_PER_HOUR = MLS_PER_MINUTE * 60 MLS_PER_MINUTE = MLS_PER_SECOND * 60 MLS_PER_SECOND = 1000 DAYS_SHIFT = datetime.date(1970, 1, 1).toordinal() NAMED_TYPES = {"record", "enum", "fixed", "error"}
from io import BytesIO
from struct import error as StructError
from datetime import datetime, time, date, timezone, timedelta
from decimal import Context
from uuid import UUID
from .io.binary_decoder import BinaryDecoder
from .io.json_decoder import AvroJSONDecoder
from .schema import extract_record_type, extract_logical_type, parse_schema
from ._read_common import (
    SchemaResolutionError,
    MAGIC,
    SYNC_SIZE,
    HEADER_SCHEMA,
    missing_codec_lib,
)
from .const import (
    MCS_PER_HOUR,
    MCS_PER_MINUTE,
    MCS_PER_SECOND,
    MLS_PER_HOUR,
    MLS_PER_MINUTE,
    MLS_PER_SECOND,
    DAYS_SHIFT,
    NAMED_TYPES,
)
import bz2
import lzma
import zlib
import json
    import snappy
    import zstandard as zstd
    import lz4.block
        import warnings
        return True
    if isinstance(writer_type, dict) or isinstance(reader_type, dict):
        try:
            return match_schemas(writer_type, reader_type)
        except SchemaResolutionError:
            return False
    if writer_type == reader_type:
        return True
    # promotion cases
    elif writer_type == "int" and reader_type in ["long", "float", "double"]:
        return True
    elif writer_type == "long" and reader_type in ["float", "double"]:
        return True
    elif writer_type == "float" and reader_type == "double":
        return True
    elif writer_type == "string" and reader_type == "bytes":
        return True
    elif writer_type == "bytes" and reader_type == "string":
        return True
    return False


def match_schemas(w_schema, r_schema):
    error_msg = f"Schema mismatch: {w_schema} is not {r_schema}"
    if isinstance(w_schema, list):
        # If the writer is a union, checks will happen in read_union after the
        # correct schema is known
        return r_schema
    elif isinstance(r_schema, list):
        # If the reader is a union, ensure one of the new schemas is the same
        # as the writer
        for schema in r_schema:
            if match_types(w_schema, schema):
                return schema
        else:
            raise SchemaResolutionError(error_msg)
    else:
        # Check for dicts as primitive types are just strings
        if isinstance(w_schema, dict):
            w_type = w_schema["type"]
        else:
            w_type = w_schema
        if isinstance(r_schema, dict):
            r_type = r_schema["type"]
        else:
            r_type = r_schema

        if w_type == r_type == "map":
            if match_types(w_schema["values"], r_schema["values"]):
                return r_schema
        elif w_type == r_type == "array":
            if match_types(w_schema["items"], r_schema["items"]):
                return r_schema
        elif w_type in NAMED_TYPES and r_type in NAMED_TYPES:
            if w_type == r_type == "fixed" and w_schema["size"] != r_schema["size"]:
                raise SchemaResolutionError(
                    f"Schema mismatch: {w_schema} size is different than {r_schema} size"
                )
            if w_schema["name"] == r_schema["name"] or w_schema["name"] in r_schema.get(
                "aliases", []
            ):
                return r_schema
        elif match_types(w_type, r_type):
            return r_schema
        raise SchemaResolutionError(error_msg)


def read_null(
    decoder,
    writer_schema=None,
    named_schemas=None,
    reader_schema=None,
    return_record_name=False,
):
    return decoder.read_null()


def skip_null(decoder, writer_schema=None, named_schemas=None):
    decoder.read_null()


def read_boolean(
    decoder,
    writer_schema=None,
    named_schemas=None,
    reader_schema=None,
    return_record_name=False,
):
    return decoder.read_boolean()


def skip_boolean(decoder, writer_schema=None, named_schemas=None):
    decoder.read_boolean()


def read_timestamp_millis(data, writer_schema=None, reader_schema=None):
    # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
    return epoch + timedelta(microseconds=data * 1000)


def read_local_timestamp_millis(
    data: int, writer_schema=None, reader_schema=None
) -> datetime:
    # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
    return epoch_naive + timedelta(microseconds=data * 1000)


def read_timestamp_micros(data, writer_schema=None, reader_schema=None):
    # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
    return epoch + timedelta(microseconds=data)


def read_local_timestamp_micros(
    data: int, writer_schema=None, reader_schema=None
) -> datetime:
    # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
    return epoch_naive + timedelta(microseconds=data)


def read_date(data, writer_schema=None, reader_schema=None):

    return date.fromordinal(data + DAYS_SHIFT)