Spaces:
Sleeping
Sleeping
File size: 4,958 Bytes
c2b7a7b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | """Source adapters that load structured data into the kiosk catalog."""
from __future__ import annotations
import csv
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional
from .utils import canonicalize_name
@dataclass
class EntityDefinition:
"""Specification for registering an entity in the data catalog."""
name: str
records: List[Dict[str, Any]]
key_field: Optional[str] = None
origin: Optional[str] = None
normalizer: Optional[Callable[[str], str]] = None
@dataclass
class SourceResult:
"""Payload returned by a data source."""
entities: List[EntityDefinition] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
class DataSource(ABC):
"""Interface for ingesting structured data into the catalog."""
def __init__(self, name: str) -> None:
self.name = name
@abstractmethod
def load(self) -> SourceResult:
raise NotImplementedError
class CSVSource(DataSource):
"""Loads a CSV file into an entity definition."""
def __init__(
self,
name: str,
path: Path,
entity_name: str,
*,
key_field: Optional[str] = None,
normalizer: Optional[Callable[[str], str]] = None,
) -> None:
super().__init__(name)
self.path = path
self.entity_name = entity_name
self.key_field = key_field
self.normalizer = normalizer
def load(self) -> SourceResult:
if not self.path.exists():
return SourceResult()
records = self._read_csv(self.path)
entity = EntityDefinition(
name=self.entity_name,
records=records,
key_field=self.key_field,
origin=str(self.path),
normalizer=self.normalizer,
)
return SourceResult(entities=[entity])
@staticmethod
def _read_csv(path: Path) -> List[Dict[str, Any]]:
with path.open(newline="", encoding="utf-8-sig") as handle:
reader = csv.DictReader(handle)
return [dict(row) for row in reader]
class FeedListSource(DataSource):
"""Loads newline-delimited feed URLs into catalog metadata."""
def __init__(self, name: str, path: Path, metadata_key: str) -> None:
super().__init__(name)
self.path = path
self.metadata_key = metadata_key
def load(self) -> SourceResult:
if not self.path.exists():
return SourceResult()
urls = [
line.strip()
for line in self.path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
return SourceResult(metadata={self.metadata_key: {"urls": urls}})
def default_sources(base_dir: Path, *, name_normalizer: Optional[Callable[[str], str]] = None) -> List[DataSource]:
"""
Produce the default set of data sources used by the backend.
Additional sources (e.g., TA office hours) can be appended to this list
without modifying the rest of the pipeline.
"""
base_dir = base_dir.resolve()
normalizer = name_normalizer or canonicalize_name
sources: List[DataSource] = [
CSVSource(
name="faculty_roster",
path=base_dir / "faculty_2.csv",
entity_name="faculty",
key_field="Name",
normalizer=normalizer,
),
CSVSource(
name="faculty_offices",
path=base_dir / "Faculty.csv",
entity_name="faculty_offices",
key_field="Assignee Name",
normalizer=normalizer,
),
CSVSource(
name="staff_roster",
path=base_dir / "staff.csv",
entity_name="staff",
key_field="Name",
normalizer=normalizer,
),
CSVSource(
name="students_roster",
path=base_dir / "students.csv",
entity_name="students",
key_field="Name",
normalizer=normalizer,
),
CSVSource(
name="office_hours",
path=base_dir / "CS Office Hours Room Reservations.csv",
entity_name="office_hours",
key_field="Course Name",
normalizer=normalizer,
),
CSVSource(
name="centers_catalog",
path=base_dir / "centers.csv",
entity_name="centers",
key_field="Name",
normalizer=normalizer,
),
CSVSource(
name="mudd_seating",
path=base_dir / "Mudd Seating Sample.csv",
entity_name="mudd_seating",
key_field="Student/Visitor",
normalizer=normalizer,
),
FeedListSource(
name="event_feeds",
path=base_dir / "feed.txt",
metadata_key="event_feeds",
),
]
return sources
|