class NERAnnotation:""" Data structure for NER annotation """def__init__(self, text: str):self.text = textself.entities = []def add_entity(self, start: int, end: int, label: str, text: str):""" Add entity span Args: start: Character offset start end: Character offset end label: Entity type (PERSON, ORG, LOC, etc.) text: The actual text span """ entity = {'start': start,'end': end,'label': label,'text': text }# Validate no overlap with existing entitiesifnotself._check_overlap(start, end):self.entities.append(entity)returnTruereturnFalsedef _check_overlap(self, start: int, end: int) ->bool:"""Check if span overlaps with existing entities"""for entity inself.entities:ifnot (end <= entity['start'] or start >= entity['end']):returnTruereturnFalsedef to_dict(self):return {'text': self.text,'entities': self.entities }# Example usageexample_text ="Apple Inc. CEO Tim Cook announced new products in Cupertino."annotation = NERAnnotation(example_text)annotation.add_entity(0, 10, 'ORG', 'Apple Inc.')annotation.add_entity(16, 24, 'PERSON', 'Tim Cook')annotation.add_entity(54, 63, 'LOC', 'Cupertino')print(annotation.to_dict())
{'text': 'Apple Inc. CEO Tim Cook announced new products in Cupertino.', 'entities': [{'start': 0, 'end': 10, 'label': 'ORG', 'text': 'Apple Inc.'}, {'start': 16, 'end': 24, 'label': 'PERSON', 'text': 'Tim Cook'}, {'start': 54, 'end': 63, 'label': 'LOC', 'text': 'Cupertino'}]}
Common Edge Cases:
Code
import pandas as pdedge_cases = pd.DataFrame({'Scenario': ['Nested entities','Discontinuous mentions','Ambiguous boundaries','Coordinated entities','Metonymy','Generic vs specific' ],'Example': ['[Bank of [America]_ORG]_ORG','New York and Los Angeles (two separate LOC)','U.S. vs U.S vs US','Google and Facebook','Wall Street (location vs financial industry)','apple (fruit vs Apple company)' ],'Resolution Strategy': ['Annotate longest span only','Mark each entity separately','Normalize to consistent form','Mark each entity individually','Use context to decide','Require capitalization for ORG' ],'Guideline Priority': ['High','Medium','High','Medium','High','High' ]})edge_cases
Table 17.2: NER edge cases and resolution strategies
Scenario
Example
Resolution Strategy
Guideline Priority
0
Nested entities
[Bank of [America]_ORG]_ORG
Annotate longest span only
High
1
Discontinuous mentions
New York and Los Angeles (two separate LOC)
Mark each entity separately
Medium
2
Ambiguous boundaries
U.S. vs U.S vs US
Normalize to consistent form
High
3
Coordinated entities
Google and Facebook
Mark each entity individually
Medium
4
Metonymy
Wall Street (location vs financial industry)
Use context to decide
High
5
Generic vs specific
apple (fruit vs Apple company)
Require capitalization for ORG
High
Synthetic Training Data Generation:
import randomfrom typing import List, Tupleclass SyntheticNERGenerator:""" Generate synthetic NER training data for testing annotation workflows """def__init__(self):self.templates = ["{PERSON} works at {ORG} in {LOC}.","{ORG} announced that {PERSON} will lead the {ORG} division.","The {ORG} headquarters in {LOC} employs {PERSON}.","{PERSON} traveled from {LOC} to {LOC} for {ORG} business.", ]self.entities = {'PERSON': ['Alice Johnson', 'Bob Smith', 'Carol Martinez', 'David Lee', 'Emma Wilson', 'Frank Chen'],'ORG': ['TechCorp', 'DataSystems Inc.', 'Global Analytics', 'Innovation Labs', 'Future Solutions'],'LOC': ['New York', 'San Francisco', 'London', 'Singapore', 'Berlin', 'Tokyo'] }def generate(self, n: int=10) -> List[dict]:"""Generate n synthetic examples""" examples = []for _ inrange(n): template = random.choice(self.templates) entities_used = {}# Replace placeholders text = templatefor entity_type in ['PERSON', 'ORG', 'LOC']: count = text.count(f'{{{entity_type}}}') entities_used[entity_type] = random.sample(self.entities[entity_type], count )# Build annotated version annotations = []for entity_type in ['PERSON', 'ORG', 'LOC']:for entity_text in entities_used[entity_type]: placeholder =f'{{{entity_type}}}' start = text.find(placeholder)if start !=-1: text = text.replace(placeholder, entity_text, 1) annotations.append({'start': start,'end': start +len(entity_text),'label': entity_type,'text': entity_text }) examples.append({'text': text,'entities': sorted(annotations, key=lambda x: x['start']) })return examples# Generate examplesgenerator = SyntheticNERGenerator()synthetic_examples = generator.generate(5)for i, example inenumerate(synthetic_examples, 1):print(f"\nExample {i}:")print(f"Text: {example['text']}")print(f"Entities: {example['entities']}")
Example 1:
Text: Frank Chen works at TechCorp in Berlin.
Entities: [{'start': 0, 'end': 10, 'label': 'PERSON', 'text': 'Frank Chen'}, {'start': 20, 'end': 28, 'label': 'ORG', 'text': 'TechCorp'}, {'start': 32, 'end': 38, 'label': 'LOC', 'text': 'Berlin'}]
Example 2:
Text: Emma Wilson traveled from Berlin to Tokyo for DataSystems Inc. business.
Entities: [{'start': 0, 'end': 11, 'label': 'PERSON', 'text': 'Emma Wilson'}, {'start': 26, 'end': 32, 'label': 'LOC', 'text': 'Berlin'}, {'start': 36, 'end': 41, 'label': 'LOC', 'text': 'Tokyo'}, {'start': 45, 'end': 61, 'label': 'ORG', 'text': 'DataSystems Inc.'}]
Example 3:
Text: The Innovation Labs headquarters in London employs Frank Chen.
Entities: [{'start': 4, 'end': 19, 'label': 'ORG', 'text': 'Innovation Labs'}, {'start': 36, 'end': 42, 'label': 'LOC', 'text': 'London'}, {'start': 40, 'end': 50, 'label': 'PERSON', 'text': 'Frank Chen'}]
Example 4:
Text: Alice Johnson traveled from London to San Francisco for TechCorp business.
Entities: [{'start': 0, 'end': 13, 'label': 'PERSON', 'text': 'Alice Johnson'}, {'start': 28, 'end': 34, 'label': 'LOC', 'text': 'London'}, {'start': 38, 'end': 51, 'label': 'LOC', 'text': 'San Francisco'}, {'start': 47, 'end': 55, 'label': 'ORG', 'text': 'TechCorp'}]
Example 5:
Text: Carol Martinez works at Future Solutions in London.
Entities: [{'start': 0, 'end': 14, 'label': 'PERSON', 'text': 'Carol Martinez'}, {'start': 24, 'end': 40, 'label': 'ORG', 'text': 'Future Solutions'}, {'start': 44, 'end': 50, 'label': 'LOC', 'text': 'London'}]
import pandas as pdvideo_challenges = pd.DataFrame({'Challenge': ['Temporal Consistency','Occlusion Handling','Object Re-identification','Motion Blur','Scale Variation','Annotation Volume','Keyframe Selection','Interpolation Accuracy' ],'Impact on Quality': ['High - ID switches common','High - Lost tracks','High - Same object different IDs','Medium - Boundary uncertainty','Medium - Small object detection','High - Prohibitive manual effort','Medium - Miss important frames','Medium - Drift between keyframes' ],'Mitigation Strategy': ['Track review UI with temporal context','Flag occlusion states explicitly','Visual similarity matching tools','Multiple frame context','Consistent zoom level','Keyframe + interpolation workflow','Scene change detection','Optical flow-based interpolation' ],'Cost Impact': ['3-5x vs single frame','1.5x (review time)','2x (manual correction)','1.2x (slower annotation)','1.3x (zoom overhead)','10-30x (30 fps video)','0.5x (reduces frames)','0.3x (auto-fills frames)' ]})video_challenges
Table 17.3: Video annotation complexity factors
Challenge
Impact on Quality
Mitigation Strategy
Cost Impact
0
Temporal Consistency
High - ID switches common
Track review UI with temporal context
3-5x vs single frame
1
Occlusion Handling
High - Lost tracks
Flag occlusion states explicitly
1.5x (review time)
2
Object Re-identification
High - Same object different IDs
Visual similarity matching tools
2x (manual correction)
3
Motion Blur
Medium - Boundary uncertainty
Multiple frame context
1.2x (slower annotation)
4
Scale Variation
Medium - Small object detection
Consistent zoom level
1.3x (zoom overhead)
5
Annotation Volume
High - Prohibitive manual effort
Keyframe + interpolation workflow
10-30x (30 fps video)
6
Keyframe Selection
Medium - Miss important frames
Scene change detection
0.5x (reduces frames)
7
Interpolation Accuracy
Medium - Drift between keyframes
Optical flow-based interpolation
0.3x (auto-fills frames)
17.3.2 Object Tracking Data Structure
from dataclasses import dataclass, fieldfrom typing import Dict, List, Optionalfrom enum import Enumclass ObjectState(Enum):"""Object visibility states""" VISIBLE ="visible" OCCLUDED ="occluded" OUT_OF_FRAME ="out_of_frame" UNCERTAIN ="uncertain"@dataclassclass TrackedObject:""" Single object tracked across multiple frames """ track_id: int label: str frames: Dict[int, dict] = field(default_factory=dict)def add_detection(self, frame_num: int, bbox: BoundingBox, state: ObjectState = ObjectState.VISIBLE, keyframe: bool=False):""" Add detection at specific frame Args: frame_num: Frame number bbox: Bounding box at this frame state: Visibility state keyframe: Is this a manually annotated keyframe? """self.frames[frame_num] = {'bbox': bbox,'state': state,'keyframe': keyframe }def interpolate_frames(self, start_frame: int, end_frame: int, method: str='linear'):""" Interpolate bounding boxes between keyframes Args: start_frame: Start frame (must be annotated) end_frame: End frame (must be annotated) method: 'linear', 'cubic', or 'optical_flow' """if start_frame notinself.frames or end_frame notinself.frames:raiseValueError("Both start and end frames must be annotated")if method =='linear':self._linear_interpolation(start_frame, end_frame)elif method =='cubic':self._cubic_interpolation(start_frame, end_frame)else:raiseValueError(f"Unknown interpolation method: {method}")def _linear_interpolation(self, start_frame: int, end_frame: int):"""Linear interpolation of bounding boxes""" start_bbox =self.frames[start_frame]['bbox'] end_bbox =self.frames[end_frame]['bbox'] num_frames = end_frame - start_framefor i inrange(1, num_frames): frame_num = start_frame + i alpha = i / num_frames# Interpolate each coordinate x = start_bbox.x + alpha * (end_bbox.x - start_bbox.x) y = start_bbox.y + alpha * (end_bbox.y - start_bbox.y) w = start_bbox.width + alpha * (end_bbox.width - start_bbox.width) h = start_bbox.height + alpha * (end_bbox.height - start_bbox.height) interpolated_bbox = BoundingBox( x=x, y=y, width=w, height=h, label=self.label )self.add_detection( frame_num, interpolated_bbox, state=ObjectState.VISIBLE, keyframe=False )def get_trajectory(self) -> List[Tuple[int, BoundingBox]]:"""Get sorted list of (frame, bbox) tuples"""returnsorted( [(f, d['bbox']) for f, d inself.frames.items()], key=lambda x: x[0] )def coverage(self, total_frames: int) ->float:"""Calculate what % of frames have annotations"""returnlen(self.frames) / total_frames@dataclassclass VideoAnnotation:""" Complete annotation for a video """ video_id: str fps: float total_frames: int width: int height: int tracks: Dict[int, TrackedObject] = field(default_factory=dict) _next_track_id: int= field(default=0, init=False)def create_track(self, label: str, first_frame: int, bbox: BoundingBox) ->int:""" Create new tracked object Returns: track_id of created track """ track_id =self._next_track_idself._next_track_id +=1 track = TrackedObject(track_id=track_id, label=label) track.add_detection(first_frame, bbox, keyframe=True)self.tracks[track_id] = trackreturn track_iddef get_frame_annotations(self, frame_num: int) -> List[Tuple[int, BoundingBox]]:"""Get all bounding boxes for a specific frame""" annotations = []for track_id, track inself.tracks.items():if frame_num in track.frames: annotations.append(( track_id, track.frames[frame_num]['bbox'] ))return annotationsdef export_to_mot_format(self, output_path: str):""" Export to MOT Challenge format Format: <frame>, <id>, <bb_left>, <bb_top>, <bb_width>, <bb_height>, <conf>, <x>, <y>, <z> """withopen(output_path, 'w') as f:for track_id, track inself.tracks.items():for frame_num insorted(track.frames.keys()): bbox = track.frames[frame_num]['bbox']# MOT format uses 1-indexed frames f.write(f"{frame_num +1},{track_id},{bbox.x},{bbox.y},"f"{bbox.width},{bbox.height},1,-1,-1,-1\n")# Example usagevideo_ann = VideoAnnotation( video_id="traffic_001", fps=30.0, total_frames=900, # 30 seconds width=1920, height=1080)# Create track for a carcar_track_id = video_ann.create_track( label="car", first_frame=0, bbox=BoundingBox(100, 200, 150, 100, "car"))# Add keyframe at frame 30video_ann.tracks[car_track_id].add_detection( frame_num=30, bbox=BoundingBox(250, 220, 160, 110, "car"), keyframe=True)# Interpolate frames 0-30video_ann.tracks[car_track_id].interpolate_frames(0, 30, method='linear')# Get all annotations at frame 15frame_15_anns = video_ann.get_frame_annotations(15)print(f"Frame 15 has {len(frame_15_anns)} objects")print(f"Car position at frame 15: {frame_15_anns[0][1].to_xywh()}")
Frame 15 has 1 objects
Car position at frame 15: (175.0, 210.0, 155.0, 105.0)
17.3.3 Video Annotation Workflow
flowchart TD
A["Load Video"] --> B["Automatic Scene Detection"]
B --> C["Present Keyframes to Annotator"]
C --> D["Annotator Labels Objects in Keyframes"]
D --> E{"Object Exits or Enters Scene"}
E -->|Yes| F["Mark Entry and Exit Frames"]
E -->|No| G["Continue to Next Keyframe"]
F --> G
G --> H["Automatic Interpolation"]
H --> I["Annotator Reviews Interpolated Frames"]
I --> J{"Quality Acceptable"}
J -->|No| K["Add Correction Keyframe"]
K --> H
J -->|Yes| L["Mark Track as Complete"]
L --> M{"More Objects"}
M -->|Yes| D
M -->|No| N["Export Annotations"]
Figure 17.2: Keyframe-based video annotation workflow
from dataclasses import dataclassfrom typing import List, Optionalimport re@dataclassclass TranscriptionSegment:""" Single segment of transcribed audio """ start_time: float# seconds end_time: float# seconds text: str speaker_id: Optional[str] =None confidence: float=1.0def duration(self) ->float:"""Segment duration in seconds"""returnself.end_time -self.start_timedef word_count(self) ->int:"""Count words in segment"""returnlen(self.text.split())def speaking_rate(self) ->float:"""Words per minute"""ifself.duration() ==0:return0return (self.word_count() /self.duration()) *60@dataclassclass AudioTranscription:""" Complete transcription of audio file """ audio_id: str duration: float# total audio duration segments: List[TranscriptionSegment]def total_words(self) ->int:"""Total word count"""returnsum(seg.word_count() for seg inself.segments)def average_speaking_rate(self) ->float:"""Average words per minute across all segments""" total_duration =sum(seg.duration() for seg inself.segments)if total_duration ==0:return0return (self.total_words() / total_duration) *60def get_speaker_segments(self, speaker_id: str) -> List[TranscriptionSegment]:"""Get all segments for a specific speaker"""return [seg for seg inself.segments if seg.speaker_id == speaker_id]def to_srt(self) ->str:""" Export to SRT subtitle format Format: 1 00:00:00,000 --> 00:00:02,500 First subtitle text 2 00:00:02,500 --> 00:00:05,000 Second subtitle text """def format_timestamp(seconds: float) ->str: hours =int(seconds //3600) minutes =int((seconds %3600) //60) secs =int(seconds %60) millis =int((seconds %1) *1000)returnf"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}" srt_output = []for i, seg inenumerate(self.segments, 1): srt_output.append(f"{i}") srt_output.append(f"{format_timestamp(seg.start_time)} --> "f"{format_timestamp(seg.end_time)}" )if seg.speaker_id: srt_output.append(f"[{seg.speaker_id}] {seg.text}")else: srt_output.append(seg.text) srt_output.append("") # Blank line between entriesreturn"\n".join(srt_output)def to_vtt(self) ->str:"""Export to WebVTT format"""def format_timestamp(seconds: float) ->str: hours =int(seconds //3600) minutes =int((seconds %3600) //60) secs = seconds %60returnf"{hours:02d}:{minutes:02d}:{secs:06.3f}" vtt_output = ["WEBVTT", ""]for seg inself.segments: vtt_output.append(f"{format_timestamp(seg.start_time)} --> "f"{format_timestamp(seg.end_time)}" )if seg.speaker_id: vtt_output.append(f"<v {seg.speaker_id}>{seg.text}</v>")else: vtt_output.append(seg.text) vtt_output.append("")return"\n".join(vtt_output)# Example usagetranscription = AudioTranscription( audio_id="interview_001", duration=180.0, # 3 minutes segments=[ TranscriptionSegment( start_time=0.0, end_time=3.5, text="Hello, thank you for joining us today.", speaker_id="SPEAKER_1" ), TranscriptionSegment( start_time=3.8, end_time=6.2, text="Thanks for having me.", speaker_id="SPEAKER_2" ), TranscriptionSegment( start_time=6.5, end_time=12.1, text="Let's start with your background in machine learning.", speaker_id="SPEAKER_1" ) ])print(f"Total words: {transcription.total_words()}")print(f"Average speaking rate: {transcription.average_speaking_rate():.1f} WPM")print("\nSRT format:")print(transcription.to_srt()[:200] +"...")
Total words: 19
Average speaking rate: 99.1 WPM
SRT format:
1
00:00:00,000 --> 00:00:03,500
[SPEAKER_1] Hello, thank you for joining us today.
2
00:00:03,799 --> 00:00:06,200
[SPEAKER_2] Thanks for having me.
3
00:00:06,500 --> 00:00:12,099
[SPEAKER_1] Let's...
17.4.3 Transcription Quality Metrics
import Levenshteinfrom typing import List, Tupleclass TranscriptionQualityMetrics:""" Calculate quality metrics for transcriptions """@staticmethoddef word_error_rate(reference: str, hypothesis: str) ->float:""" Calculate Word Error Rate (WER) WER = (Substitutions + Deletions + Insertions) / Total Words in Reference Standard metric for ASR evaluation """ ref_words = reference.lower().split() hyp_words = hypothesis.lower().split()# Levenshtein distance at word level distance = Levenshtein.distance(ref_words, hyp_words)iflen(ref_words) ==0:return0.0iflen(hyp_words) ==0elsefloat('inf')return distance /len(ref_words)@staticmethoddef character_error_rate(reference: str, hypothesis: str) ->float:""" Calculate Character Error Rate (CER) More fine-grained than WER """ distance = Levenshtein.distance(reference.lower(), hypothesis.lower())iflen(reference) ==0:return0.0iflen(hypothesis) ==0elsefloat('inf')return distance /len(reference)@staticmethoddef calculate_agreement( transcriptions: List[str], use_wer: bool=True ) ->dict:""" Calculate inter-annotator agreement for transcriptions Args: transcriptions: List of transcription strings use_wer: Use WER (True) or CER (False) Returns: dict with mean agreement, pairwise agreements """iflen(transcriptions) <2:return {'mean_agreement': 1.0, 'pairwise': []} metric_func = (TranscriptionQualityMetrics.word_error_rate if use_wer else TranscriptionQualityMetrics.character_error_rate) pairwise_errors = [] n =len(transcriptions)for i inrange(n):for j inrange(i +1, n): error_rate = metric_func(transcriptions[i], transcriptions[j]) agreement =1- error_rate # Convert error to agreement pairwise_errors.append({'transcriber_1': i,'transcriber_2': j,'agreement': max(0, agreement), # Clamp to [0, 1]'error_rate': error_rate }) mean_agreement = np.mean([p['agreement'] for p in pairwise_errors])return {'mean_agreement': mean_agreement,'pairwise': pairwise_errors,'metric': 'WER'if use_wer else'CER' }@staticmethoddef calculate_majority_vote(transcriptions: List[str]) ->str:""" Find consensus transcription using edit distance Returns transcription with minimum total distance to all others """ifnot transcriptions:return""iflen(transcriptions) ==1:return transcriptions[0] min_total_distance =float('inf') consensus = transcriptions[0]for candidate in transcriptions: total_distance =sum( Levenshtein.distance(candidate.lower(), other.lower())for other in transcriptions if other != candidate )if total_distance < min_total_distance: min_total_distance = total_distance consensus = candidatereturn consensus# Example usagereference ="The quick brown fox jumps over the lazy dog"hypothesis1 ="The quick brown fox jumped over the lazy dog"hypothesis2 ="The quick brown fox jumps over a lazy dog"hypothesis3 ="The quick braun fox jumps over the lazy dog"# Calculate WERwer1 = TranscriptionQualityMetrics.word_error_rate(reference, hypothesis1)wer2 = TranscriptionQualityMetrics.word_error_rate(reference, hypothesis2)wer3 = TranscriptionQualityMetrics.word_error_rate(reference, hypothesis3)print(f"WER (hypothesis 1): {wer1:.3f}")print(f"WER (hypothesis 2): {wer2:.3f}")print(f"WER (hypothesis 3): {wer3:.3f}")# Calculate inter-annotator agreementtranscriptions = [reference, hypothesis1, hypothesis2, hypothesis3]agreement = TranscriptionQualityMetrics.calculate_agreement(transcriptions)print(f"\nMean agreement: {agreement['mean_agreement']:.3f}")# Find consensusconsensus = TranscriptionQualityMetrics.calculate_majority_vote(transcriptions)print(f"\nConsensus transcription: {consensus}")
WER (hypothesis 1): 0.111
WER (hypothesis 2): 0.111
WER (hypothesis 3): 0.111
Mean agreement: 0.833
Consensus transcription: The quick brown fox jumps over the lazy dog
Table 18.2: Decision guide for label aggregation methods
Method
Best When
Computational Cost
Min Annotators
Handles Missing Data
Output
Implementation
0
Majority Vote
All annotators roughly equal quality
O(n)
3
Yes
Hard labels
scipy.stats.mode
1
Weighted Vote
Known annotator quality scores
O(n)
3
Yes
Hard labels
Custom
2
Dawid-Skene
Unknown annotator quality, sufficient data
O(n × k × iter)
3
Yes
Soft labels + annotator quality
Custom (shown above)
3
MACE
Bayesian approach preferred
O(n × k × iter)
3
Yes
Soft labels + annotator competence
GitHub: dirko/mace
4
GLAD
Item difficulty varies greatly
O(n × k × iter)
3
Yes
Soft labels + item difficulty
GitHub: welinder
5
Expert Adjudication
High-stakes, low-volume
O(n × expert_time)
1
N/A
Hard labels
Human review
19 Label Studio: Implementation
19.1 Installation & Setup
19.1.1 Docker Installation (Recommended)
# Pull latest Label Studio imagedocker pull heartexlabs/label-studio:latest# Run with persistent storagedocker run -d\--name label-studio \-p 8080:8080 \-v$(pwd)/mydata:/label-studio/data \ heartexlabs/label-studio:latest# Access at http://localhost:8080# Default credentials: Set up on first visit
19.1.2 Python Installation
# Install via pippip install label-studio# Or install with ML backend supportpip install label-studio[ml]# Run serverlabel-studio start# Specify portlabel-studio start --port 8080# With custom data directorylabel-studio start --data-dir ./my-label-studio-data
import requestsimport json# Label Studio API endpointBASE_URL ="http://localhost:8080"API_KEY ="your_api_key_here"# Get from Account & Settingsheaders = {"Authorization": f"Token {API_KEY}","Content-Type": "application/json"}# Create NER projectner_project = {"title": "Named Entity Recognition","description": "Annotate entities in text documents","label_config": ''' <View> <Text name="text" value="$text"/> <Labels name="label" toName="text"> <Label value="PERSON" background="red"/> <Label value="ORG" background="blue"/> <Label value="LOC" background="green"/> </Labels> </View> ''',"sampling": "uniform","show_collab_predictions": False}response = requests.post(f"{BASE_URL}/api/projects", headers=headers, json=ner_project)project_id = response.json()['id']print(f"Created project ID: {project_id}")
19.3 Data Import
19.3.1 Importing Tasks
Code
# Import tasks from JSONtasks = [ {"data": {"text": "Apple Inc. CEO Tim Cook announced new products." } }, {"data": {"text": "Microsoft is headquartered in Redmond, Washington." } }]response = requests.post(f"{BASE_URL}/api/projects/{project_id}/import", headers=headers, json=tasks)print(f"Imported {len(tasks)} tasks")
19.3.2 Import from Files
Code
import pandas as pd# Create datasetdf = pd.DataFrame({'text': ['Example text 1','Example text 2','Example text 3' ],'metadata': [ {'source': 'web'}, {'source': 'pdf'}, {'source': 'api'} ]})# Convert to Label Studio formattasks = []for idx, row in df.iterrows(): tasks.append({'data': {'text': row['text'],'id': idx },'meta': row['metadata'] })# Save and importwithopen('tasks.json', 'w') as f: json.dump(tasks, f)# Import via file uploadfiles = {'file': open('tasks.json', 'rb')}response = requests.post(f"{BASE_URL}/api/projects/{project_id}/import", headers=headers, files=files)
19.4 Workflow Automation
19.4.1 Task Assignment & Routing
Code
class TaskRouter:""" Intelligent task routing based on annotator performance """def__init__(self, api_key: str, base_url: str="http://localhost:8080"):self.api_key = api_keyself.base_url = base_urlself.headers = {"Authorization": f"Token {api_key}","Content-Type": "application/json" }def get_annotator_stats(self, project_id: int) -> pd.DataFrame:"""Get performance stats for all annotators""" response = requests.get(f"{self.base_url}/api/projects/{project_id}/annotators", headers=self.headers ) annotators = response.json() stats = []for ann in annotators: stats.append({'annotator_id': ann['id'],'email': ann['email'],'total_annotations': ann.get('total_annotations', 0),'avg_time': ann.get('avg_lead_time', 0),'accuracy': ann.get('accuracy', 0) # If gold tasks exist })return pd.DataFrame(stats)def assign_tasks_by_skill(self, project_id: int, task_ids: List[int], difficulty_scores: List[float]):""" Assign tasks to annotators based on skill and task difficulty Args: project_id: Label Studio project ID task_ids: List of task IDs to assign difficulty_scores: Difficulty score for each task (0-1) """# Get annotator stats annotators =self.get_annotator_stats(project_id)# Sort annotators by accuracy annotators = annotators.sort_values('accuracy', ascending=False)# Assign difficult tasks to skilled annotatorsfor task_id, difficulty inzip(task_ids, difficulty_scores):if difficulty >0.7:# Hard task - assign to top annotator annotator = annotators.iloc[0]elif difficulty >0.4:# Medium task - assign to middle annotator annotator = annotators.iloc[len(annotators)//2]else:# Easy task - can go to anyone annotator = annotators.sample(1).iloc[0]# Make assignment via APIself._assign_task(project_id, task_id, annotator['annotator_id'])def _assign_task(self, project_id: int, task_id: int, annotator_id: int):"""Assign specific task to annotator""" payload = {'task_id': task_id,'annotator_id': annotator_id } response = requests.post(f"{self.base_url}/api/projects/{project_id}/tasks/{task_id}/assignments", headers=self.headers, json=payload )return response.json()# Example usagerouter = TaskRouter(api_key=API_KEY)# Get list of tasksresponse = requests.get(f"{BASE_URL}/api/projects/{project_id}/tasks", headers=headers)tasks = response.json()# Calculate difficulty (example: based on text length)task_ids = [t['id'] for t in tasks]difficulty_scores = [min(1.0, len(t['data']['text']) /1000) for t in tasks]# Assign tasksrouter.assign_tasks_by_skill(project_id, task_ids, difficulty_scores)
19.4.2 Quality Control Automation
Code
class QualityController:""" Automated quality control for Label Studio """def__init__(self, api_key: str, base_url: str="http://localhost:8080"):self.api_key = api_keyself.base_url = base_urlself.headers = {"Authorization": f"Token {api_key}","Content-Type": "application/json" }def inject_gold_tasks(self, project_id: int, gold_ratio: float=0.1):""" Inject gold standard tasks for quality monitoring Args: project_id: Project ID gold_ratio: Proportion of gold tasks (0-1) """# Get existing tasks response = requests.get(f"{self.base_url}/api/projects/{project_id}/tasks", headers=self.headers ) tasks = response.json()# Randomly select tasks to be gold n_gold =int(len(tasks) * gold_ratio) gold_task_ids = np.random.choice( [t['id'] for t in tasks], size=n_gold, replace=False )# Mark as gold and add ground truthfor task_id in gold_task_ids:# Get task task =next(t for t in tasks if t['id'] == task_id)# Add ground truth annotation (example for NER) gold_annotation = {'result': [ {'value': {'start': 0,'end': 9,'text': 'Apple Inc','labels': ['ORG'] },'from_name': 'label','to_name': 'text','type': 'labels' } ],'ground_truth': True }# Update task requests.post(f"{self.base_url}/api/tasks/{task_id}/annotations", headers=self.headers, json=gold_annotation )def check_annotator_quality(self, project_id: int, min_accuracy: float=0.8) -> List[int]:""" Check annotator quality against gold tasks Args: project_id: Project ID min_accuracy: Minimum acceptable accuracy Returns: List of annotator IDs below threshold """# Get all annotations response = requests.get(f"{self.base_url}/api/projects/{project_id}/annotations", headers=self.headers ) annotations = response.json()# Calculate accuracy per annotator on gold tasks annotator_accuracy = {}for ann in annotations:if ann.get('ground_truth'):continue# Skip ground truth annotations task_id = ann['task'] annotator_id = ann['completed_by']# Get ground truth for this task gt_response = requests.get(f"{self.base_url}/api/tasks/{task_id}/annotations", headers=self.headers, params={'ground_truth': True} ) ground_truth = gt_response.json()ifnot ground_truth:continue# Compare annotation to ground truth is_correct =self._compare_annotations( ann['result'], ground_truth[0]['result'] )if annotator_id notin annotator_accuracy: annotator_accuracy[annotator_id] = [] annotator_accuracy[annotator_id].append(is_correct)# Find annotators below threshold low_performers = []for annotator_id, results in annotator_accuracy.items(): accuracy = np.mean(results)if accuracy < min_accuracy: low_performers.append(annotator_id)return low_performersdef _compare_annotations(self, ann1: List[dict], ann2: List[dict]) ->bool:""" Compare two annotations for equality Simplified - actual implementation depends on annotation type """# For NER, compare entity spans and labelsiflen(ann1) !=len(ann2):returnFalse# Sort by start position ann1_sorted =sorted(ann1, key=lambda x: x['value']['start']) ann2_sorted =sorted(ann2, key=lambda x: x['value']['start'])for a1, a2 inzip(ann1_sorted, ann2_sorted):if (a1['value']['start'] != a2['value']['start'] or a1['value']['end'] != a2['value']['end'] or a1['value']['labels'] != a2['value']['labels']):returnFalsereturnTruedef auto_review_consensus(self, project_id: int, consensus_threshold: int=2):""" Automatically accept annotations with sufficient consensus Args: project_id: Project ID consensus_threshold: Minimum number of agreeing annotators """# Get tasks with multiple annotations response = requests.get(f"{self.base_url}/api/projects/{project_id}/tasks", headers=self.headers, params={'annotations__gt': 1} ) tasks = response.json()for task in tasks: task_id = task['id']# Get all annotations for task ann_response = requests.get(f"{self.base_url}/api/tasks/{task_id}/annotations", headers=self.headers ) annotations = ann_response.json()# Find consensusifself._has_consensus(annotations, consensus_threshold):# Auto-acceptself._accept_task(task_id)def _has_consensus(self, annotations: List[dict], threshold: int) ->bool:"""Check if annotations have consensus"""# Simplified - actual implementation depends on annotation type# For classification, check if threshold annotators agree on labeliflen(annotations) < threshold:returnFalse# Count label frequenciesfrom collections import Counter labels = [ann['result'][0]['value']['choices'][0] for ann in annotations if ann['result']] most_common = Counter(labels).most_common(1)if most_common and most_common[0][1] >= threshold:returnTruereturnFalsedef _accept_task(self, task_id: int):"""Mark task as accepted""" requests.patch(f"{self.base_url}/api/tasks/{task_id}", headers=self.headers, json={'is_labeled': True} )# Example usageqc = QualityController(api_key=API_KEY)# Inject 10% gold tasksqc.inject_gold_tasks(project_id, gold_ratio=0.1)# Check annotator qualitylow_performers = qc.check_annotator_quality(project_id, min_accuracy=0.8)print(f"Annotators below threshold: {low_performers}")# Auto-review with consensusqc.auto_review_consensus(project_id, consensus_threshold=3)
19.5 ML Backend Integration
19.5.1 Pre-annotation with Models
Code
from label_studio_ml.api import LabelStudioMLBase, init_appfrom label_studio_ml.utils import get_single_tag_keysimport torchfrom transformers import pipelineclass NERPredictor(LabelStudioMLBase):""" Custom ML backend for NER pre-annotation """def__init__(self, **kwargs):super(NERPredictor, self).__init__(**kwargs)# Load pre-trained modelself.model = pipeline("ner", model="dslim/bert-base-NER", aggregation_strategy="simple" )# Map model labels to Label Studio labelsself.label_map = {'PER': 'PERSON','ORG': 'ORG','LOC': 'LOC','MISC': 'MISC' }def predict(self, tasks, **kwargs):""" Generate predictions for tasks Args: tasks: List of Label Studio tasks Returns: List of predictions in Label Studio format """ predictions = []for task in tasks: text = task['data']['text']# Run NER model entities =self.model(text)# Convert to Label Studio format results = []for entity in entities:# Map label label =self.label_map.get( entity['entity_group'], entity['entity_group'] ) results.append({'from_name': 'label','to_name': 'text','type': 'labels','value': {'start': entity['start'],'end': entity['end'],'text': entity['word'],'labels': [label] },'score': entity['score'] }) predictions.append({'result': results,'score': np.mean([r['score'] for r in results]) if results else0,'model_version': 'bert-base-NER' })return predictionsdef fit(self, completions, workdir=None, **kwargs):""" Fine-tune model on completed annotations Args: completions: Completed annotations from Label Studio workdir: Working directory for model storage """# Extract training data texts = [] labels = []for completion in completions: text = completion['data']['text'] annotations = completion['annotations'][0]['result']# Convert to token-level labels tokens = text.split() token_labels = ['O'] *len(tokens)for ann in annotations: start = ann['value']['start'] end = ann['value']['end'] label = ann['value']['labels'][0]# Find tokens in span# (Simplified - actual implementation needs word tokenization) token_labels =self._label_tokens( text, tokens, start, end, label ) texts.append(tokens) labels.append(token_labels)# Fine-tune model (pseudo-code - actual implementation varies)# self.model.fine_tune(texts, labels, epochs=3)return {'model_version': 'fine-tuned-v1'}# Run ML backend serverif__name__=='__main__': app = init_app(NERPredictor) app.run(host='0.0.0.0', port=9090)
19.5.2 Active Learning Loop
Code
class ActiveLearningManager:""" Manage active learning loop for efficient annotation """def__init__(self, api_key: str, ml_backend_url: str):self.api_key = api_keyself.ml_backend_url = ml_backend_urlself.base_url ="http://localhost:8080"self.headers = {"Authorization": f"Token {api_key}","Content-Type": "application/json" }def select_uncertain_samples(self, project_id: int, n_samples: int=100, strategy: str='entropy') -> List[int]:""" Select most uncertain samples for annotation Args: project_id: Label Studio project ID n_samples: Number of samples to select strategy: 'entropy', 'margin', or 'random' Returns: List of task IDs to annotate """# Get all unlabeled tasks response = requests.get(f"{self.base_url}/api/projects/{project_id}/tasks", headers=self.headers, params={'is_labeled': False} ) tasks = response.json()ifnot tasks:return []# Get predictions from ML backend task_ids = [t['id'] for t in tasks] predictions_response = requests.post(f"{self.ml_backend_url}/predict", json={'tasks': tasks} ) predictions = predictions_response.json()# Calculate uncertainty scores uncertainties = []for task, pred inzip(tasks, predictions):if strategy =='entropy':# Calculate entropy from prediction scores scores = [r['score'] for r in pred['result']]if scores:# Normalize scores to probabilities probs = np.array(scores) / np.sum(scores) entropy =-np.sum(probs * np.log(probs +1e-10)) uncertainties.append(entropy)else: uncertainties.append(0)elif strategy =='margin':# Margin sampling - difference between top 2 predictions scores =sorted([r['score'] for r in pred['result']], reverse=True)iflen(scores) >=2: margin = scores[0] - scores[1] uncertainties.append(1- margin) # Lower margin = higher uncertaintyelse: uncertainties.append(0)elif strategy =='random': uncertainties.append(np.random.random())# Select top-k uncertain samples uncertain_indices = np.argsort(uncertainties)[-n_samples:] selected_task_ids = [task_ids[i] for i in uncertain_indices]return selected_task_idsdef run_active_learning_cycle(self, project_id: int, n_iterations: int=5, samples_per_iter: int=100):""" Run complete active learning cycle 1. Train model on labeled data 2. Select uncertain samples 3. Send for annotation 4. Wait for completion 5. Repeat """for iteration inrange(n_iterations):print(f"\n=== Active Learning Iteration {iteration +1} ===")# Step 1: Train model on current labeled dataself._trigger_model_training(project_id)# Step 2: Select uncertain samples selected_tasks =self.select_uncertain_samples( project_id, n_samples=samples_per_iter, strategy='entropy' )print(f"Selected {len(selected_tasks)} uncertain tasks")# Step 3: Prioritize these tasks for annotationself._prioritize_tasks(project_id, selected_tasks)# Step 4: Wait for annotations (in practice, this would be async)print("Waiting for annotations...")self._wait_for_annotations(project_id, selected_tasks)# Step 5: Evaluate progress labeled_count =self._get_labeled_count(project_id)print(f"Total labeled tasks: {labeled_count}")def _trigger_model_training(self, project_id: int):"""Trigger model retraining on ML backend"""# Get completed annotations response = requests.get(f"{self.base_url}/api/projects/{project_id}/annotations", headers=self.headers ) completions = response.json()# Send to ML backend for training train_response = requests.post(f"{self.ml_backend_url}/train", json={'completions': completions} )return train_response.json()def _prioritize_tasks(self, project_id: int, task_ids: List[int]):"""Move tasks to top of queue"""for task_id in task_ids: requests.patch(f"{self.base_url}/api/tasks/{task_id}", headers=self.headers, json={'priority': 10} # High priority )def _wait_for_annotations(self, project_id: int, task_ids: List[int]):"""Wait until tasks are annotated (simplified)"""import timewhileTrue:# Check if all tasks are labeled all_labeled =Truefor task_id in task_ids: response = requests.get(f"{self.base_url}/api/tasks/{task_id}", headers=self.headers ) task = response.json()ifnot task.get('is_labeled'): all_labeled =Falsebreakif all_labeled:break time.sleep(60) # Check every minutedef _get_labeled_count(self, project_id: int) ->int:"""Count labeled tasks""" response = requests.get(f"{self.base_url}/api/projects/{project_id}/tasks", headers=self.headers, params={'is_labeled': True} )returnlen(response.json())# Run active learningal_manager = ActiveLearningManager( api_key=API_KEY, ml_backend_url="http://localhost:9090")al_manager.run_active_learning_cycle( project_id=project_id, n_iterations=5, samples_per_iter=100)
19.6 Export & Analysis
19.6.1 Export Annotations
Code
class AnnotationExporter:""" Export annotations in various formats """def__init__(self, api_key: str, base_url: str="http://localhost:8080"):self.api_key = api_keyself.base_url = base_urlself.headers = {"Authorization": f"Token {api_key}","Content-Type": "application/json" }def export_to_coco(self, project_id: int, output_path: str):""" Export bounding box annotations to COCO format Args: project_id: Label Studio project ID output_path: Path to save COCO JSON file """# Get all annotations response = requests.get(f"{self.base_url}/api/projects/{project_id}/export", headers=self.headers, params={'exportType': 'JSON'} ) annotations = response.json()# Convert to COCO format coco = {'images': [],'annotations': [],'categories': [] }# Build category list categories =set()for ann in annotations:for result in ann['annotations'][0]['result']:if result['type'] =='rectanglelabels':for label in result['value']['rectanglelabels']: categories.add(label) coco['categories'] = [ {'id': i, 'name': cat}for i, cat inenumerate(sorted(categories)) ] category_map = {cat['name']: cat['id'] for cat in coco['categories']}# Build images and annotations annotation_id =0for image_id, ann inenumerate(annotations):# Image info coco['images'].append({'id': image_id,'file_name': ann['data'].get('image', ''),'width': ann['data'].get('width', 0),'height': ann['data'].get('height', 0) })# Annotations for this imagefor result in ann['annotations'][0]['result']:if result['type'] =='rectanglelabels': bbox_value = result['value']# Convert percentage to pixels x = bbox_value['x'] * ann['data']['width'] /100 y = bbox_value['y'] * ann['data']['height'] /100 w = bbox_value['width'] * ann['data']['width'] /100 h = bbox_value['height'] * ann['data']['height'] /100for label in bbox_value['rectanglelabels']: coco['annotations'].append({'id': annotation_id,'image_id': image_id,'category_id': category_map[label],'bbox': [x, y, w, h],'area': w * h,'iscrowd': 0 }) annotation_id +=1# Save to filewithopen(output_path, 'w') as f: json.dump(coco, f, indent=2)print(f"Exported {len(coco['images'])} images with {len(coco['annotations'])} annotations")def export_to_conll(self, project_id: int, output_path: str):""" Export NER annotations to CoNLL format Format: token1 O token2 B-PER token3 I-PER token4 O """# Get annotations response = requests.get(f"{self.base_url}/api/projects/{project_id}/export", headers=self.headers, params={'exportType': 'JSON'} ) annotations = response.json()withopen(output_path, 'w') as f:for ann in annotations: text = ann['data']['text'] tokens = text.split() # Simplified tokenization# Initialize all tokens as O (outside) labels = ['O'] *len(tokens)# Get entity spansfor result in ann['annotations'][0]['result']:if result['type'] =='labels': start = result['value']['start'] end = result['value']['end'] entity_type = result['value']['labels'][0]# Find tokens in span (simplified) char_pos =0for i, token inenumerate(tokens): token_start = char_pos token_end = char_pos +len(token)# Check overlap with entity spanif token_start >= start and token_end <= end:if token_start == start: labels[i] =f'B-{entity_type}'else: labels[i] =f'I-{entity_type}' char_pos = token_end +1# +1 for space# Write tokens and labelsfor token, label inzip(tokens, labels): f.write(f"{token}{label}\n") f.write("\n") # Blank line between documentsprint(f"Exported {len(annotations)} documents to CoNLL format")def export_to_yolo(self, project_id: int, output_dir: str):""" Export to YOLO format Creates: - images/ directory with images - labels/ directory with .txt files - data.yaml with class names """import os os.makedirs(f"{output_dir}/images", exist_ok=True) os.makedirs(f"{output_dir}/labels", exist_ok=True)# Get annotations response = requests.get(f"{self.base_url}/api/projects/{project_id}/export", headers=self.headers, params={'exportType': 'JSON'} ) annotations = response.json()# Get class names classes =set()for ann in annotations:for result in ann['annotations'][0]['result']:if result['type'] =='rectanglelabels':for label in result['value']['rectanglelabels']: classes.add(label) classes =sorted(classes) class_to_id = {cls: i for i, cls inenumerate(classes)}# Process each imagefor ann in annotations: image_name = os.path.basename(ann['data']['image']) label_name = os.path.splitext(image_name)[0] +'.txt' img_width = ann['data']['width'] img_height = ann['data']['height']# Write label filewithopen(f"{output_dir}/labels/{label_name}", 'w') as f:for result in ann['annotations'][0]['result']:if result['type'] =='rectanglelabels': bbox_value = result['value']# Convert to YOLO format (normalized center coordinates) x_center = (bbox_value['x'] + bbox_value['width'] /2) /100 y_center = (bbox_value['y'] + bbox_value['height'] /2) /100 width = bbox_value['width'] /100 height = bbox_value['height'] /100for label in bbox_value['rectanglelabels']: class_id = class_to_id[label] f.write(f"{class_id}{x_center}{y_center}{width}{height}\n")# Write data.yamlwithopen(f"{output_dir}/data.yaml", 'w') as f: f.write(f"path: {output_dir}\n") f.write(f"train: images\n") f.write(f"val: images\n") f.write(f"nc: {len(classes)}\n") f.write(f"names: {classes}\n")print(f"Exported to YOLO format in {output_dir}")# Example usageexporter = AnnotationExporter(api_key=API_KEY)# Export to different formatsexporter.export_to_coco(project_id, 'coco_annotations.json')exporter.export_to_conll(project_id, 'ner_annotations.conll')exporter.export_to_yolo(project_id, 'yolo_dataset')
20 Workforce Management
20.1 Compensation & Pricing
20.1.1 Fair Wage Calculator
class FairWageCalculator:""" Calculate fair compensation for annotation tasks """def__init__(self, min_hourly_wage: float=15.0):""" Args: min_hourly_wage: Minimum hourly wage in USD """self.min_hourly_wage = min_hourly_wageself.wage_per_second = min_hourly_wage /3600def calculate_piece_rate(self, task_type: str, pilot_times: List[float], complexity_multiplier: float=1.0) ->dict:""" Calculate fair piece rate based on pilot timing Args: task_type: Type of task (for logging) pilot_times: List of completion times in seconds from pilot complexity_multiplier: Adjustment for task difficulty Returns: Dict with pricing information """ median_time = np.median(pilot_times) p75_time = np.percentile(pilot_times, 75)# Use 75th percentile to account for learning curve# Add 20% buffer for quality work estimated_time = p75_time *1.2* complexity_multiplier# Calculate base rate base_rate = estimated_time *self.wage_per_second# Round up to nearest cent piece_rate = np.ceil(base_rate *100) /100# Calculate expected hourly rate tasks_per_hour =3600/ estimated_time effective_hourly = piece_rate * tasks_per_hourreturn {'task_type': task_type,'piece_rate_usd': piece_rate,'median_time_sec': median_time,'p75_time_sec': p75_time,'estimated_time_sec': estimated_time,'tasks_per_hour': tasks_per_hour,'effective_hourly_usd': effective_hourly,'meets_minimum_wage': effective_hourly >=self.min_hourly_wage }def calculate_tiered_pricing(self, base_rate: float, quality_tiers: dict=None) -> pd.DataFrame:""" Create tiered pricing based on quality Args: base_rate: Base piece rate quality_tiers: Dict of {tier_name: multiplier} Returns: DataFrame with tier pricing """if quality_tiers isNone: quality_tiers = {'Entry Level (< 80% accuracy)': 0.8,'Standard (80-90% accuracy)': 1.0,'Experienced (90-95% accuracy)': 1.2,'Expert (> 95% accuracy)': 1.5 } tiers = []for tier_name, multiplier in quality_tiers.items(): tiers.append({'tier': tier_name,'multiplier': multiplier,'piece_rate': base_rate * multiplier,'hourly_equivalent': base_rate * multiplier *100# Assuming 100 tasks/hr })return pd.DataFrame(tiers)# Example usage with pilot datacalculator = FairWageCalculator(min_hourly_wage=15.0)# Simulate pilot timing data for different tasksnp.random.seed(42)task_types = {'Text Classification': np.random.normal(8, 2, 50), # Mean 8 sec, std 2'NER (per document)': np.random.normal(45, 10, 50),'Bounding Box': np.random.normal(25, 8, 50),'Image Segmentation': np.random.normal(180, 30, 50)}pricing_table = []for task_type, pilot_times in task_types.items(): pricing = calculator.calculate_piece_rate( task_type, pilot_times, complexity_multiplier=1.0 ) pricing_table.append(pricing)pricing_df = pd.DataFrame(pricing_table)print("Fair Wage Pricing:")print(pricing_df)# Create tiered pricing for NER taskner_rate = pricing_df[pricing_df['task_type'] =='NER (per document)']['piece_rate_usd'].values[0]tiered_pricing = calculator.calculate_tiered_pricing(ner_rate)print("\nTiered Pricing for NER:")print(tiered_pricing)
C:\Users\miken\AppData\Local\Temp\ipykernel_38248\1126693969.py:209: SettingWithCopyWarning:
A value is trying to be set on a copy of a slice from a DataFrame.
Try using .loc[row_indexer,col_indexer] = value instead
See the caveats in the documentation: https://pandas.pydata.org/pandas-docs/stable/user_guide/indexing.html#returning-a-view-versus-a-copy
20.3 Payment Processing
20.3.1 Payment Calculator & Tracker
Code
from datetime import datetime, timedeltafrom typing import Dict, Listimport pandas as pdimport numpy as npclass PaymentProcessor:""" Manage worker payments and invoicing """def__init__(self, payment_cycle: str='weekly'):""" Args: payment_cycle: 'weekly', 'biweekly', or 'monthly' """self.payment_cycle = payment_cycleself.payment_ledger = pd.DataFrame()def calculate_worker_payment(self, worker_id: str, annotations: pd.DataFrame, piece_rates: Dict[str, float], bonus_structure: Dict =None) ->dict:""" Calculate payment for worker Args: worker_id: Worker ID annotations: DataFrame with annotation logs piece_rates: Dict of {task_type: rate_per_task} bonus_structure: Optional bonus configuration Returns: Payment details dict """ worker_annotations = annotations[annotations['worker_id'] == worker_id]iflen(worker_annotations) ==0:return {'worker_id': worker_id,'total_tasks': 0,'base_payment': 0,'bonuses': 0,'total_payment': 0,'effective_hourly': 0,'tasks_by_type': {} }# Calculate base payment payment_details = {'tasks_by_type': {}} total_base =0for task_type, group in worker_annotations.groupby('task_type'): n_tasks =len(group) rate = piece_rates.get(task_type, 0) amount = n_tasks * rate payment_details['tasks_by_type'][task_type] = {'count': n_tasks,'rate': rate,'amount': amount } total_base += amount payment_details['base_payment'] = total_base# Calculate bonuses bonuses =0if bonus_structure:# Quality bonusif'quality_bonus'in bonus_structure: gold_tasks = worker_annotations[worker_annotations['is_gold'] ==True]iflen(gold_tasks) >0: accuracy = gold_tasks['gold_correct'].mean()if accuracy >= bonus_structure['quality_bonus']['threshold']: bonus_amount = total_base * bonus_structure['quality_bonus']['percentage'] bonuses += bonus_amount payment_details['quality_bonus'] = {'accuracy': accuracy,'amount': bonus_amount }# Volume bonusif'volume_bonus'in bonus_structure: total_tasks =len(worker_annotations)if total_tasks >= bonus_structure['volume_bonus']['threshold']: bonuses += bonus_structure['volume_bonus']['amount'] payment_details['volume_bonus'] = bonus_structure['volume_bonus']['amount'] payment_details['worker_id'] = worker_id payment_details['total_tasks'] =len(worker_annotations) payment_details['bonuses'] = bonuses payment_details['total_payment'] = total_base + bonuses payment_details['effective_hourly'] = ( (total_base + bonuses) / (worker_annotations['time_seconds'].sum() /3600)if worker_annotations['time_seconds'].sum() >0else0 )return payment_detailsdef generate_payroll(self, annotations: pd.DataFrame, piece_rates: Dict[str, float], period_start: datetime =None, period_end: datetime =None) -> pd.DataFrame:""" Generate payroll for all workers Args: annotations: All annotation logs piece_rates: Piece rates by task type period_start: Start of payment period period_end: End of payment period Returns: DataFrame with payroll details """if period_start isNoneor period_end isNone:# Use last payment cycle period_end = datetime.now()ifself.payment_cycle =='weekly': period_start = period_end - timedelta(days=7)elifself.payment_cycle =='biweekly': period_start = period_end - timedelta(days=14)else: # monthly period_start = period_end - timedelta(days=30)# Filter annotations to period period_annotations = annotations[ (annotations['timestamp'] >= period_start) & (annotations['timestamp'] <= period_end) ]# Bonus structure bonus_structure = {'quality_bonus': {'threshold': 0.95,'percentage': 0.20 },'volume_bonus': {'threshold': 1000,'amount': 50.0 } }# Calculate payments for all workers payroll_list = []for worker_id in period_annotations['worker_id'].unique(): payment =self.calculate_worker_payment( worker_id, period_annotations, piece_rates, bonus_structure )# Flatten the payment dict for DataFrame payroll_row = {'worker_id': payment['worker_id'],'total_tasks': payment['total_tasks'],'base_payment': payment['base_payment'],'bonuses': payment['bonuses'],'total_payment': payment['total_payment'],'effective_hourly': payment['effective_hourly'],'period_start': period_start,'period_end': period_end,'payment_cycle': self.payment_cycle,'payment_status': 'pending' }# Store full details separately for invoice generation payroll_row['_full_details'] = payment payroll_list.append(payroll_row)ifnot payroll_list:# Return empty DataFrame with correct columnsreturn pd.DataFrame(columns=['worker_id', 'total_tasks', 'base_payment', 'bonuses','total_payment', 'effective_hourly', 'period_start','period_end', 'payment_cycle', 'payment_status' ]) payroll_df = pd.DataFrame(payroll_list)return payroll_dfdef generate_invoice(self, worker_id: str, payment_details: dict) ->str:""" Generate invoice for worker Args: worker_id: Worker ID payment_details: Payment calculation dict Returns: Invoice as formatted string """ invoice =f"""╔════════════════════════════════════════════════════════════════╗║ ANNOTATION SERVICES INVOICE ║╚════════════════════════════════════════════════════════════════╝Worker ID: {worker_id}Payment Period: {payment_details.get('period_start', 'N/A')} to {payment_details.get('period_end', 'N/A')}Invoice Date: {datetime.now().strftime('%Y-%m-%d')}────────────────────────────────────────────────────────────────WORK COMPLETED:"""for task_type, details in payment_details.get('tasks_by_type', {}).items(): invoice +=f"""{task_type}: Tasks Completed: {details['count']} Rate per Task: ${details['rate']:.3f} Subtotal: ${details['amount']:.2f}""" invoice +=f"""────────────────────────────────────────────────────────────────BASE PAYMENT: ${payment_details['base_payment']:.2f}"""if payment_details.get('quality_bonus'): invoice +=f"""QUALITY BONUS (Accuracy: {payment_details['quality_bonus']['accuracy']:.1%}): ${payment_details['quality_bonus']['amount']:.2f}"""if payment_details.get('volume_bonus'): invoice +=f"""VOLUME BONUS: ${payment_details['volume_bonus']:.2f}""" invoice +=f"""────────────────────────────────────────────────────────────────TOTAL PAYMENT: ${payment_details['total_payment']:.2f}Effective Hourly Rate: ${payment_details['effective_hourly']:.2f}/hour────────────────────────────────────────────────────────────────Payment will be processed within 5 business days.Questions? Contact: finance@annotationservices.comThank you for your valuable contributions!"""return invoice# Example usage# First, create some sample annotation datanp.random.seed(42)# Create sample tracker metricssample_data = []for worker_id in ['W001', 'W002', 'W003']:for i inrange(150): sample_data.append({'worker_id': worker_id,'task_id': f'T{i:05d}','task_type': np.random.choice(['NER', 'Text Classification', 'Bounding Box']),'time_seconds': np.random.normal(45, 10),'quality_score': None,'is_gold': np.random.random() <0.1,'gold_correct': np.random.random() <0.96,'timestamp': datetime.now() - timedelta(days=np.random.randint(0, 7)) })annotations_df = pd.DataFrame(sample_data)# Create processorprocessor = PaymentProcessor(payment_cycle='weekly')# Define piece ratespiece_rates = {'NER': 0.18,'Text Classification': 0.03,'Bounding Box': 0.10}# Generate payrollpayroll = processor.generate_payroll( annotations_df, piece_rates, period_start=datetime.now() - timedelta(days=7), period_end=datetime.now())print("Payroll Summary:")print(payroll[['worker_id', 'total_tasks', 'base_payment', 'bonuses', 'total_payment', 'effective_hourly']])# Generate invoice for top earneriflen(payroll) >0: top_earner = payroll.nlargest(1, 'total_payment').iloc[0]# Get full payment details from the stored details payment_details = top_earner['_full_details'] payment_details['period_start'] = top_earner['period_start'].strftime('%Y-%m-%d') payment_details['period_end'] = top_earner['period_end'].strftime('%Y-%m-%d') invoice = processor.generate_invoice(top_earner['worker_id'], payment_details)print("\n"+ invoice)else:print("No payroll data for the period")
Payroll Summary:
worker_id total_tasks base_payment bonuses total_payment \
0 W001 150 15.71 0 15.71
1 W002 150 15.95 0 15.95
2 W003 150 15.91 0 15.91
effective_hourly
0 8.345076
1 8.189064
2 8.476884
╔════════════════════════════════════════════════════════════════╗
║ ANNOTATION SERVICES INVOICE ║
╚════════════════════════════════════════════════════════════════╝
Worker ID: W002
Payment Period: 2026-06-14 to 2026-06-21
Invoice Date: 2026-06-21
────────────────────────────────────────────────────────────────
WORK COMPLETED:
Bounding Box:
Tasks Completed: 50
Rate per Task: $0.100
Subtotal: $5.00
NER:
Tasks Completed: 53
Rate per Task: $0.180
Subtotal: $9.54
Text Classification:
Tasks Completed: 47
Rate per Task: $0.030
Subtotal: $1.41
────────────────────────────────────────────────────────────────
BASE PAYMENT: $15.95
────────────────────────────────────────────────────────────────
TOTAL PAYMENT: $15.95
Effective Hourly Rate: $8.19/hour
────────────────────────────────────────────────────────────────
Payment will be processed within 5 business days.
Questions? Contact: finance@annotationservices.com
Thank you for your valuable contributions!