technical_drawings_extraction/venv/lib/python3.7/site-packages/ezdxf/lldxf/loader.py

# Purpose: DXF structure loader and validator
# Created: 25.01.2018
# Copyright (c) 2018, Manfred Moitzi
# License: MIT License
import logging
from typing import Callable, Dict, Iterable, List, Union, TYPE_CHECKING

from .const import DXFStructureError
from .tags import group_tags, DXFTag, Tags
from .extendedtags import ExtendedTags
from .validator import entity_structure_validator

from ezdxf.options import options

if TYPE_CHECKING:  # import forward declarations
    from ezdxf.eztypes import EntityDB

logger = logging.getLogger('ezdxf')

TagProcessor = Callable[[ExtendedTags], ExtendedTags]
modern_post_load_tag_processors = {}  # type: Dict[str, TagProcessor]
legacy_post_load_tag_processors = {}  # type: Dict[str, TagProcessor]


SectionDict = Dict[str, List[Union[Tags, ExtendedTags]]]


def is_registered(entity: str, legacy: bool = False):
    # just for testing
    processors = legacy_post_load_tag_processors if legacy else modern_post_load_tag_processors
    return entity in processors


def register(entity: str, legacy: bool = False) -> Callable:
    """
    Register (decorator) functions to process from DXF file loaded tags.

    Args:
        entity: DXF type like 'LINE' or 'VERTEX'
        legacy: use for legacy tag structure (DXF version <= AC1009) or modern tag structures

    """
    logger.debug('Register post load tag processor for DXF type: {}; legacy: {}'.format(entity, legacy))

    def decorator(processor: TagProcessor) -> TagProcessor:
        """

        Args:
            processor: function with one parameter 'tags'

        Returns: processor

        """
        processors = legacy_post_load_tag_processors if legacy else modern_post_load_tag_processors
        processors[entity] = processor
        return processor

    return decorator


def load_dxf_structure(tagger: Iterable[DXFTag], ignore_missing_eof: bool = False) -> SectionDict:
    """
    Divide input tag stream from tagger into DXF structure entities. Each DXF structure entity starts with a DXF
    structure (0, ...) tag, and ends before the next DXF structure tag.

    Generated structure:

    each entity is a Tags() object

    {
        'HEADER': [entity],                # 1. section, HEADER section contains only the SECTION head tag
        'CLASSES': [entity, entity, ...],  # 2. section
        'TABLES': [entity, entity, ...],   # 3. section
        ...
        'OBJECTS': [entity, entity, ...],
    }

    {
        'HEADER': [(0, 'SECTION'), (2, 'HEADER'), .... ],  # HEADER section contains only the SECTION head tag
        'CLASSES': [[(0, 'SECTION'), (2, 'CLASSES')], [(0, 'CLASS'), ...], [(0, 'CLASS'), ...]],
        'TABLES': [[(0, 'SECTION'), (2, 'TABLES')], [(0, 'TABLE'), (2, 'VPORT')], [(0, 'VPORT'), ...], ... , [(0, 'ENDTAB')]],
        ...
        'OBJECTS': [[(0, 'SECTION'), (2, 'OBJECTS')], ...]
    }

    Args:
        tagger: generates DXFTag() entities from input data
        ignore_missing_eof: raises DXFStructureError() if False and EOF tag is not present, set to True only in tests

    Returns:
        dict of sections, each section is a list of DXF structure entities as Tags() objects

    """

    def inside_section() -> bool:
        if len(section):
            return section[0][0] == (0, 'SECTION')  # first entity, first tag
        return False

    def outside_section() -> bool:
        if len(section):
            return section[0][0] != (0, 'SECTION')  # first entity, first tag
        return True

    sections = {}  # type: SectionDict
    section = []  # type: List[Tags]
    eof = False
    for entity in group_tags(tagger):
        tag = entity[0]
        if tag == (0, 'SECTION'):
            if inside_section():
                raise DXFStructureError("DXFStructureError: missing ENDSEC tag.")
            if len(section):
                logger.warning('DXF Structure Warning: found tags outside a SECTION, ignored by ezdxf.')
            section = [entity]
        elif tag == (0, 'ENDSEC'):  # not collected
            if outside_section():
                raise DXFStructureError("DXFStructureError: found ENDSEC tag without previous SECTION tag.")
            section_header = section[0]
            if len(section_header) < 2 or section_header[1].code != 2:
                raise DXFStructureError(
                    'DXFStructureError: missing required section NAME tag (2, name) at start of section.')
            name_tag = section_header[1]
            sections[name_tag.value] = section
            section = []  # collect tags outside of sections, but ignore it
        elif tag == (0, 'EOF'):  # not collected
            if eof:
                logger.warning('DXF Structure Warning: found more than one EOF tags.')
            eof = True
        else:
            section.append(entity)
    if inside_section():
        raise DXFStructureError("DXFStructureError: missing ENDSEC tag.")
    if not eof and not ignore_missing_eof:
        raise DXFStructureError('DXFStructureError: missing EOF tag.')
    return sections


DATABASE_EXCLUDE = frozenset(['SECTION', 'ENDSEC', 'EOF', 'TABLE', 'ENDTAB', 'CLASS', 'ACDSRECORD', 'ACDSSCHEMA'])


def load_dxf_entities_into_database(database: 'EntityDB', dxf_entities: List[Tags]) -> Iterable[ExtendedTags]:
    check_tag_structure = options.check_entity_tag_structures
    for entity in dxf_entities:
        if len(entity) == 0:
            raise DXFStructureError('Invalid empty DXF entity.')
        code, dxftype = entity[0]
        if code != 0:
            raise DXFStructureError('Invalid first tag in DXF entity, group code={} .'.format(code))
        if dxftype not in DATABASE_EXCLUDE:
            if check_tag_structure:
                entity = entity_structure_validator(entity)
            entity = ExtendedTags(entity)
            database.add_tags(entity)
        yield entity


def fill_database(database: 'EntityDB', sections: SectionDict, dxfversion: str = 'AC1009') -> None:
    post_processors = legacy_post_load_tag_processors if dxfversion <= 'AC1009' else modern_post_load_tag_processors
    for name in ['TABLES', 'ENTITIES', 'BLOCKS', 'OBJECTS']:
        if name in sections:
            section = sections[name]
            # entities stored in the database are converted from Tags() to ExtendedTags()
            for index, entity in enumerate(load_dxf_entities_into_database(database, section)):
                # entities not stored in database are still Tags() e.g. CLASS
                processor = post_processors.get(entity.dxftype())
                if processor:
                    processor(entity)
                section[index] = entity