#!/usr/bin/env python3
"""
Wiki -> Brain semantic extraction pipeline.
Parses the SeedVault wiki (~/wiki/) and ingests structured knowledge
into the brain system with tag-based semantic associations.
What this does:
- Parses entity pages: frontmatter, sections, structured data
- Parses concept pages: companion tables, planting schedules, zone viability
- Extracts [[wikilink]] cross-references into entity graph
- Ingests everything as knowledge events with semantic tags
- Builds a test suite to verify retrieval quality
Run: python3 ~/wiki/wiki_brain_ingest.py
"""
import sys
import os
import re
import json
from pathlib import Path
from collections import defaultdict
sys.path.insert(0, str(Path.home() / '.hermes' / 'brain'))
from brain import init_brain, log_event, query_events, get_session_id, set_session_id, _update_synapses
WIKI = Path.home() / 'wiki'
# ---------------------------------------------------------------------------
# Markdown parsing utilities
# ---------------------------------------------------------------------------
def parse_frontmatter(content):
"""Extract YAML frontmatter and body from markdown."""
if not content.startswith('---'):
return {}, content
parts = content.split('---', 2)
if len(parts) < 3:
return {}, content
fm = {}
for line in parts[1].strip().split('\n'):
if ':' not in line:
continue
k, v = line.split(':', 1)
k = k.strip().lower()
v = v.strip()
# Parse inline lists [a, b, c]
if v.startswith('[') and v.endswith(']'):
items = v[1:-1].split(',')
v = []
for item in items:
item = item.strip().strip(chr(39) + chr(34)) # Strip quotes
if item:
v.append(item)
fm[k] = v
return fm, parts[2].strip()
def parse_markdown_table(text, start_marker=None):
"""
Parse a markdown table into list of dicts.
Handles tables like:
| Col1 | Col2 | Col3 |
|------|------|------|
| val1 | val2 | val3 |
If start_marker is provided, only parse the table after that marker line.
Returns list of dicts with column headers as keys.
"""
tables = []
lines = text.split('\n')
if start_marker:
# Find the marker line and start from there
start_idx = 0
for i, line in enumerate(lines):
if start_marker in line:
start_idx = i + 1
break
lines = lines[start_idx:]
headers = None
current_table = []
for line in lines:
line = line.strip()
if not line.startswith('|'):
if headers and current_table:
tables.append((headers, current_table))
headers = None
current_table = []
continue
# Parse row
cells = [c.strip() for c in line.split('|')[1:-1]]
# Detect separator row (--- | ---)
if all(re.match(r'^[-:]+$', c) for c in cells):
continue
if headers is None:
headers = cells
else:
row = {}
for i, h in enumerate(headers):
if i < len(cells):
row[h] = cells[i]
current_table.append(row)
if headers and current_table:
tables.append((headers, current_table))
return tables
def extract_wikilinks(content):
"""Extract [[wikilink]] references from markdown content."""
return re.findall(r'\[\[([^]]+)\]\]', content)
def extract_wikilinks_in_context(content, context_window=200):
"""
Extract wikilinks along with surrounding context text.
Returns list of (link_target, surrounding_text) tuples.
"""
results = []
for match in re.finditer(r'\[\[([^]]+)\]\]', content):
start = max(0, match.start() - context_window)
end = min(len(content), match.end() + context_window)
context = content[start:end].strip()
results.append((match.group(1), context))
return results
def slug_to_display(slug):
"""Convert slug to display name: 'roma-tomatoes' -> 'Roma Tomatoes'"""
return ' '.join(word.capitalize() for word in slug.split('-'))
# ---------------------------------------------------------------------------
# Entity parsing
# ---------------------------------------------------------------------------
def parse_entity(filepath):
"""Parse an entity page and extract all structured data."""
slug = filepath.stem
with open(filepath) as f:
content = f.read()
fm, body = parse_frontmatter(content)
data = {
'slug': slug,
'title': fm.get('title', slug_to_display(slug)),
'type': fm.get('type', 'entity'),
'tags': fm.get('tags', []),
'created': fm.get('created', ''),
'updated': fm.get('updated', ''),
'sources': fm.get('sources', []),
'wikilinks': extract_wikilinks(body),
}
if isinstance(data['tags'], str):
data['tags'] = [data['tags']] if data['tags'] else []
# Clean tags - strip any remaining quotes
if isinstance(data['tags'], list):
data['tags'] = [t.strip(chr(39)+chr(34)) for t in data['tags'] if t.strip(chr(39)+chr(34))]
# Extract key details from the body
# Category
cat_match = re.search(r'\*\*Category:\*\*\s*(.+)', body)
if cat_match:
data['category'] = cat_match.group(1).strip()
# Price
price_match = re.search(r'\*\*Price:\*\*\s*(\$[\d.]+)', body)
if price_match:
data['price'] = price_match.group(1).strip()
# Days to harvest
dt_match = re.search(r'\*\*Days to Harvest:\*\*\s*(.+)', body)
if dt_match:
data['days_to_harvest'] = dt_match.group(1).strip()
# Germination
germ_match = re.search(r'\*\*Germination:\*\*\s*(.+)', body)
if germ_match:
data['germination'] = germ_match.group(1).strip()
# Seed viability
viab_match = re.search(r'\*\*Viability:\*\*\s*(.+)', body)
if viab_match:
data['seed_viability'] = viab_match.group(1).strip()
# Soil pH
ph_match = re.search(r'\*\*pH:\*\*\s*(.+)', body)
if ph_match:
data['soil_ph'] = ph_match.group(1).strip()
# Type (bush/vining/etc)
type_match = re.search(r'\*\*Type:\*\*\s*(.+)', body)
if type_match:
data['growth_type'] = type_match.group(1).strip()
# Variety type
vtype_match = re.search(r'\*\*Variety Type:\*\*\s*(.+)', body)
if vtype_match:
data['variety_type'] = vtype_match.group(1).strip()
# Overview text
overview_match = re.search(r'## Overview\s*\n+(.+?)(?=##|\Z)', body, re.DOTALL)
if overview_match:
data['overview'] = overview_match.group(1).strip()[:200] # Truncate
# Pest & disease
pest_match = re.search(r'## Pest & Disease\s*\n+(.+?)(?=##|\Z)', body, re.DOTALL)
if pest_match:
data['pests'] = pest_match.group(1).strip()[:200]
# Seed saving
saving_match = re.search(r'## Seed Saving\s*\n+(.+?)(?=##|\Z)', body, re.DOTALL)
if saving_match:
data['seed_saving'] = saving_match.group(1).strip()[:200]
return data
# ---------------------------------------------------------------------------
# Concept page parsers
# ---------------------------------------------------------------------------
def parse_companion_planting(filepath):
"""
Parse companion-planting.md into structured relationships.
Returns list of dicts:
{crop, companions, why, category, wikilinks}
"""
if not filepath.exists():
return []
with open(filepath) as f:
content = f.read()
relationships = []
avoid_together = []
classic_combos = []
# Parse companion tables (crop | companions | why)
# Each table is under a "### Category" header
current_category = None
lines = content.split('\n')
for line in lines:
# Track section headers
cat_match = re.match(r'### (.+)', line.strip())
if cat_match:
current_category = cat_match.group(1).strip()
# Parse all tables in the companion pairings section
pairings_section = re.search(
r'## Companion Pairings by Crop(.+?)(?=## Classic|\Z)',
content, re.DOTALL
)
if pairings_section:
tables = parse_markdown_table(pairings_section.group(1))
for headers, rows in tables:
for row in rows:
crop = row.get('Crop', '').strip()
companions = row.get('Good Companions', '').strip()
why = row.get('Why', '').strip()
if not crop:
continue
# Extract wikilinks from crop name
crop_slug = crop
for link in extract_wikilinks(crop):
crop_slug = link
break
relationships.append({
'crop': crop_slug,
'companions': companions,
'why': why,
'category': current_category or 'unknown',
})
# Parse avoid-together table
avoid_section = re.search(
r'## Plants to Avoid Together(.+?)(?=## |\Z)',
content, re.DOTALL
)
if avoid_section:
tables = parse_markdown_table(avoid_section.group(1))
for headers, rows in tables:
for row in rows:
avoid = row.get('Avoid', '').strip()
reason = row.get('Reason', '').strip()
if avoid and ' + ' in avoid:
# Parse "X + Y" pairs
pair = [p.strip() for p in avoid.split('+')]
avoid_together.append({
'crops': pair,
'reason': reason,
})
# Parse classic combinations
combos_section = re.search(
r'## Classic Combinations(.+?)(?=## Plants to Avoid|\Z)',
content, re.DOTALL
)
if combos_section:
combos_text = combos_section.group(1)
# Each combo starts with ###
for combo_match in re.finditer(r'### (.+?)\n+(.+?)(?=###|\Z)', combos_text, re.DOTALL):
combo_name = combo_match.group(1).strip()
combo_body = combo_match.group(2).strip()
combo_links = extract_wikilinks(combo_body)
classic_combos.append({
'name': combo_name,
'description': combo_body.strip()[:200],
'products': combo_links,
})
return {
'relationships': relationships,
'avoid_together': avoid_together,
'classic_combos': classic_combos,
}
def parse_planting_schedules(filepath):
"""Parse planting-schedules.md into structured planting events."""
if not filepath.exists():
return []
with open(filepath) as f:
content = f.read()
schedules = []
season_headers = [
('Spring Planting', 'spring'),
('Late Spring/Early Summer', 'late-spring'),
('Early Summer', 'early-summer'),
('Fall Planting', 'fall'),
]
for i, (pattern, season) in enumerate(season_headers):
header_match = re.search(rf'## {re.escape(pattern)}', content)
if not header_match:
continue
start = header_match.end()
next_idx = len(content)
for j in range(i+1, len(season_headers)):
next_match = re.search(rf'## {re.escape(season_headers[j][0])}', content[start:])
if next_match:
end = start + next_match.start()
if end < next_idx:
next_idx = end
break
section_text = content[start:next_idx]
# Split section into sub-sections
sub_sections = section_text.split('\n## ')
for idx, sub in enumerate(sub_sections):
# Parse tables in this sub-section
tables = parse_markdown_table(sub)
for headers, rows in tables:
# Detect method from column headers (per-table)
header_keys = [h.lower() for h in headers]
method = 'direct-sow'
if any('start' in h and 'indoors' in h for h in header_keys):
method = 'start-indoors'
elif any('transplant' in h for h in header_keys):
method = 'transplant'
elif any('sow' in h for h in header_keys):
method = 'direct-sow'
for row in rows:
crop = None
sowing_time = ''
days = ''
notes = ''
for h, v in row.items():
h_lower = h.lower()
if 'crop' in h_lower:
crop = v.strip()
elif 'sow' in h_lower or 'when' in h_lower or 'start' in h_lower or 'transplant' in h_lower:
sowing_time = v.strip()
elif 'days' in h_lower and 'harvest' in h_lower:
days = v.strip()
elif 'notes' in h_lower:
notes = v.strip()
if not crop or crop.startswith('---'):
continue
crop_slug = crop
for link in extract_wikilinks(crop):
crop_slug = link
break
schedules.append({
'crop': crop_slug,
'season': season,
'method': method,
'sowing_time': sowing_time,
'days_to_harvest': days,
'notes': notes[:200],
})
return schedules
def parse_hardiness_zones(filepath):
"""
Parse hardiness-zones.md into zone viability data.
Returns structured zone data per crop.
"""
if not filepath.exists():
return {'zone_ref': [], 'crop_viability': {}, 'tree_zones': {}}
with open(filepath) as f:
content = f.read()
result = {
'zone_ref': [],
'crop_viability': {}, # crop -> {zone, viability, days}
'tree_zones': {},
}
# Parse zone reference table
zone_ref_section = re.search(
r'### Last Frost Dates.*?\s*\n+(.*?)(?=## Crop Viability|\Z)',
content, re.DOTALL
)
if zone_ref_section:
tables = parse_markdown_table(zone_ref_section.group(1))
for headers, rows in tables:
for row in rows:
result['zone_ref'].append({
'zone': row.get('Zone', ''),
'last_spring_frost': row.get('Avg Last Spring Frost', ''),
'first_fall_frost': row.get('Avg First Fall Frost', ''),
'growing_days': row.get('Growing Days', ''),
})
# Parse crop viability sections
# Zone 3-4 (with parenthetical suffix)
zone34 = re.search(
r'### Zone 3-4\s*\(.*?\)\s*\n+(.*?)(?=### Zone|\Z)',
content, re.DOTALL
)
if zone34:
text = zone34.group(1)
for line in text.split('\n'):
line = line.strip()
if not line.startswith('-'):
continue
# Extract crop slug
links = extract_wikilinks(line)
if not links:
continue
crop = links[0]
days_match = re.search(r'\((\d+-\d+)\s+days\)', line)
days = days_match.group(1) if days_match else ''
if 'ā' in line:
viability = 'viable'
elif 'start indoors' in line.lower() or 'indoors' in line.lower():
viability = 'challenging'
elif 'ā' in line:
viability = 'not-viable'
elif 'Challenging' in line or 'possible' in line.lower():
viability = 'challenging'
else:
viability = 'unknown'
result['crop_viability'][crop] = {
'zone': '3-4',
'viability': viability,
'days': days,
}
# Zone 5-6
zone56 = re.search(
r'### Zone 5-6\s*\(.*?\)\s*\n+(.*?)(?=### Zone|\Z)',
content, re.DOTALL
)
if zone56:
text = zone56.group(1)
# "Full catalog viable except..."
result['zone_summary_5_6'] = text.strip()[:200]
# Zone 7-8
zone78 = re.search(
r'### Zone 7-8\s*\(.*?\)\s*\n+(.*?)(?=### Zone|\Z)',
content, re.DOTALL
)
if zone78:
result['zone_summary_7_8'] = zone78.group(1).strip()[:200]
# Zone 9-10
zone910 = re.search(
r'### Zone 9-10\s*\(.*?\)\s*\n+(.*?)(?=## |\Z)',
content, re.DOTALL
)
if zone910:
result['zone_summary_9_10'] = zone910.group(1).strip()[:200]
# Parse tree zones table
tree_section = re.search(
r'## Tree Seeds ā Zone Requirements\s*\n+(.*?)(?=## |\Z)',
content, re.DOTALL
)
if tree_section:
tables = parse_markdown_table(tree_section.group(1))
for headers, rows in tables:
for row in rows:
tree_name = row.get('Tree', '').strip()
tree_slug = tree_name
for link in extract_wikilinks(tree_name):
tree_slug = link
break
result['tree_zones'][tree_slug] = {
'zones': row.get('Zones', ''),
'notes': row.get('Notes', '').strip()[:200],
}
return result
# ---------------------------------------------------------------------------
# Curriculum parsing
# ---------------------------------------------------------------------------
def parse_curriculum_entity(filepath):
"""Parse a curriculum lesson topic entity page."""
slug = filepath.stem
with open(filepath) as f:
file_content = f.read()
fm, body = parse_frontmatter(file_content)
data = {
'slug': slug,
'title': fm.get('title', slug_to_display(slug)),
'subject': fm.get('subject', ''),
'weeks': fm.get('weeks', []),
'phase': fm.get('phase', ''),
'standard': fm.get('standard', ''),
'tags': fm.get('tags', []),
'wikilinks': extract_wikilinks(body),
'type': 'curriculum_topic',
}
if isinstance(data['tags'], str):
data['tags'] = [data['tags']] if data['tags'] else []
# Clean tags - strip any remaining quotes
if isinstance(data['tags'], list):
data['tags'] = [t.strip(chr(39)+chr(34)) for t in data['tags'] if t.strip(chr(39)+chr(34))]
# Extract description
desc_match = re.search(r'## Description\s*\n+(.+?)(?=##|\Z)', body, re.DOTALL)
if desc_match:
data['description'] = desc_match.group(1).strip()[:300]
# Extract overview
overview_match = re.search(r'## Overview\s*\n+(.+?)(?=##|\Z)', body, re.DOTALL)
if overview_match:
data['overview'] = overview_match.group(1).strip()[:300]
return data
def parse_curriculum_map(filepath):
"""Parse a curriculum map concept page."""
slug = filepath.stem
with open(filepath) as f:
file_content = f.read()
fm, body = parse_frontmatter(file_content)
data = {
'slug': slug,
'title': fm.get('title', slug_to_display(slug)),
'subject': fm.get('subject', ''),
'type': 'curriculum_map',
'tags': fm.get('tags', []),
}
if isinstance(data['tags'], str):
data['tags'] = [data['tags']] if data['tags'] else []
# Clean tags - strip any remaining quotes
if isinstance(data['tags'], list):
data['tags'] = [t.strip(chr(39)+chr(34)) for t in data['tags'] if t.strip(chr(39)+chr(34))]
# Extract all topic wikilinks from the map
all_links = extract_wikilinks(body)
data['topics'] = all_links
# Extract phase/unit headers
phases = re.findall(r'### (.+?)\n+\|', body)
data['phases'] = phases
return data
# ---------------------------------------------------------------------------
# Wiki cross-link graph
# ---------------------------------------------------------------------------
def build_wiki_link_graph():
"""Build a graph of all wiki cross-references."""
graph = defaultdict(set)
for md_file in WIKI.rglob('*.md'):
with open(md_file) as f:
content = f.read()
source = md_file.stem
links = extract_wikilinks(content)
for link in links:
graph[source].add(link)
return dict(graph)
def ingest_curriculum():
"""Parse and ingest all curriculum entity pages and concept maps."""
curriculum_dir = WIKI / 'entities' / 'curriculum'
concepts_dir = WIKI / 'concepts' / 'curriculum'
count = 0
if not curriculum_dir.exists():
print(" No curriculum directory found")
return 0
# Ingest lesson topics by subject
for subject in ['math', 'science', 'ela']:
subject_dir = curriculum_dir / subject
if not subject_dir.exists():
continue
subject_count = 0
for fname in sorted(os.listdir(subject_dir)):
if not fname.endswith('.md'):
continue
filepath = subject_dir / fname
data = parse_curriculum_entity(filepath)
# Normalize weeks to list of ints
weeks = data.get('weeks', [])
if isinstance(weeks, str):
weeks = [int(w) for w in re.findall(r'\d+', weeks)]
elif isinstance(weeks, list):
weeks = [int(w) for w in weeks if isinstance(w, (int, str)) and w]
# Build tags with subject prefix
tags = list(data.get('tags', []))
if subject:
tags.append(f'{subject}_topic')
tags.append('curriculum')
tags.append('grade-2')
# Build context for brain event
context = {
'type': 'curriculum_topic',
'subject': subject,
'title': data.get('title', ''),
'slug': data.get('slug', ''),
'weeks': weeks,
'phase': data.get('phase', ''),
'description': data.get('description', '')[:200],
'wikilinks': data.get('wikilinks', []),
}
# Determine primary week
primary_week = weeks[0] if weeks else 0
event_title = f"Curriculum: {subject.title()} - {data.get('title', '')}"
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'title': event_title, 'section': 'curriculum'},
context=context,
tags=tags,
skip_synapse=True,
)
subject_count += 1
print(f" Ingested {subject_count} {subject} topics")
count += subject_count
# Ingest curriculum maps
if concepts_dir.exists():
map_count = 0
for fname in sorted(os.listdir(concepts_dir)):
if not fname.endswith('.md'):
continue
filepath = concepts_dir / fname
data = parse_curriculum_map(filepath)
tags = list(data.get('tags', []))
tags.append('curriculum')
tags.append('curriculum-map')
context = {
'type': 'curriculum_map',
'subject': data.get('subject', ''),
'title': data.get('title', ''),
'slug': data.get('slug', ''),
'topics': data.get('topics', []),
'phases': data.get('phases', []),
}
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'title': f"Curriculum Map: {data.get('title', '')}", 'section': 'curriculum'},
context=context,
tags=tags,
skip_synapse=True,
)
map_count += 1
print(f" Ingested {map_count} curriculum maps")
count += map_count
return count
# ---------------------------------------------------------------------------
# Brain ingestion
# ---------------------------------------------------------------------------
def purge_existing_wiki():
"""Remove previously ingested wiki events for idempotent reruns."""
init_brain()
events = query_events(event_type='knowledge_ingest', tool='wiki_ingest', limit=10000)
if not events:
return 0
brain_dir = Path.home() / '.hermes' / 'brain'
db_path = brain_dir / 'index' / 'events.db'
event_ids = [e['id'] for e in events]
try:
import sqlite3
db = sqlite3.connect(str(db_path))
# Delete in batches ā FTS5 has auto-delete trigger, no manual cleanup needed
batch_size = 500
for i in range(0, len(event_ids), batch_size):
batch = event_ids[i:i+batch_size]
placeholders = ','.join(['?'] * len(batch))
db.execute(f"DELETE FROM events WHERE id IN ({placeholders})", batch)
db.commit()
db.close()
except Exception as e:
print(f" Warning: purge error: {e}")
return len(event_ids)
def ingest_entities():
"""Parse and ingest all entity pages."""
entities_dir = WIKI / 'entities'
count = 0
for fname in sorted(os.listdir(entities_dir)):
if not fname.endswith('.md'):
continue
filepath = entities_dir / fname
slug = fname.replace('.md', '')
data = parse_entity(filepath)
# Build semantic tags for association
tags = ['seedvault', 'wiki', 'entity', slug]
# Category tag
if data.get('category'):
cat_tag = data['category'].lower().replace(' ', '-')
tags.append(cat_tag)
# Existing frontmatter tags
for tag in data.get('tags', []):
if isinstance(tag, str):
tags.append(tag)
# Ingest as knowledge event
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'slug': slug, 'section': 'entity'},
context={
'source': f'wiki/entities/{slug}.md',
'type': 'seed_product',
'slug': slug,
'title': data.get('title', slug_to_display(slug)),
'category': data.get('category', ''),
'price': data.get('price', ''),
'germination': data.get('germination', ''),
'seed_viability': data.get('seed_viability', ''),
'soil_ph': data.get('soil_ph', ''),
'days_to_harvest': data.get('days_to_harvest', ''),
'growth_type': data.get('growth_type', ''),
'variety_type': data.get('variety_type', ''),
'overview': data.get('overview', ''),
'pests': data.get('pests', ''),
'seed_saving': data.get('seed_saving', ''),
'wikilinks': data.get('wikilinks', []),
},
tags=tags,
skip_synapse=True
)
count += 1
return count
def ingest_companion_planting():
"""Parse and ingest companion planting relationships."""
filepath = WIKI / 'concepts' / 'companion-planting.md'
data = parse_companion_planting(filepath)
count = 0
# Ingest companion relationships
for rel in data['relationships']:
crop = rel['crop']
companions = rel['companions']
why = rel['why']
category = rel['category']
# Parse individual companion crops from the list
companion_list = [c.strip().strip('*') for c in re.split(r'[, ]+', companions) if c.strip().strip('*')]
tags = ['seedvault', 'wiki', 'companion', crop]
# Add companion crop slugs as tags for association
for comp in companion_list:
comp_tag = comp.lower().replace(' ', '-')
if comp_tag not in tags:
tags.append(comp_tag)
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'slug': crop, 'section': 'companion_relationship'},
context={
'source': 'wiki/concepts/companion-planting.md',
'type': 'companion_relationship',
'crop': crop,
'companions': companions,
'companion_list': companion_list,
'why': why,
'category': category,
},
tags=tags,
skip_synapse=True
)
count += 1
# Ingest avoid-together pairs
for avoid in data['avoid_together']:
crops = avoid['crops']
reason = avoid['reason']
tags = ['seedvault', 'wiki', 'avoid-together'] + [
c.lower().replace(' ', '-') for c in crops
]
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'section': 'avoid_together'},
context={
'source': 'wiki/concepts/companion-planting.md',
'type': 'avoid_together',
'crops': crops,
'reason': reason,
},
tags=tags,
skip_synapse=True
)
count += 1
# Ingest classic combinations
for combo in data['classic_combos']:
tags = ['seedvault', 'wiki', 'companion', 'combination',
combo['name'].lower().replace(' ', '-')] + combo['products']
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'section': 'classic_combination'},
context={
'source': 'wiki/concepts/companion-planting.md',
'type': 'garden_combination',
'name': combo['name'],
'description': combo['description'],
'products': combo['products'],
},
tags=tags,
skip_synapse=True
)
count += 1
return count
def ingest_planting_schedules():
"""Parse and ingest planting schedule data."""
filepath = WIKI / 'concepts' / 'planting-schedules.md'
schedules = parse_planting_schedules(filepath)
count = 0
for sched in schedules:
crop = sched['crop']
season = sched['season']
method = sched['method']
tags = ['seedvault', 'wiki', 'planting', season, crop, method]
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'crop': crop, 'section': 'planting_schedule'},
context={
'source': 'wiki/concepts/planting-schedules.md',
'type': 'planting_schedule',
'crop': crop,
'season': season,
'method': method,
'sowing_time': sched.get('sowing_time', ''),
'days_to_harvest': sched.get('days_to_harvest', ''),
'notes': sched.get('notes', ''),
},
tags=tags,
skip_synapse=True
)
count += 1
return count
def ingest_hardiness_zones():
"""Parse and ingest hardiness zone data."""
filepath = WIKI / 'concepts' / 'hardiness-zones.md'
zone_data = parse_hardiness_zones(filepath)
count = 0
# Ingest crop zone viability
for crop, info in zone_data['crop_viability'].items():
# Parse zone numbers into individual tags (e.g., "3-4" -> ["zone-3", "zone-4"])
zone_nums = info['zone'].replace('-', ' ')
zone_tags = [f'zone-{n}' for n in zone_nums.strip().split()]
tags = ['seedvault', 'wiki', 'zone', crop] + zone_tags
if info['viability'] == 'viable':
tags.append('viable')
elif info['viability'] == 'not-viable':
tags.append('not-viable')
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'crop': crop, 'section': 'zone_viability'},
context={
'source': 'wiki/concepts/hardiness-zones.md',
'type': 'zone_viability',
'crop': crop,
'zone': info['zone'],
'viability': info['viability'],
'days': info.get('days', ''),
},
tags=tags,
skip_synapse=True
)
count += 1
# Ingest tree zone requirements
for tree, info in zone_data['tree_zones'].items():
zones = info['zones']
# Parse zone numbers into individual tags: "3-4" -> ["zone-3", "zone-4"]
zone_parts = zones.replace('-', ' ').split()
zone_tags = [f'zone-{z}' for z in zone_parts]
tags = ['seedvault', 'wiki', 'zone', 'tree', tree] + zone_tags
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'crop': tree, 'section': 'tree_zones'},
context={
'source': 'wiki/concepts/hardiness-zones.md',
'type': 'zone_requirement',
'crop': tree,
'zones': zones,
'notes': info.get('notes', ''),
},
tags=tags,
skip_synapse=True
)
count += 1
return count
def ingest_wiki_links():
"""Ingest wiki cross-link graph as semantic associations."""
graph = build_wiki_link_graph()
count = 0
for source, targets in graph.items():
if not targets:
continue
tags = ['seedvault', 'wiki', 'wikilink', source] + list(targets)
log_event(
event_type='knowledge_ingest',
tool='wiki_ingest',
args={'source': source, 'section': 'wikilink_graph'},
context={
'source': f'wiki_link_graph:{source}',
'type': 'wiki_link',
'from': source,
'to': list(targets),
},
tags=tags,
skip_synapse=True
)
count += 1
return count
# ---------------------------------------------------------------------------
# Test queries
# ---------------------------------------------------------------------------
def test_queries():
"""Run test queries against ingested knowledge and compare to wiki ground truth."""
print("\n" + "=" * 60)
print("QUERY TESTS")
print("=" * 60)
events = query_events(event_type='knowledge_ingest', tool='wiki_ingest', limit=10000)
# Helper: find events by crop
def find_by_crop(crop_slug, event_type=None):
results = []
for e in events:
ctx = e.get('context', {})
if not isinstance(ctx, dict):
continue
# Check crop field (used by companion/planting/zone events)
if ctx.get('crop') == crop_slug:
if event_type is None or ctx.get('type') == event_type:
results.append(ctx)
# Also check slug field (used by seed_product events)
elif ctx.get('slug') == crop_slug:
if event_type is None or ctx.get('type') == event_type:
results.append(ctx)
# Also check wikilinks and product lists
for field in ['wikilinks', 'products', 'companion_list', 'to']:
if isinstance(ctx.get(field), list) and crop_slug in ctx[field]:
if event_type is None or ctx.get('type') == event_type:
results.append(ctx)
return results
# Helper: find events by tag
def find_by_tag(tag):
results = []
for e in events:
if tag in (e.get('tags') or []):
results.append(e.get('context', {}))
return results
# TEST 1: What grows well with tomatoes?
print("\n--- Q1: What grows well with Roma Tomatoes? ---")
tomato_companions = find_by_crop('roma-tomatoes', 'companion_relationship')
if tomato_companions:
for c in tomato_companions:
print(f" Companions: {c.get('companions', '')}")
print(f" Why: {c.get('why', '')}")
else:
# Fallback: check classic combos containing roma-tomatoes
combos = find_by_tag('roma-tomatoes')
combo_hits = [c for c in combos if c.get('type') == 'garden_combination']
if combo_hits:
for c in combo_hits:
print(f" Combo '{c.get('name')}': products={c.get('products')}")
else:
print(" NO MATCH")
# Verify against wiki
tomato_path = WIKI / 'entities' / 'roma-tomatoes.md'
if tomato_path.exists():
with open(tomato_path) as f:
tc = f.read()
# The wiki companion data is in the concept page, not the entity page
# Ground truth from companion-planting.md:
print(" Wiki answer: Basil, carrots, peppers, lettuce (Basil repels tomato hornworms)")
# TEST 2: What can grow in zone 3?
print("\n--- Q2: What crops are viable in Zone 3? ---")
zone3_viable = find_by_tag('zone-3')
zone3_viable = [c for c in zone3_viable if c.get('type') == 'zone_viability' and c.get('viability') == 'viable']
zone3_crops = list(set(c.get('crop', '') for c in zone3_viable if c.get('crop')))
print(f" Brain found {len(zone3_crops)} viable crops:")
for crop in sorted(zone3_crops):
print(f" - {crop}")
known_zone3 = [
'snowball-turnips', 'butter-crunch-butterhead-lettuce',
'provider-green-bush-beans', 'red-russian-kale',
'bulls-blood-beets', 'scarlett-nantes-carrot',
'sugar-lace-snap-peas'
]
missing = set(known_zone3) - set(zone3_crops)
extra = set(zone3_crops) - set(known_zone3)
if missing:
print(f" MISSING: {missing}")
if extra:
print(f" EXTRA: {extra}")
if not missing and not extra:
print(" EXACT MATCH ā")
# TEST 3: Seed storage for carrots
print("\n--- Q3: Carrot seed viability ---")
carrot = find_by_crop('scarlett-nantes-carrot', 'seed_product')
if carrot:
viab = carrot[0].get('seed_viability', '')
slug = carrot[0].get('slug', '')
if slug:
print(f" ā Found slug={slug}, viability: {viab}")
else:
print(f" ā Found, viability: {viab} (slug not set)")
else:
print(" ā NO MATCH")
# TEST 4: What season to plant basil?
print("\n--- Q4: What season/method to plant basil? ---")
basil_planting = find_by_crop('large-leaf-basil', 'planting_schedule')
if basil_planting:
for bp in basil_planting:
print(f" Brain: season={bp.get('season')}, method={bp.get('method')}, "
f"sow={bp.get('sowing_time')}, days={bp.get('days_to_harvest')}")
else:
print(" NO MATCH")
print(" Wiki answer: spring, start-indoors, 3-4 weeks before last frost")
# TEST 5: Product count
print("\n--- Q5: How many seed products does the brain know? ---")
products = [e for e in events if isinstance(e.get('context', {}), dict)
and e['context'].get('type') == 'seed_product']
print(f" Brain knows: {len(products)} seed products")
print(f" Wiki has: 27 entity pages + 1 starter pack + 2 concept pages = ~30")
# TEST 6: FTS5 search
print("\n--- Q6: FTS5 full-text search 'roma' ---")
nquery_results = query_events(full_text_search='roma', limit=10)
if nquery_results:
print(f" Found {len(nquery_results)} results:")
types = {}
for e in nquery_results:
ctx = e.get('context_json', e.get('context', {}))
if isinstance(ctx, dict):
t = ctx.get('type', 'unknown')
types[t] = types.get(t, 0) + 1
title = ctx.get('title', ctx.get('crop', ctx.get('source', '')))
print(f" - {t}: {title}")
else:
print(f" - (raw ctx)")
print(f" By type: {types}")
else:
print(" NO RESULTS")
# TEST 7: Synapse associations
print("\n--- Q7: Synapse semantic associations ---")
synapse_path = Path.home() / '.hermes' / 'brain' / 'synapses' / 'graph.json'
if synapse_path.exists():
with open(synapse_path) as f:
graph = json.load(f)
nodes = graph.get('nodes', {})
edges = graph.get('edges', {})
total_edges = sum(len(v) for v in edges.values())
print(f" Synapse nodes: {len(nodes)}")
print(f" Synapse edges: {total_edges}")
# Check tag-based connections (events sharing tags)
# Count edges by tag type
tag_edge_types = {}
for src, targets in edges.items():
if not src.startswith('evt_'):
continue
for tgt, meta in targets.items():
if isinstance(meta, dict):
etype = meta.get('type', 'unknown')
if etype.endswith('_related'):
tag = etype.replace('_related', '')
tag_edge_types[tag] = tag_edge_types.get(tag, 0) + 1
# Show top semantic tag connections (excluding seedvault/wiki which are too broad)
interesting = {k: v for k, v in tag_edge_types.items()
if k not in ('seedvault', 'wiki')}
if interesting:
print(f" Semantic tag connections (top 10):")
for tag, count in sorted(interesting.items(), key=lambda x: -x[1])[:10]:
print(f" {tag}: {count} edges")
else:
print(" (no interesting tag connections beyond seedvault/wiki)")
# Check specific association: crop-specific events share meaningful tags
roma_companion = tag_edge_types.get('roma-tomatoes', 0)
basil_companion = tag_edge_types.get('basil', 0)
print(f" roma-tomatoes tag edges: {roma_companion}")
print(f" basil tag edges: {basil_companion}")
# Pass/fail: meaningful crop tags should have edges, broad tags should not dominate
has_crop_edges = roma_companion > 0 or basil_companion > 0
total_tag_edges = sum(tag_edge_types.values())
broad_count = tag_edge_types.get('seedvault', 0) + tag_edge_types.get('wiki', 0)
broad_dominates = total_tag_edges > 0 and broad_count / total_tag_edges > 0.5
if has_crop_edges and not broad_dominates:
print(f" ā Semantic connections working (crop edges present, broad tags filtered)")
elif has_crop_edges:
print(f" ā Crop edges present but broad tags still dominate ({broad_count}/{total_tag_edges})")
else:
print(f" ā No meaningful crop-specific connections")
else:
print(" NO SYNAPSE GRAPH")
# TEST 8: Avoid-together query
print("\n--- Q8: Plants to avoid growing together ---")
avoids = find_by_tag('avoid-together')
if avoids:
for a in avoids:
print(f" Don't grow together: {a.get('crops')}")
print(f" Reason: {a.get('reason', '')}")
else:
print(" NO AVOID DATA")
print(" Wiki: Tomatoes+Brassicas, Beans+Onions, Squash+Potatoes, Corn+Tomatoes")
# TEST 9: Classic combinations
print("\n--- Q9: Classic garden combinations ---")
combos = find_by_tag('combination')
combo_hits = [c for c in combos if c.get('type') == 'garden_combination']
if combo_hits:
for c in combo_hits:
print(f" {c.get('name')}: {c.get('products')}")
else:
print(" NO COMBINATION DATA")
# TEST 10: Curriculum query - what did we teach in a specific week?
print("\n--- Q10: Math topics taught in Week 5 ---")
math_week5 = find_by_tag('math_topic')
week5_hits = []
for m in math_week5:
ctx = m.get('context', {})
weeks = ctx.get('weeks', [])
if 5 in weeks:
week5_hits.append(ctx.get('title', ctx.get('slug', '?')))
if week5_hits:
for h in week5_hits:
print(f" ā {h}")
else:
print(" NO WEEK 5 MATH DATA")
print(" Wiki: Subtraction Facts to 20")
# TEST 11: Curriculum query - what ELA topics cover phonics?
print("\n--- Q11: ELA topics covering phonics patterns ---")
phonics_topics = find_by_tag('phonics')
if phonics_topics:
for p in phonics_topics:
ctx = p.get('context', {})
title = ctx.get('title', '?')
weeks = ctx.get('weeks', [])
phase = ctx.get('phase', '')
print(f" ā {title} (Weeks {weeks}, {phase})")
else:
print(" NO PHONICS DATA")
# TEST 12: Curriculum map query
print("\n--- Q12: Curriculum map overview ---")
maps = find_by_tag('curriculum-map')
if maps:
for m in maps:
ctx = m.get('context', {})
title = ctx.get('title', '?')
phases = ctx.get('phases', [])
subject = ctx.get('subject', '')
print(f" ā {title}: {subject.title()}, {len(phases)} phases")
else:
print(" NO CURRICULUM MAPS")
# TEST 13: Science unit query
print("\n--- Q13: Science topics in 'Life Science' unit ---")
life_sci = find_by_tag('ecosystems')
if not life_sci:
life_sci = find_by_tag('plants')
if life_sci:
for ls in life_sci:
ctx = ls.get('context', {})
title = ctx.get('title', '?')
weeks = ctx.get('weeks', [])
phase = ctx.get('phase', '')
print(f" ā {title} (Weeks {weeks})")
else:
print(" NO LIFE SCIENCE DATA")
print("\n" + "=" * 60)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
print("=" * 60)
print("WIKI -> BRAIN SEMANTIC EXTRACTION")
print("=" * 60)
init_brain()
# Pre-ingestion count
pre_count = len(query_events(event_type='knowledge_ingest', tool='wiki_ingest', limit=10000))
print(f"\nPre-ingestion wiki events: {pre_count}")
# Purge old data for clean run
if pre_count > 0:
purged = purge_existing_wiki()
print(f"Purged {purged} existing wiki events")
# Reset synapse graph to avoid stale data
synapse_path = Path.home() / '.hermes' / 'brain' / 'synapses' / 'graph.json'
if synapse_path.exists():
print(f"\nResetting synapse graph...")
synapse_path.write_text('{"nodes": {}, "edges": {}}')
# 1. Ingest entities
print("\n[1/6] Parsing and ingesting entity pages...")
entity_count = ingest_entities()
print(f" ā Ingested {entity_count} entity pages")
# 2. Ingest companion planting
print("\n[2/6] Parsing and ingesting companion planting...")
companion_count = ingest_companion_planting()
print(f" ā Ingested {companion_count} companion relationships")
# 3. Ingest planting schedules
print("\n[3/6] Parsing and ingesting planting schedules...")
planting_count = ingest_planting_schedules()
print(f" ā Ingested {planting_count} planting schedule entries")
# 4. Ingest hardiness zones
print("\n[4/6] Parsing and ingesting hardiness zones...")
zone_count = ingest_hardiness_zones()
print(f" ā Ingested {zone_count} zone entries")
# 5. Ingest curriculum topics
print("\n[5/6] Parsing and ingesting curriculum topics...")
curriculum_count = ingest_curriculum()
print(f" ā Ingested {curriculum_count} curriculum entries")
# 6. Ingest wiki cross-links
print("\n[6/6] Building and ingesting wiki cross-link graph...")
link_count = ingest_wiki_links()
print(f" ā Ingested {link_count} wiki link associations")
# Run synapse update in single pass over all ingested events
print("\nBuilding synapse graph (single pass over all events)...")
wiki_events = query_events(event_type='knowledge_ingest', tool='wiki_ingest', limit=10000)
for event in wiki_events:
_update_synapses(event)
print(f" ā Synapse graph built ({len(wiki_events)} events processed)")
# Post-ingestion stats
total = len(query_events(event_type='knowledge_ingest', tool='wiki_ingest', limit=10000))
print(f"\n{'=' * 60}")
print(f"TOTAL WIKI EVENTS INGESTED: {total}")
# Breakdown by type
all_events = query_events(event_type='knowledge_ingest', tool='wiki_ingest', limit=10000)
type_counts = {}
for e in all_events:
ctx = e.get('context', {})
if isinstance(ctx, dict):
t = ctx.get('type', 'unknown')
type_counts[t] = type_counts.get(t, 0) + 1
print("\nBy type:")
for t, c in sorted(type_counts.items(), key=lambda x: -x[1]):
print(f" {t}: {c}")
# Breakdown by tag frequency
tag_counts = defaultdict(int)
for e in all_events:
for tag in (e.get('tags') or []):
tag_counts[tag] += 1
print("\nTop tags:")
for tag, c in sorted(tag_counts.items(), key=lambda x: -x[1])[:15]:
print(f" {tag}: {c}")
# Run test queries
test_queries()
print("\nā Semantic extraction pipeline complete.")
if __name__ == '__main__':
main()