6 Interactive Entity Mapping

from collections import Counter
from pathlib import Path
from pprint import pprint

from presidio_analyzer import AnalyzerEngine, PatternRecognizer

from presidio_evaluator import InputSample
from presidio_evaluator.entity_mapping import (
    CanonicalMapper,
    EntityHierarchy,
    IncompleteMapping,
    IssueType,
    IssueSeverity,
)
from presidio_evaluator.evaluation import SpanEvaluator
from presidio_evaluator.models import PresidioAnalyzerWrapper

# Load dataset
dataset_name = "synth_dataset_v2.json"
dataset = InputSample.read_dataset_json(Path(Path.cwd().parent, "data", dataset_name))
print(f"Loaded {len(dataset)} samples")

# Show dataset entities
dataset_entities = Counter()
for sample in dataset:
    for span in sample.spans:
        dataset_entities[span.entity_type] += 1

print(f"\nDataset has {len(dataset_entities)} entity types:")
pprint(dict(dataset_entities.most_common(10)), compact=True)

tokenizing input:   0%|          | 1/1500 [00:00<04:17,  5.81it/s]

loading model en_core_web_sm


tokenizing input: 100%|██████████| 1500/1500 [00:04<00:00, 307.68it/s]

Loaded 1500 samples

Dataset has 17 entity types:
{'AGE': 74,
 'CREDIT_CARD': 136,
 'DATE_TIME': 119,
 'GPE': 411,
 'NRP': 55,
 'ORGANIZATION': 250,
 'PERSON': 857,
 'PHONE_NUMBER': 92,
 'STREET_ADDRESS': 598,
 'TITLE': 92}

# Load Presidio Analyzer
analyzer = AnalyzerEngine(default_score_threshold=0.4)

# Let's add a custom entity recognizer 
# with an unknown entity type to show how the mapping works
composers_recognizer = PatternRecognizer(supported_entity="COMPOSER", 
                                         deny_list=["A", "Mo", "Bee", "Ba"])

analyzer.registry.add_recognizer(composers_recognizer)

print(f"Presidio supports {len(analyzer.get_supported_entities('en'))} entity types:")
pprint(sorted(analyzer.get_supported_entities("en")), compact=True)

Presidio supports 20 entity types:
['COMPOSER', 'CREDIT_CARD', 'CRYPTO', 'DATE_TIME', 'EMAIL_ADDRESS', 'IBAN_CODE',
 'IP_ADDRESS', 'LOCATION', 'MAC_ADDRESS', 'MEDICAL_LICENSE', 'NRP', 'PERSON',
 'PHONE_NUMBER', 'UK_NHS', 'URL', 'US_BANK_NUMBER', 'US_DRIVER_LICENSE',
 'US_ITIN', 'US_PASSPORT', 'US_SSN']

# Load and predict on a small subset
subset = dataset[:100]
wrapped_analyzer = PresidioAnalyzerWrapper(analyzer_engine=analyzer, language='en')
results_df = wrapped_analyzer.predict_dataset(subset)

# Create mapper and analyze
mapper = CanonicalMapper()

# analyze() identifies all labels in the hierarchy (single phase)
mapper.analyze(results_df)

# render_html() shows a color-coded audit table
# (falls back to plain text outside Jupyter)
mapper.render_html()

[UNRESOLVED] COMPOSER  -- no automatic match found

# View all detected issues with full detail
print(f'Total issues: {len(mapper.get_issues())}')
for issue in mapper.get_issues():
    blocking = 'BLOCKING' if issue.severity != IssueSeverity.INFO else 'non-blocking'
    print(f'  [{issue.severity.value.upper():7}] {issue.type.value:30} | {blocking} | {issue.labels}')
    if issue.annotation_count or issue.prediction_count:
        print(f'             annotations={issue.annotation_count}, predictions={issue.prediction_count}')

Total issues: 3
  [ERROR  ] unresolved                     | BLOCKING | ['COMPOSER']
             annotations=0, predictions=27
  [WARNING] collision_cross_branch         | BLOCKING | ['NRP']
             annotations=0, predictions=4
  [WARNING] dataset_only                   | BLOCKING | ['ORGANIZATION']
             annotations=20, predictions=0

# Programmatic batch resolution — iterate over issues by type

# PREDICTION_ONLY: suppress labels the model predicts but dataset never annotates
prediction_only = [i for i in mapper.get_issues() if i.type == IssueType.PREDICTION_ONLY]
for issue in prediction_only:
    print(f'Suppressing PREDICTION_ONLY: {issue.labels}')
    mapper.map({lbl: None for lbl in issue.labels})

# COLLISION_CROSS_BRANCH: remap to the correct branch
cross_branch = [i for i in mapper.get_issues() if i.type == IssueType.COLLISION_CROSS_BRANCH]
for issue in cross_branch:
    for lbl in issue.labels:
        # Map cross-branch labels to an appropriate canonical entity
        print(f'Needs remapping (COLLISION_CROSS_BRANCH): {lbl}')
        # mapper.map({lbl: 'TARGET_ENTITY'})  # uncomment and set target as appropriate

# Suppress other known noise
mapper.map({'TITLE': None})
mapper.map({'COMPOSER': None})

# Re-analyze to refresh issue detection
mapper.analyze(results_df)
mapper.render_html()

Needs remapping (COLLISION_CROSS_BRANCH): NRP

# Check remaining blocking issues
blocking = [i for i in mapper.get_issues() if i.severity != IssueSeverity.INFO]
if blocking:
    print(f'{len(blocking)} blocking issue(s) remain:')
    for issue in blocking:
        print(f'  [{issue.severity.value}] {issue.type.value}: {issue.labels}')
else:
    print('No blocking issues - ready to call get_mapped_results_dataframe()')

# DATASET_ONLY issues are INFO (non-blocking) - they tell you what the model never predicts
dataset_only = [i for i in mapper.get_issues() if i.type == IssueType.DATASET_ONLY]
if dataset_only:
    all_ds_only = [lbl for i in dataset_only for lbl in i.labels]
    print(f'\nDataset-only entities (INFO, non-blocking): {all_ds_only}')
    print('  These entities never appear in model predictions — recall for them will be 0.')

2 blocking issue(s) remain:
  [warning] collision_cross_branch: ['NRP']
  [warning] dataset_only: ['ORGANIZATION']

Dataset-only entities (INFO, non-blocking): ['ORGANIZATION']
  These entities never appear in model predictions — recall for them will be 0.

# Look up individual entities
print('EMAIL_ADDRESS ->', mapper.get_mapping(entity='EMAIL_ADDRESS'))
print('B-PERSON ->', mapper.get_mapping(entity='B-PERSON'))  # BIO prefix stripped
print('GERMANY_PASSPORT_NUMBER ->', mapper.get_mapping(entity='GERMANY_PASSPORT_NUMBER'))  # COUNTRY tier
print('US_SSN ->', mapper.get_mapping(entity='US_SSN'))

# Full mapping dict for all labels seen so far
full_mapping = mapper.get_mapping()
print('\nFull mapping:')
pprint(full_mapping, compact=True)

EMAIL_ADDRESS -> EMAIL_ADDRESS
B-PERSON -> PERSON
GERMANY_PASSPORT_NUMBER -> PASSPORT
US_SSN -> SSN

Full mapping:
{'AGE': 'AGE',
 'COMPOSER': None,
 'CREDIT_CARD': 'FINANCIAL',
 'DATE_TIME': 'DATE_TIME',
 'DOMAIN_NAME': 'DOMAIN',
 'EMAIL_ADDRESS': 'EMAIL_ADDRESS',
 'GPE': 'GPE',
 'IBAN_CODE': 'FINANCIAL',
 'LOCATION': 'LOCATION',
 'NRP': 'NATIONALITY',
 'ORGANIZATION': 'ORGANIZATION',
 'PERSON': 'PERSON',
 'PHONE_NUMBER': 'PHONE_NUMBER',
 'STREET_ADDRESS': 'ADDRESS',
 'TITLE': None,
 'URL': 'URL',
 'US_DRIVER_LICENSE': 'DRIVER_LICENSE',
 'US_SSN': 'SSN',
 'ZIP_CODE': 'ADDRESS'}

# Demonstrate hierarchy depth
# EntityHierarchy.get_depth() shows where each entity sits in the hierarchy
h = EntityHierarchy(canonical_depth=10)
for entity in ['PERSON', 'NAME', 'FIRST_NAME', 'EMAIL_ADDRESS', 'SSN']:
    try:
        depth = h.get_depth(entity)
        branch = h.canonical_to_branch.get(entity, [])
        print(f'  {entity:25} depth={depth}  branch={" → ".join(branch)}')
    except Exception:
        print(f'  {entity:25} not in hierarchy')

  PERSON                    depth=2  branch=PII → PERSON
  NAME                      depth=3  branch=PII → PERSON → NAME
  FIRST_NAME                depth=4  branch=PII → PERSON → NAME → FIRST_NAME
  EMAIL_ADDRESS             depth=3  branch=PII → CONTACT → EMAIL_ADDRESS
  SSN                       depth=3  branch=PII → GOVERNMENT_ID → SSN

# Inspect resolved labels per record
# After analyze(), inspect _records to see tier and resolved entity per label
print(f'{"Label":25} {"Tier":15} {"Resolved":25} {"Score":10}')
print('-' * 80)
for lbl, rec in sorted(mapper._records.items()):
    score_str = f'{rec.score:.0%}' if rec.score is not None else '-'
    print(f'{lbl:25} {rec.tier:15} {str(rec.resolved):25} {score_str:10}')

Label                     Tier            Resolved                  Score     
--------------------------------------------------------------------------------
AGE                       EXACT           AGE                       -         
COMPOSER                  NONE            None                      -         
CREDIT_CARD               EXACT           FINANCIAL                 -         
DATE_TIME                 EXACT           DATE_TIME                 -         
DOMAIN_NAME               EXACT           DOMAIN                    -         
EMAIL_ADDRESS             EXACT           EMAIL_ADDRESS             -         
GPE                       EXACT           GPE                       -         
IBAN_CODE                 EXACT           FINANCIAL                 -         
LOCATION                  EXACT           LOCATION                  -         
NRP                       EXACT           NATIONALITY               -         
ORGANIZATION              EXACT           ORGANIZATION              -         
PERSON                    EXACT           PERSON                    -         
PHONE_NUMBER              EXACT           PHONE_NUMBER              -         
STREET_ADDRESS            EXACT           ADDRESS                   -         
TITLE                     NONE            None                      -         
URL                       EXACT           URL                       -         
US_DRIVER_LICENSE         EXACT           DRIVER_LICENSE            -         
US_SSN                    EXACT           SSN                       -         
ZIP_CODE                  EXACT           ADDRESS                   -

# Adjusting the fuzzy threshold: lower = more aggressive auto-resolution
loose_mapper = CanonicalMapper(fuzzy_threshold=0.65)
loose_mapper.analyze(results_df)
print(f'Pending with threshold=0.65: {loose_mapper.pending}')
loose_mapper.render_html()

[UNRESOLVED] COMPOSER  -- no automatic match found


Pending with threshold=0.65: ['COMPOSER']

# Interactive resolution in a terminal or Jupyter cell
# (commented out here to avoid blocking automated notebook runs)
# mapper.resolve_interactively()

# Get the full mapping dict once no blocking issues remain
if mapper.pending:
    print(f'Still {len(mapper.pending)} pending: {mapper.pending}')
else:
    entity_mapping = mapper.get_mapping()
    print('Entity mapping ready:')
    pprint(entity_mapping, compact=True)

# Get the MappedResults object — four DataFrames at different granularity levels
try:
    mapped_results = mapper.get_mapped_results_dataframe()
    print(f'\nMappedResults fields: original, binary, branch, detailed')
    print(f'Each DataFrame has {mapped_results.original.shape[0]} rows')
except IncompleteMapping as e:
    print(f'Blocked: {e}')

Entity mapping ready:
{'AGE': 'AGE',
 'COMPOSER': None,
 'CREDIT_CARD': 'FINANCIAL',
 'DATE_TIME': 'DATE_TIME',
 'DOMAIN_NAME': 'DOMAIN',
 'EMAIL_ADDRESS': 'EMAIL_ADDRESS',
 'GPE': 'GPE',
 'IBAN_CODE': 'FINANCIAL',
 'LOCATION': 'LOCATION',
 'NRP': 'NATIONALITY',
 'ORGANIZATION': 'ORGANIZATION',
 'PERSON': 'PERSON',
 'PHONE_NUMBER': 'PHONE_NUMBER',
 'STREET_ADDRESS': 'ADDRESS',
 'TITLE': None,
 'URL': 'URL',
 'US_DRIVER_LICENSE': 'DRIVER_LICENSE',
 'US_SSN': 'SSN',
 'ZIP_CODE': 'ADDRESS'}

MappedResults fields: original, binary, branch, detailed
Each DataFrame has 1832 rows

## 7. Multi-Model Comparison

When evaluating multiple models on the same dataset, use the same mapper instance and call
`analyze()` for each model's results. The resolved mappings are consistent across models
because the hierarchy is fixed.

  Cell In[13], line 4
    `analyze()` for each model's results. The resolved mappings are consistent across models
                              ^
SyntaxError: unterminated string literal (detected at line 4)

# Demonstrate consistent mapping across two models (same mapper, different analyze() calls)
comparison_mapper = CanonicalMapper()

# Model 1
comparison_mapper.analyze(results_df)
records_after_model1 = set(comparison_mapper._records.keys())
print(f'Labels after model 1: {len(records_after_model1)}')

# Model 2 (using same results_df for illustration; in practice this would be a different model)
comparison_mapper.analyze(results_df)
records_after_model2 = set(comparison_mapper._records.keys())
print(f'Labels after model 2: {len(records_after_model2)}')
print('Same label set across models:', records_after_model1 == records_after_model2)

## 8. EntityHierarchy Customisation

Pass a custom hierarchy dict or `EntityHierarchy` instance to `CanonicalMapper(hierarchy=...)` to:
- Add aliases for labels not in the default taxonomy
- Extend the hierarchy with new entity types
- Override the default mappings

from presidio_evaluator.evaluation import SpanEvaluator

# Get mapped results (requires no pending blocking issues)
try:
    mapped_results = mapper.get_mapped_results_dataframe()
    print(f'MappedResults ready: {mapped_results.detailed.shape[0]} rows')
    print(f'Fields: original={mapped_results.original.shape}, binary={mapped_results.binary.shape},')
    print(f'        branch={mapped_results.branch.shape}, detailed={mapped_results.detailed.shape}')
except IncompleteMapping as e:
    print(f'Cannot get MappedResults yet: {e}')
    mapped_results = None

if mapped_results is not None:
    evaluator = SpanEvaluator()

    # calculate_hierarchical_scores returns scores at binary, branch and detailed levels
    hierarchical_results = evaluator.calculate_hierarchical_scores(mapped_results)

    for level, scores in hierarchical_results.items():
        pprint_score = {
            'precision': round(scores.pii_precision, 3),
            'recall': round(scores.pii_recall, 3),
            'f1': round(scores.pii_f, 3),
        }
        print(f'{level:10}: {pprint_score}')

print('''
=== CanonicalMapper Quick Reference ===

# 1. Create (no constructor args needed)
mapper = CanonicalMapper()

# 2. Pre-map known labels before analyze() (optional)
mapper.map({'MY_LABEL': 'PERSON', 'NOISE': None})

# 3. Analyze — Identify-only (single phase)
mapper.analyze(results_df)                    # default min_severity='WARNING'
mapper.analyze(results_df, min_severity='INFO')  # show COLLISION_SAME_BRANCH too

# 4. Review
mapper.render_html()          # HTML audit table (Jupyter)
mapper.get_issues()           # structured MappingIssue list (filtered by min_severity)
mapper.pending                # UNRESOLVED-tier labels

# 5. Resolve blocking issues
mapper.map({'CROSS_BRANCH_LBL': 'CANONICAL'})  # remap COLLISION_CROSS_BRANCH
mapper.map({'PREDICTION_ONLY_LBL': None})       # suppress PREDICTION_ONLY
# mapper.resolve_interactively()               # guided interactive session

# 6. Single entity lookup
mapper.get_mapping(entity='B-PERSON')   # strips BIO, returns resolved value

# 7. Get results
entity_mapping = mapper.get_mapping()    # full {label: resolved} dict (UNRESOLVED excluded)
mapped_results = mapper.get_mapped_results_dataframe()  # raises IncompleteMapping if blocking
# mapped_results.original  — raw labels
# mapped_results.binary    — PII / O
# mapped_results.branch    — PERSON, LOCATION, ...
# mapped_results.detailed  — NAME, GPE, ...


# 8. Evaluate

evaluator = SpanEvaluator(iou_threshold=0.75)''')

scores = evaluator.calculate_score_on_df(results_df=mapped_results.detailed)  # single levelhier = evaluator.calculate_hierarchical_scores(mapped_results)                 # all levels