diff --git a/.github/workflows/umi-v2.yml b/.github/workflows/umi-v2.yml index d79d502e..cea04d98 100644 --- a/.github/workflows/umi-v2.yml +++ b/.github/workflows/umi-v2.yml @@ -11,9 +11,9 @@ jobs: coordinated: name: UMI v2 # Update the workflow and runner revision together. - uses: KartikP/brainscore-unified/.github/workflows/offline-tests.yml@a95ad140a226cf5dd56ca9559e2504214078f2c1 + uses: KartikP/brainscore-unified/.github/workflows/offline-tests.yml@b8880276242296c6857363777b4bd12a02ab5213 with: candidate: core candidate_repository: ${{ github.event.pull_request.head.repo.full_name || github.repository }} candidate_revision: ${{ github.event.pull_request.head.sha || github.sha }} - unified_revision: a95ad140a226cf5dd56ca9559e2504214078f2c1 + unified_revision: b8880276242296c6857363777b4bd12a02ab5213 diff --git a/brainscore_core/behavioral.py b/brainscore_core/behavioral.py index 71d68a04..9bf9fa24 100644 --- a/brainscore_core/behavioral.py +++ b/brainscore_core/behavioral.py @@ -17,11 +17,66 @@ def __init__(self, owner) -> None: self.task_context: Optional[TaskContext] = None self.readout_classifier: Any = None self.use_generation_for_task = False + # The TaskContext object the current configuration came from, so a + # repeated send of the same task keeps the fitted readout. + self.configured_from: Optional[TaskContext] = None + self._configured_as: Optional[set] = None # fingerprints of the fitted configuration + self._stale = False @classmethod def requires_behavioral_readout(cls, task_type: str) -> bool: return task_type in cls.BEHAVIORAL_TASK_TYPES + @staticmethod + def _fingerprint(task_context: TaskContext) -> tuple: + """Everything the fitted readout depends on: task, fitting rows and labels, generation settings.""" + fitting = task_context.fitting_stimuli + rows = labels = None + columns = getattr(fitting, 'columns', None) + if columns is not None: + try: + if 'stimulus_id' in columns: + rows = tuple(fitting['stimulus_id']) + for column in ('image_label', 'label'): + if column in columns: + labels = tuple(fitting[column]) + break + except (TypeError, KeyError, AttributeError): + rows = labels = None # not a table; identity alone must do + label_set = task_context.label_set + return (task_context.task_type, id(fitting), rows, labels, task_context.instruction, + tuple(label_set) if label_set is not None else None, + task_context.prefer_path or 'auto') + + def is_configured_for(self, task_context: TaskContext) -> bool: + """Whether ``task_context`` is the task already configured, unchanged since. + + True for the object the configuration came from and for the context + this component holds (the same task after input projection), as long + as its task, fitting table, labels and generation settings are the + ones fitted. TaskContext is mutable: a caller that swaps the fitting + table or relabels it and calls start_task again gets a new fit. + """ + if self._stale or self.task_context is None or not self._configured_as: + return False + if task_context is not self.configured_from and task_context is not self.task_context: + return False + return self._fingerprint(task_context) in self._configured_as + + def mark_configured(self, task_context: Optional[TaskContext]) -> None: + """Remember the configuration just fitted, as the caller's context and as the one held.""" + self.configured_from = task_context + self._configured_as = None + if task_context is not None: + self._configured_as = {self._fingerprint(task_context)} + if self.task_context is not None and self.task_context is not task_context: + self._configured_as.add(self._fingerprint(self.task_context)) + self._stale = False + + def invalidate(self) -> None: + """The model changed (a state change was applied or undone): fit again next time.""" + self._stale = True + def start_task(self, task_context_or_task, fitting_stimuli=None, **kwargs) -> None: del kwargs @@ -108,6 +163,9 @@ def reset(self) -> None: self.task_context = None self.readout_classifier = None self.use_generation_for_task = False + self.configured_from = None + self._configured_as = None + self._stale = False def fit_behavioral_readout(self, fitting_stimuli) -> None: """Extract readout-layer features and fit a probabilities classifier.""" @@ -145,29 +203,29 @@ def extract_labels(stimuli): def extract_behavioral_features(self, stimuli): """Run the model once at the behavioral readout layer.""" owner = self.owner - saved_layer = owner._recording_layer - saved_layers = owner._recording_layers - saved_regions = owner._recording_regions - saved_multi = owner._is_multi_region - saved_composite = owner._composite_recording + saved_layer = owner._recorder.recording_layer + saved_layers = owner._recorder.recording_layers + saved_regions = owner._recorder.recording_regions + saved_multi = owner._recorder.is_multi_region + saved_composite = owner._recorder.composite_recording saved_classifier = self.readout_classifier - owner._recording_layer = owner._behavioral_readout_layer - owner._recording_layers = ( + owner._recorder.recording_layer = owner._behavioral_readout_layer + owner._recorder.recording_layers = ( [owner._behavioral_readout_layer] if owner._behavioral_readout_layer else [] ) - owner._recording_regions = [] - owner._is_multi_region = False - owner._composite_recording = False + owner._recorder.recording_regions = [] + owner._recorder.is_multi_region = False + owner._recorder.composite_recording = False self.readout_classifier = None try: features = owner.process(stimuli) finally: - owner._recording_layer = saved_layer - owner._recording_layers = saved_layers - owner._recording_regions = saved_regions - owner._is_multi_region = saved_multi - owner._composite_recording = saved_composite + owner._recorder.recording_layer = saved_layer + owner._recorder.recording_layers = saved_layers + owner._recorder.recording_regions = saved_regions + owner._recorder.is_multi_region = saved_multi + owner._recorder.composite_recording = saved_composite self.readout_classifier = saved_classifier if 'presentation' in features.dims and 'neuroid' in features.dims: diff --git a/brainscore_core/brainscore_model.py b/brainscore_core/brainscore_model.py index f70e07a7..b7749694 100644 --- a/brainscore_core/brainscore_model.py +++ b/brainscore_core/brainscore_model.py @@ -23,20 +23,41 @@ from .streaming import StreamEvent from .streaming_helpers import ( _drain_stream_events as _drain_neural_stream_events, - _drive_behavior_session_streaming, - _drive_behavior_session_via_process, - _drive_environment_session_via_process, - _drive_neural_session_streaming, - _drive_neural_session_via_process, - _drive_state_change_session_via_process, _reconstruct_stimulus_set_from_events as _reconstruct_neural_stimulus_set, _requested_output_channels, _stimuli_from_stream_session as _neural_stimuli_from_stream_session, ) +_PREFERRED_NAMES = { + 'preprocessors': "extractor={modality: extractor} (configure input preparation " + "on the extractor; declare extractor-less inputs with " + "supported_modalities=)", + 'activations_model': 'extractor=', + 'generation_fn': 'answer=', + 'action_fn': 'policy=', + 'state_change_fn': 'intervention=', +} + + +def _warn_old_names(**given) -> None: + """Deprecation warning for each older constructor name that was used.""" + import warnings + for name, value in given.items(): + if value is None or (name == 'preprocessors' and not value): + continue + warnings.warn( + f"BrainScoreModel({name}=...) is deprecated; use " + f"{_PREFERRED_NAMES[name]}. It keeps working until the public release, " + "when the old names are removed together with the interact-to-participate rename.", + DeprecationWarning, stacklevel=3) + + class BrainScoreModel(UnifiedModel): - """Compositional subject with session and legacy typed evaluation paths.""" + """The one concrete subject, assembled from parts (extractors, region map, + answer, policy, intervention, capabilities). Evaluated through ``interact``; + ``process`` is a typed convenience and the legacy benchmark methods forward + to :class:`~brainscore_core.legacy_facade.LegacyBenchmarkFacade`.""" COLUMN_TO_MODALITY: Dict[str, str] = { 'image_file_name': 'vision', @@ -127,6 +148,15 @@ def __init__( Note: passing capability functions POSITIONALLY (via the legacy tail) is deprecated and error-prone; always pass them by keyword. """ + # One name per part (Contract Rules v3): extractor, answer, policy, + # intervention. The older names keep working with a deprecation warning. + _warn_old_names(preprocessors=preprocessors, + activations_model=activations_model, + generation_fn=generation_fn, action_fn=action_fn, + state_change_fn=state_change_fn, + **{k: legacy_capability_config[k] for k in + ('generation_fn', 'action_fn', 'state_change_fn') + if k in legacy_capability_config}) # Translate the public vocabulary once; existing plugins keep their # callback names and the internal capability contract stays unchanged. if extractor is not None: @@ -148,8 +178,7 @@ def __init__( "Passing capability functions positionally to BrainScoreModel is " "deprecated and error-prone: a mis-counted positional silently lands " "in the wrong capability slot. Pass them by keyword instead " - "(behavioral_readout_layer=, generation_fn=, action_fn=, " - "state_change_fn=).", + "(behavioral_readout_layer=, answer=, policy=, intervention=).", DeprecationWarning, stacklevel=2) # Fold the now-explicit capability kwargs into the config the normalizer # sees, so they are visible in the signature/help() yet routed uniformly. @@ -229,33 +258,33 @@ def _declared_capabilities(self): @property def in_channels(self): - channels = super().in_channels | set().union( - *(set(c.input_channels) for c in self._declared_capabilities())) - if self._action_fn is not None: - channels.add('observation') - if self._generation_fn is not None: - channels.add('instruction') - return channels + return super().in_channels | set().union( + *(c.declared_input_channels(self) for c in self._declared_capabilities())) @property def out_channels(self): - return super().out_channels | set().union( - *(set(c.output_channels) for c in self._declared_capabilities())) + # Every output comes from a registered part; nothing is found by name. + return set().union( + *(c.declared_output_channels(self) for c in self._declared_capabilities())) def check_session_support(self, channels): - """Reject unsupported or ambiguous output combinations before consuming input.""" + """The one capability that runs a session requesting ``channels``. + + Raises before any input is consumed when no capability, or more than + one at the same priority, claims the combination. + """ channels = frozenset(channels) if not channels: raise ValueError('A session must request at least one output channel.') - custom = [c for c in self._declared_capabilities() + claims = [c for c in self._declared_capabilities() if c.supports_session(self, channels)] - if len(custom) > 1: - raise ValueError(f'Ambiguous session handlers: {[c.identifier for c in custom]}') - stock = (channels in ({'behavior'}, {'motor'}, {'perturbation'}) - or all(c.startswith('neural:') for c in channels)) - if not custom and not stock: + if not claims: raise NotImplementedError(f'No session handler for {sorted(channels)}') - return custom[0] if custom else None + first = min(c.order for c in claims) + winners = [c for c in claims if c.order == first] + if len(winners) > 1: + raise ValueError(f'Ambiguous session handlers: {[c.identifier for c in winners]}') + return winners[0] @property def identifier(self) -> str: @@ -351,6 +380,9 @@ def required_modalities(self) -> Set[str]: def backbone_id(self) -> str: return self._backbone_id + # Compatibility surface for tests and notebooks that read recording, + # task and perturbation state off the model. Core components read the + # owning component (_recorder, _behavioral, _perturbations) directly. @property def _recording_layer(self) -> Optional[str]: return self._recorder.recording_layer @@ -440,7 +472,8 @@ def process(self, input_event: InputEvent, multi_modality: bool = False) -> OutputEvent: """Evaluate one input event and return the corresponding output event. - The legacy typed evaluation entry point; dispatch is by ``input_event`` type: + A typed convenience, not part of the subject contract (``interact`` is + the evaluation method). Dispatch is by ``input_event`` type: a ``StimulusSet`` / ``MultimodalStimulusSet`` -> ``NeuroidAssembly`` of recorded activations (call ``start_recording`` first); a ``StateChange`` -> ``PerturbationApplied`` (apply) or ``None`` (reset); an @@ -451,43 +484,28 @@ def process(self, input_event: InputEvent, modality and concat along the neuroid axis; default routes a single modality by ``MODALITY_PRIORITY``. """ - return self._dispatcher.process(input_event, multi_modality) + output = self._dispatcher.process(input_event, multi_modality) + time_bins = self._recorder.time_bins + if time_bins is not None and len(time_bins) > 1 and not isinstance( + input_event, (StateChange, EnvironmentStep, Message)): + # Honor every requested time bin: a static response is repeated + # across several bins, as legacy vision models do. + from .legacy_facade import repeat_over_time_bins + output = repeat_over_time_bins(output, time_bins) + return output def interact(self, session) -> None: - """Drive a v2 streaming session through existing model paths.""" - requested_channels = _requested_output_channels(session) - custom = self.check_session_support(requested_channels) - if custom is not None: - from .capabilities.registry import _setup_once - _setup_once(self, custom) - custom.interact(self, session) - return - if requested_channels == ["behavior"]: - if getattr(session, "streaming", False): - _drive_behavior_session_streaming(self, session) - else: - _drive_behavior_session_via_process(self, session) - return - if requested_channels == ["perturbation"]: - _drive_state_change_session_via_process(self, session) - return - if requested_channels == ["motor"]: - _drive_environment_session_via_process(self, session) - return - if all(channel.startswith("neural:") for channel in requested_channels): - # A session may ask to be consumed one input at a time (inputs generated - # on demand rather than known up front). Same scores, different arrival. - if getattr(session, "streaming", False): - _drive_neural_session_streaming(self, session) - else: - _drive_neural_session_via_process(self, session) - return - raise NotImplementedError( - "BrainScoreModel.interact currently supports requested " - "neural:, behavior, perturbation, or motor output " - "channels; got " - f"{requested_channels!r}." - ) + """Run a session through the one registered capability that claims it. + + The handler sees only the streams this model declares in + ``in_channels``; every other stream the experiment sends is skipped + before it reaches any part (Contract Rules v3, rule 5.5). + """ + handler = self.check_session_support(_requested_output_channels(session)) + from .capabilities.registry import _setup_once + from .session_view import DeclaredInputSession + _setup_once(self, handler) + handler.interact(self, DeclaredInputSession(session, self.in_channels)) def _handle_environment_step(self, input_event: 'EnvironmentStep') -> OutputEvent: return self._dispatcher.handle_environment_step(input_event) @@ -525,9 +543,6 @@ def _tag_neuroids_with_regions(self, assembly, modality=None): def _process_composite_regions(self, stimuli, modality): return self._recorder.process_composite_regions(stimuli, modality) - def _gather_composite_units(self, assembly, selector): - return self._recorder.gather_composite_units(assembly, selector) - def start_recording(self, recording_target: Union[str, List[str]], time_bins: Optional[List[Tuple[int, int]]] = None, @@ -544,8 +559,31 @@ def start_recording(self, def start_task(self, task_context, fitting_stimuli=None, **kwargs) -> None: - self._behavioral.start_task(task_context, fitting_stimuli, + """Configure the behavioral task. + + Fitting stimuli reach the readout with only the input modalities this + model declares (rule 5.5); the caller's table is never changed. The + same TaskContext object sent again, as a facade session does for every + measurement, keeps the fitted readout; a new context, or any context + after a state change, fits anew. + """ + from dataclasses import replace + from .session_view import declared_only + if isinstance(task_context, TaskContext): + if self._behavioral.is_configured_for(task_context): + return + fitting = task_context.fitting_stimuli + projected = declared_only(fitting, self.in_channels) + context = (task_context if projected is fitting + else replace(task_context, fitting_stimuli=projected)) + self._behavioral.start_task(context, **kwargs) + self._behavioral.mark_configured(task_context) + return + self._behavioral.start_task(task_context, + declared_only(fitting_stimuli, self.in_channels), **kwargs) + # The context built from (task, fitting) is what a facade sends back. + self._behavioral.mark_configured(self._behavioral.task_context) def reset(self) -> None: failures = [] @@ -582,19 +620,13 @@ def reset(self) -> None: if failures: raise RuntimeError('A model provider failed to reset; the model is not clean.') from failures[0] - def _resolve_selection(self, state_change: 'StateChange') -> 'StateChange': - return self._perturbations.resolve_selection(state_change) - def _dispatch_state_change(self, state_change: 'StateChange') -> Any: - return self._dispatcher.dispatch_state_change(state_change) - - @staticmethod - def _drain_stream_events(session) -> List[StreamEvent]: - return _drain_neural_stream_events(session) - - @staticmethod - def _stimuli_from_stream_session(session, events: List[StreamEvent]): - return _neural_stimuli_from_stream_session(session, events) + try: + return self._dispatcher.dispatch_state_change(state_change) + finally: + # The model changed under the readout: the next start_task fits + # again even when the task object is the one already configured. + self._behavioral.invalidate() @staticmethod def _reconstruct_stimulus_set_from_events(events: List[StreamEvent]): @@ -607,10 +639,6 @@ def _reconstruct_stimulus_set_from_events(events: List[StreamEvent]): """ return _reconstruct_neural_stimulus_set(events) - @staticmethod - def _requested_output_channels(session) -> List[str]: - return _requested_output_channels(session) - @classmethod def _requires_behavioral_readout(cls, task_type: str) -> bool: return BehavioralReadout.requires_behavioral_readout(task_type) @@ -618,10 +646,6 @@ def _requires_behavioral_readout(cls, task_type: str) -> bool: def _fit_behavioral_readout(self, fitting_stimuli) -> None: self._behavioral.fit_behavioral_readout(fitting_stimuli) - @staticmethod - def _extract_labels(stimuli): - return BehavioralReadout.extract_labels(stimuli) - def _extract_behavioral_features(self, stimuli): return self._behavioral.extract_behavioral_features(stimuli) @@ -631,40 +655,45 @@ def _predict_probabilities(self, stimuli): def _generate_predictions(self, stimuli): return self._behavioral.generate_predictions(stimuli) - def look_at(self, stimuli, number_of_trials=1, **kwargs): - """Vision benchmark compatibility. Delegates to process().""" - return self.process(stimuli) + # --- Legacy benchmark surface ------------------------------------------ + # Permanent: legacy benchmarks are never removed. Behavior lives in + # LegacyBenchmarkFacade, which runs each legacy call as a session through + # ``interact``; these forwards seed a facade with this model's current + # recording and task state so direct calls keep working. + + @property + def _legacy_facade(self): + from .legacy_facade import LegacyBenchmarkFacade + facade = LegacyBenchmarkFacade(self, visual_degrees=self._visual_degrees_val) + regions = self._recorder.recording_regions + if regions: + facade.start_recording(regions[0] if len(regions) == 1 else list(regions), + time_bins=self._recorder.time_bins) + if self._behavioral.task_context is not None: + facade.start_task(self._behavioral.task_context) + return facade + + def look_at(self, stimuli, number_of_trials=1, require_variance=False, **kwargs): + """Vision benchmark compatibility (see LegacyBenchmarkFacade.look_at).""" + return self._legacy_facade.look_at(stimuli, number_of_trials, + require_variance, **kwargs) def visual_degrees(self) -> int: """Vision benchmark compatibility.""" return self._visual_degrees_val def digest_text(self, text) -> Dict[str, Any]: - """Language benchmark compatibility.""" - import hashlib - import pandas as pd - from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet - - import numpy as np - if isinstance(text, np.ndarray): - text = text.tolist() - if isinstance(text, (str, list)): - if isinstance(text, str): - text = [text] - stimuli = StimulusSet(pd.DataFrame({ - 'sentence': text, - 'stimulus_id': list(range(len(text))), - 'context_id': ['passage'] * len(text), - })) - content_hash = hashlib.md5( - '|'.join(text).encode('utf-8')).hexdigest()[:12] - stimuli.identifier = f'text_stimuli_{len(text)}_{content_hash}' - else: - stimuli = text - - result = self.process(stimuli) - return {'neural': result} + """Language benchmark compatibility (see LegacyBenchmarkFacade.digest_text).""" + return self._legacy_facade.digest_text(text) def start_neural_recording(self, recording_target, recording_type='fMRI'): """Language benchmark compatibility.""" self.start_recording(recording_target, recording_type=recording_type) + + def start_behavioral_task(self, task): + """Language benchmark compatibility; typed models have no legacy language task path.""" + raise NotImplementedError( + f"Subject {self.identifier!r} has no path for the legacy language " + f"behavioral task {getattr(task, 'value', task)!r} (reading times / " + f"next-word prediction). Legacy language behavioral benchmarks need " + f"a subject that implements that task.") diff --git a/brainscore_core/capabilities/__init__.py b/brainscore_core/capabilities/__init__.py index 1861868a..022d5e69 100644 --- a/brainscore_core/capabilities/__init__.py +++ b/brainscore_core/capabilities/__init__.py @@ -13,3 +13,4 @@ from . import embodied_action as _embodied_action # noqa: F401 from . import neural as _neural # noqa: F401 from . import state_change as _state_change # noqa: F401 +from . import sessions as _sessions # noqa: F401 diff --git a/brainscore_core/capabilities/base.py b/brainscore_core/capabilities/base.py index 89e5888c..1ee081f0 100644 --- a/brainscore_core/capabilities/base.py +++ b/brainscore_core/capabilities/base.py @@ -17,6 +17,16 @@ class Capability(ABC): input_channels = frozenset() output_channels = frozenset() + def declared_input_channels(self, model): + """Input channels this capability adds for ``model``.""" + del model + return set(self.input_channels) + + def declared_output_channels(self, model): + """Output channels this capability adds for ``model``.""" + del model + return set(self.output_channels) + def reset(self, model, state) -> None: """Release experiment state. Retain configuration needed by later calls.""" state.clear() diff --git a/brainscore_core/capabilities/behavioral_generation.py b/brainscore_core/capabilities/behavioral_generation.py index b2427b81..a8f46990 100644 --- a/brainscore_core/capabilities/behavioral_generation.py +++ b/brainscore_core/capabilities/behavioral_generation.py @@ -12,9 +12,9 @@ class BehavioralGenerationCapability(Capability): def enabled_for(self, model) -> bool: return ( - model._use_generation_for_task - and model._task_context is not None - and model._requires_behavioral_readout(model._task_context.task_type) + model._behavioral.use_generation_for_task + and model._behavioral.task_context is not None + and model._requires_behavioral_readout(model._behavioral.task_context.task_type) ) def handles(self, model, event, **kwargs) -> bool: diff --git a/brainscore_core/capabilities/behavioral_readout.py b/brainscore_core/capabilities/behavioral_readout.py index 09e0eb50..2541bf2e 100644 --- a/brainscore_core/capabilities/behavioral_readout.py +++ b/brainscore_core/capabilities/behavioral_readout.py @@ -12,9 +12,9 @@ class BehavioralReadoutCapability(Capability): def enabled_for(self, model) -> bool: return ( - model._readout_classifier is not None - and model._task_context is not None - and model._requires_behavioral_readout(model._task_context.task_type) + model._behavioral.readout_classifier is not None + and model._behavioral.task_context is not None + and model._requires_behavioral_readout(model._behavioral.task_context.task_type) ) def handles(self, model, event, **kwargs) -> bool: diff --git a/brainscore_core/capabilities/neural.py b/brainscore_core/capabilities/neural.py index 721a0f7f..423f3b95 100644 --- a/brainscore_core/capabilities/neural.py +++ b/brainscore_core/capabilities/neural.py @@ -37,7 +37,7 @@ def process(self, model, stimuli, **kwargs): layers = model._recorder.current_layers() modality = model._pick_modality(detected) requested_modalities = { - model._region_modality_map[r] for r in model._recording_regions + model._region_modality_map[r] for r in model._recorder.recording_regions if r in model._region_modality_map} requested_modalities = {canonical_modality(m) for m in requested_modalities} missing = requested_modalities - detected @@ -56,7 +56,7 @@ def process(self, model, stimuli, **kwargs): # false-positive the selected plain-callable path. will_run = detected if (multi_modality and len(detected) > 1) else {modality} if (not layers and model._region_layer_map_dict - and not model._composite_recording + and not model._recorder.composite_recording and any(model._supports_layer_extraction(m) for m in will_run)): raise ValueError( f"No active recording layer for model '{model.identifier}': " @@ -80,12 +80,12 @@ def process(self, model, stimuli, **kwargs): UserWarning, stacklevel=2, ) - if (model._composite_recording + if (model._recorder.composite_recording and model._supports_layer_extraction(modality)): return model._process_composite_regions(stimuli, modality) layers = model._filter_layers_for_modality(layers, modality) assembly = model._extract_for_modality(stimuli, modality, layers) - if model._is_multi_region and model._supports_layer_extraction(modality): + if model._recorder.is_multi_region and model._supports_layer_extraction(modality): assembly = model._tag_neuroids_with_regions(assembly, modality) return assembly diff --git a/brainscore_core/capabilities/sessions.py b/brainscore_core/capabilities/sessions.py new file mode 100644 index 00000000..ec001f4a --- /dev/null +++ b/brainscore_core/capabilities/sessions.py @@ -0,0 +1,114 @@ +"""Built-in session handlers for BrainScoreModel's parts. + +Each part of a BrainScoreModel (region map, answer or readout, policy, +intervention) handles some channels. These capabilities declare those +channels and run the matching sessions, registered the same way as any added +capability, so ``interact``, ``check_session_support`` and the declared +channels all read one registry. They sort after added capabilities +(``order`` 10000), so an added capability that claims the same outputs wins. +""" + +from .base import Capability +from .registry import register_capability +from .. import streaming_helpers as _drivers + + +class _SessionCapability(Capability): + """A capability that only runs sessions; it handles no ``process`` events.""" + + order = 10000 + + def handles(self, model, event, **kwargs) -> bool: + del model, event, kwargs + return False + + def process(self, model, event, **kwargs): # pragma: no cover - never handled + raise NotImplementedError(self.identifier) + + +class NeuralSessions(_SessionCapability): + identifier = 'neural-sessions' + + def enabled_for(self, model) -> bool: + # A region map names where to record; only a wired extractor can + # record there, so a map without one declares nothing (rule 2.4). + return bool(model.region_layer_map) and bool(model._extractors) + + def declared_output_channels(self, model): + return {f'neural:{region}' for region in model.region_layer_map} + + def supports_session(self, model, channels) -> bool: + return bool(channels) and all(c.startswith('neural:') for c in channels) + + def interact(self, model, session) -> None: + if getattr(session, 'streaming', False): + _drivers._drive_neural_session_streaming(model, session) + else: + _drivers._drive_neural_session_via_process(model, session, + allow_multi_modality=True) + + +class BehaviorSessions(_SessionCapability): + identifier = 'behavior-sessions' + + def enabled_for(self, model) -> bool: + return (model._behavioral_readout_layer is not None + or model._generation_fn is not None) + + def declared_input_channels(self, model): + return {'instruction'} if model._generation_fn is not None else set() + + def declared_output_channels(self, model): + return {'behavior'} + + def supports_session(self, model, channels) -> bool: + return set(channels) == {'behavior'} + + def interact(self, model, session) -> None: + if getattr(session, 'streaming', False): + _drivers._drive_behavior_session_streaming(model, session) + else: + _drivers._drive_behavior_session_via_process(model, session) + + +class MotorSessions(_SessionCapability): + identifier = 'motor-sessions' + + def enabled_for(self, model) -> bool: + return model._action_fn is not None + + def declared_input_channels(self, model): + return {'observation'} + + def declared_output_channels(self, model): + return {'motor'} + + def supports_session(self, model, channels) -> bool: + return set(channels) == {'motor'} + + def interact(self, model, session) -> None: + _drivers._drive_environment_session_via_process(model, session) + + +class PerturbationSessions(_SessionCapability): + identifier = 'perturbation-sessions' + + def enabled_for(self, model) -> bool: + return model._state_change_fn is not None + + def declared_input_channels(self, model): + return {'lesion', 'stimulation'} + + def declared_output_channels(self, model): + return {'perturbation'} + + def supports_session(self, model, channels) -> bool: + return set(channels) == {'perturbation'} + + def interact(self, model, session) -> None: + _drivers._drive_state_change_session_via_process(model, session) + + +for _capability in (NeuralSessions(), BehaviorSessions(), MotorSessions(), + PerturbationSessions()): + register_capability(_capability) diff --git a/brainscore_core/compatibility.py b/brainscore_core/compatibility.py index e21d200d..3ae7244f 100644 --- a/brainscore_core/compatibility.py +++ b/brainscore_core/compatibility.py @@ -1,52 +1,37 @@ """ -Pre-flight compatibility checker for Brain-Score. - -Single-direction modality validation (post April 30, 2026 design decision): - -- Benchmark declares ``required_modalities`` — the modalities its stimuli - provide. Each leaderboard benchmark declares **exactly one** input format; - multiple input formats become separate registered variants. -- Model declares ``available_modalities`` (everything it can consume) and - optionally ``required_modalities`` (hard requirement — model cannot run - without these in the stimulus set, e.g. a locked-fusion model). - -A pairing is valid when: - -- ``benchmark.required ⊆ model.available`` — model can process ALL of the - benchmark's required modalities (all-of). -- ``benchmark.accepted ∩ model.available ≠ ∅`` (if ``accepted_modalities`` is - set) — the model provides AT LEAST ONE accepted input format (any-of). This - is for benchmarks that adapt to the candidate, e.g. a naturalistic video - benchmark that runs native-video models and falls back to frame-aggregation - for still-image (vision) models, so it accepts ``{video, vision}``. -- ``benchmark.region ∈ model.region_layer_map`` (if the benchmark targets a - neural region). - -The earlier symmetric check ``model.required ⊆ benchmark.required ∪ -benchmark.available`` was dropped: ``benchmark.available_modalities`` is -deprecated. A locked-fusion model whose required modalities aren't present -in the benchmark's single input format is caught by check #1 (its required -set is a superset of the benchmark's required set, so the benchmark's -required set isn't a subset of the model's available set unless it -exactly matches). - -For models with required_modalities that don't include the benchmark's -required modality, the symmetric check is still useful — e.g. a model that -hard-requires text + vision but the benchmark only provides vision. We keep -that case via a tightened check on model_required: the model's required set -must be a subset of the benchmark's required set (the benchmark's only -input format). +Pre-flight compatibility checker for Brain-Score (Contract Rules v3, rule 5). + +Compatibility is decided from declarations alone: what the benchmark sends +and asks for, and what the subject says it takes in, produces and requires. +A subject is compatible with a benchmark when all three hold: + +1. every requested output is in ``subject.out_channels``; +2. at least one benchmark input is in ``subject.in_channels``; +3. every channel in ``subject.required_channels`` is among the benchmark's + inputs. + +The benchmark declares the inputs its human participants received. A subject +that takes in only some of them still runs: the streams it cannot take in +never enter it. :func:`check_compatibility` returns the match as data +(:class:`CompatibilityMatch`) so the score records which inputs were used. +An incompatible pair raises :class:`CompatibilityError` and is scored N/A. + +Benchmark inputs are ``required_input_channels`` when declared, otherwise the +channels of ``required_modalities`` plus ``accepted_modalities``. Requested +outputs are ``requested_output_channels`` when declared, otherwise +``neural:`` for a benchmark with a ``region``. Benchmarks that still set ``available_modalities`` emit a ``DeprecationWarning`` directing the maintainer to register separate variants. """ import warnings -from typing import Optional, Set +from dataclasses import dataclass +from typing import Dict, FrozenSet, Optional, Set from . import io_catalog from .io_catalog import canonical_modality, modalities_to_input_channels -from .model_interface import Subject, UnifiedModel +from .model_interface import Subject from .streaming import parse_channel @@ -69,88 +54,112 @@ def ensure_legacy_benchmark_modalities(benchmark, modalities): return benchmark -def check_channel_compatibility(subject: Subject, benchmark) -> None: - """Check the UMI v2.0 channel compatibility contract. +@dataclass(frozen=True) +class CompatibilityMatch: + """How a subject matches a benchmark, returned by pre-flight. - Raises :class:`CompatibilityError` before compute when any of the three - channel-set conditions fail: - - 1. ``benchmark.required_input_channels <= subject.in_channels`` - 2. ``benchmark.requested_output_channels <= subject.out_channels`` - 3. ``subject.required_channels <= benchmark.required_input_channels`` + ``consumed_inputs`` are the benchmark inputs the subject takes in and + ``dropped_inputs`` the ones it cannot, which never enter it. ``kind`` is + ``'full'`` when nothing is dropped, otherwise ``'partial'``. + """ + subject: str + benchmark: str + consumed_inputs: FrozenSet[str] + dropped_inputs: FrozenSet[str] + requested_outputs: FrozenSet[str] + + @property + def kind(self) -> str: + return 'partial' if self.dropped_inputs else 'full' + + def as_attrs(self) -> Dict[str, object]: + """Plain-data form for ``score.attrs['compatibility']``.""" + return { + 'match': self.kind, + 'consumed_inputs': sorted(self.consumed_inputs), + 'dropped_inputs': sorted(self.dropped_inputs), + 'requested_outputs': sorted(self.requested_outputs), + } + + +def check_channel_compatibility(subject: Subject, benchmark) -> CompatibilityMatch: + """Check the three compatibility conditions and return the match. + + Raises :class:`CompatibilityError` before compute when a condition fails. + Every declared channel must be registered in the catalog with the right + direction. """ subject_in = set(subject.in_channels) subject_out = set(subject.out_channels) subject_required = set(subject.required_channels) - bench_required = benchmark_required_input_channels(benchmark) - bench_accepted = benchmark_accepted_input_channels(benchmark) + bench_inputs = benchmark_input_channels(benchmark) bench_requested = benchmark_requested_output_channels(benchmark) - - validate_channels( - subject_in, io_catalog.INPUT, f"subject '{subject.identifier}' in_channels") - validate_channels( - subject_out, io_catalog.OUTPUT, f"subject '{subject.identifier}' out_channels") - validate_channels( - subject_required, io_catalog.INPUT, - f"subject '{subject.identifier}' required_channels") - validate_channels( - bench_required, io_catalog.INPUT, - f"benchmark '{_benchmark_identifier(benchmark)}' required_input_channels") - validate_channels( - bench_accepted, io_catalog.INPUT, - f"benchmark '{_benchmark_identifier(benchmark)}' accepted_input_channels") - validate_channels( - bench_requested, io_catalog.OUTPUT, - f"benchmark '{_benchmark_identifier(benchmark)}' requested_output_channels") - - missing_inputs = bench_required - subject_in - if missing_inputs: - raise CompatibilityError( - f"Subject '{subject.identifier}' is missing required input " - f"channels for benchmark '{_benchmark_identifier(benchmark)}': " - f"{_format_channels(missing_inputs)}. Subject in_channels: " - f"{_format_channels(subject_in)}." - ) - - if bench_accepted and not (bench_accepted & subject_in): - raise CompatibilityError( - f"Subject '{subject.identifier}' provides none of the accepted " - f"input channels for benchmark '{_benchmark_identifier(benchmark)}': " - f"{_format_channels(bench_accepted)}. Subject in_channels: " - f"{_format_channels(subject_in)}." - ) + subject_id = subject.identifier + benchmark_id = _benchmark_identifier(benchmark) + + validate_channels(subject_in, io_catalog.INPUT, f"subject '{subject_id}' in_channels") + validate_channels(subject_out, io_catalog.OUTPUT, f"subject '{subject_id}' out_channels") + validate_channels(subject_required, io_catalog.INPUT, + f"subject '{subject_id}' required_channels") + validate_channels(bench_inputs, io_catalog.INPUT, + f"benchmark '{benchmark_id}' input channels") + validate_channels(bench_requested, io_catalog.OUTPUT, + f"benchmark '{benchmark_id}' requested_output_channels") missing_outputs = bench_requested - subject_out if missing_outputs: + message = ( + f"Subject '{subject_id}' cannot produce output channels requested by " + f"benchmark '{benchmark_id}': {_format_channels(missing_outputs)}. " + f"Subject out_channels: {_format_channels(subject_out)}." + ) + regions = sorted(address for family, address in map(parse_channel, missing_outputs) + if family == 'neural' and address is not None) + if regions: + message += ( + f" It has no layer mapping for region '{regions[0]}'. Use the " + f"layer-mapping explorer (brainscore_vision.tools.layer_mapping_explorer) " + f"to find and commit a region-to-layer mapping." + ) + raise CompatibilityError(message) + + consumed = bench_inputs & subject_in + if bench_inputs and not consumed: raise CompatibilityError( - f"Subject '{subject.identifier}' cannot emit requested output " - f"channels for benchmark '{_benchmark_identifier(benchmark)}': " - f"{_format_channels(missing_outputs)}. Subject out_channels: " - f"{_format_channels(subject_out)}." + f"Subject '{subject_id}' takes in none of the inputs benchmark " + f"'{benchmark_id}' sends: {_format_channels(bench_inputs)}. " + f"Subject in_channels: {_format_channels(subject_in)}." ) - unmet_requirements = subject_required - (bench_required | bench_accepted) + unmet_requirements = subject_required - bench_inputs if unmet_requirements: raise CompatibilityError( - f"Subject '{subject.identifier}' hard-requires input channels " + f"Subject '{subject_id}' hard-requires input channels " f"{_format_channels(unmet_requirements)} but benchmark " - f"'{_benchmark_identifier(benchmark)}' provides " - f"{_format_channels(bench_required | bench_accepted)}." + f"'{benchmark_id}' sends {_format_channels(bench_inputs)}." ) if getattr(benchmark, "uses_session", False): probe = getattr(subject, "check_session_support", None) - if not callable(probe): - # Native subjects implement the session contract directly. The - # optional probe belongs to configurable compatibility wrappers. - if not (isinstance(subject, Subject) and not isinstance(subject, UnifiedModel)): - raise CompatibilityError("Session benchmark requires a session-support probe") - else: + if callable(probe): try: probe(bench_requested) except (ValueError, NotImplementedError) as error: raise CompatibilityError(str(error)) from error + return CompatibilityMatch( + subject=subject_id, + benchmark=benchmark_id, + consumed_inputs=frozenset(consumed), + dropped_inputs=frozenset(bench_inputs - subject_in), + requested_outputs=frozenset(bench_requested), + ) + + +def benchmark_input_channels(benchmark) -> Set[str]: + """Every input channel the benchmark sends.""" + return benchmark_required_input_channels(benchmark) | benchmark_accepted_input_channels(benchmark) + def benchmark_required_input_channels(benchmark) -> Set[str]: """Return v2 required input channels, deriving from v1.5 modalities.""" @@ -235,36 +244,23 @@ def _benchmark_identifier(benchmark) -> str: return getattr(benchmark, "identifier", "?") -def check_compatibility(model: Subject, benchmark) -> None: - """Fast pre-flight check (see module docstring for the full contract). +def check_compatibility(model: Subject, benchmark) -> CompatibilityMatch: + """Fast pre-flight check; returns the match or raises CompatibilityError. - Raises :class:`CompatibilityError` for hard mismatches. - Emits :class:`DeprecationWarning` if the benchmark still sets the - deprecated ``available_modalities`` field. + One rule for every subject (see the module docstring). Emits a + :class:`DeprecationWarning` if the benchmark still sets the deprecated + ``available_modalities`` field, before any compatibility error. """ - # Native subjects declare channels, not modality names or layer maps. - # Keep the legacy checks unchanged for typed models and domain adapters. - if isinstance(model, Subject) and not isinstance(model, UnifiedModel): - check_channel_compatibility(model, benchmark) - return - - # canonicalize so aliases match (e.g. a benchmark requiring 'video' is - # satisfied by a model reporting the canonical 'vision') - model_available: Set[str] = {canonical_modality(m) for m in model.available_modalities} - model_required: Set[str] = {canonical_modality(m) for m in model.required_modalities} bench_required: Set[str] = {canonical_modality(m) for m in getattr(benchmark, 'required_modalities', set())} - - # Deprecation warning — must fire before any short-circuit error so - # benchmark maintainers see it even when the pairing is incompatible. bench_available_attr = getattr(benchmark, 'available_modalities', None) if bench_available_attr is not None: bench_available: Set[str] = {canonical_modality(m) for m in bench_available_attr} - # Only warn if it actually adds something beyond required (otherwise - # it's a redundant declaration not a multi-input claim). + # Only warn if it adds something beyond required (otherwise it is a + # redundant declaration, not a multi-input claim). if bench_available - bench_required: warnings.warn( - f"Benchmark '{benchmark.identifier}' sets " + f"Benchmark '{_benchmark_identifier(benchmark)}' sets " f"`available_modalities={bench_available}` — this field is " f"deprecated as of the April 30, 2026 design decision. " f"Benchmarks now declare exactly one input format via " @@ -275,58 +271,10 @@ def check_compatibility(model: Subject, benchmark) -> None: stacklevel=2, ) - # Check 1: Hard gate — benchmark's required modalities must ALL be - # consumable by the model. - missing_required = bench_required - model_available - if missing_required: - raise CompatibilityError( - f"Model '{model.identifier}' does not support modalities required by " - f"benchmark '{benchmark.identifier}': {missing_required}. " - f"Model available modalities: {model_available}." - ) - - # Check 1b: Any-of gate — some benchmarks accept ANY ONE of several input - # formats (e.g. a naturalistic benchmark that runs native-video models AND - # falls back to frame-aggregation for still-image models). Declared as - # `accepted_modalities`; the model must provide at least one. Distinct from - # `required_modalities` (all-of); empty/absent means no any-of constraint. - bench_accepted: Set[str] = {canonical_modality(m) - for m in getattr(benchmark, 'accepted_modalities', set())} - if bench_accepted and not (bench_accepted & model_available): - raise CompatibilityError( - f"Model '{model.identifier}' provides none of the input formats " - f"benchmark '{benchmark.identifier}' accepts; needs at least one of " - f"{bench_accepted}, model available: {model_available}." - ) - - # Check 2: Hard gate — model's required modalities must be a subset of the - # modalities the benchmark can provide (its required set plus any any-of - # accepted formats). A locked-fusion model that needs modalities the - # benchmark never provides cannot run. - model_missing = model_required - (bench_required | bench_accepted) - if model_missing: - raise CompatibilityError( - f"Model '{model.identifier}' hard-requires modalities " - f"{model_missing} but benchmark '{benchmark.identifier}' does not " - f"provide them. Benchmark provides: " - f"{(bench_required | bench_accepted) or '{}'}." - ) - - # Check 3: Region mapping. - required_region: Optional[str] = getattr(benchmark, 'region', None) - if required_region and required_region not in model.region_layer_map: - raise CompatibilityError( - f"Model '{model.identifier}' has no layer mapping for region " - f"'{required_region}'. Mapped regions: {set(model.region_layer_map.keys())}. " - f"Use the layer-mapping explorer " - f"(brainscore_vision.tools.layer_mapping_explorer) to find and commit " - f"a region-to-layer mapping." - ) - - # Check 4 (v1.5): Input/Output Catalog documentation conformance. Warn-only; - # never changes pass/fail. Surfaces inputs or outputs that are not documented - # in the catalog so undocumented channels are flagged at pre-flight. + match = check_channel_compatibility(model, benchmark) + # Warn-only documentation conformance; never changes the result. check_io_catalog(model, benchmark) + return match def check_io_catalog(model: Subject, benchmark) -> None: diff --git a/brainscore_core/contract.py b/brainscore_core/contract.py index 51560611..3119512a 100644 --- a/brainscore_core/contract.py +++ b/brainscore_core/contract.py @@ -87,16 +87,15 @@ def reset(self) -> None: class UnifiedModel(Subject): - """Subject base with task, recording, and process methods. + """Subject base for models evaluated through typed paths. - Existing implementations retain ``process``, ``start_task``, - ``start_recording``, layer maps, and modality declarations. Channel defaults - are derived from those declarations. ``BrainScoreModel`` and domain adapters - provide session drivers over their existing computation paths. + Implementations provide ``process``, ``start_task``, ``start_recording``, + a layer map and modality declarations. ``interact`` (the evaluation + method of the contract) runs a session through those typed paths, so a + typed model needs no session code of its own. ``process`` is a typed + convenience of this base, not part of the subject contract. New session-native implementations should inherit ``Subject`` directly. - ``UnifiedModel`` remains a ``Subject`` subclass, but is no longer an alias - for it: the legacy requirements must not constrain the native contract. """ @property @@ -137,43 +136,33 @@ def required_modalities(self) -> Set[str]: @property def in_channels(self) -> Set[str]: - """Input stream channels this subject can consume. - - Defaults to the v1.5 modality declarations so existing subjects get a - conservative v2 channel identity without overriding this property. - """ - return self._modalities_to_input_channels( - self._declared_input_modalities() - ) + """Input channels: the declared modalities as channel families.""" + return modalities_to_input_channels(self.available_modalities) @property def out_channels(self) -> Set[str]: - """Output stream channels this subject can emit.""" - channels = { - f"neural:{region}" - for region in self.region_layer_map - } - if self._has_behavioral_output_path(): - channels.add("behavior") - if self._has_action_output_path(): - channels.add("motor") - return channels + """Output channels: ``neural:`` for every mapped region. + + Typed models that also answer tasks or act override this to add + ``behavior`` or ``motor``; nothing is discovered from attribute names. + """ + return {f"neural:{region}" for region in self.region_layer_map} @property def required_channels(self) -> Set[str]: """Input stream channels this subject hard-requires.""" - return self._modalities_to_input_channels(self.required_modalities) + return modalities_to_input_channels(self.required_modalities) - @abstractmethod def process(self, input_event: InputEvent) -> OutputEvent: - ... - - def interact(self, session) -> None: - identifier = self._safe_identifier() + """Typed convenience used by the default ``interact``; not part of the contract.""" raise NotImplementedError( - f"Subject {identifier} has no v2 interact() path yet; use " - f"process()/start_recording()." - ) + f"{type(self).__name__} defines no process(); implement interact(session) " + "or process(input_event) for the typed session driver.") + + def interact(self, session: "Session") -> None: + """Run ``session`` through this model's typed paths.""" + from .streaming_helpers import drive_typed_session + drive_typed_session(self, session) def start_task(self, task_context: TaskContext) -> None: self._task_context: Optional[TaskContext] = task_context @@ -186,52 +175,3 @@ def start_recording(self, def reset(self) -> None: pass - - def _declared_input_modalities(self) -> Set[str]: - for attr in ("available_modalities", "supported_modalities"): - try: - modalities = getattr(self, attr) - except (AttributeError, NotImplementedError): - continue - if modalities is not None: - return set(modalities) - - preprocessors = getattr(self, "_preprocessors", None) - if hasattr(preprocessors, "keys"): - return set(preprocessors.keys()) - return set() - - @staticmethod - def _modalities_to_input_channels(modalities) -> Set[str]: - return modalities_to_input_channels(modalities) - - def _has_behavioral_output_path(self) -> bool: - for attr in ( - "_behavioral_readout_layer", - "_generation_fn", - "behavioral_readout_layer", - "generation_fn", - ): - try: - value = getattr(self, attr) - except Exception: - continue - if value is not None: - return True - return False - - def _has_action_output_path(self) -> bool: - for attr in ("_action_fn", "action_fn"): - try: - value = getattr(self, attr) - except Exception: - continue - if value is not None: - return True - return False - - def _safe_identifier(self) -> str: - try: - return str(self.identifier) - except Exception: - return type(self).__name__ diff --git a/brainscore_core/dispatch.py b/brainscore_core/dispatch.py index ea12a3a7..289513b4 100644 --- a/brainscore_core/dispatch.py +++ b/brainscore_core/dispatch.py @@ -113,7 +113,8 @@ def normalize_extractors(owner, preprocessors, activations_model): continue import warnings warnings.warn( - 'Extraction callbacks in preprocessors are deprecated. Use ' + 'Extraction callbacks in preprocessors are deprecated and are removed at ' + 'the public release. Use ' 'activations_model={modality: extractor}, where extractor accepts ' '(stimuli, layers=...).', DeprecationWarning, stacklevel=3) @@ -232,7 +233,7 @@ def extract_for_modality(self, stimuli, modality: str, layers: List[str]): # activation cache key does not fingerprint, so a cached UNperturbed # extraction would be wrongly returned for a perturbed run. Bypass the # cache while any perturbation is active. - if self.owner._active_perturbations: + if self.owner._perturbations.active_perturbations: with _activation_cache_disabled(): return self._extract_for_modality_inner(stimuli, modality, layers) return self._extract_for_modality_inner(stimuli, modality, layers) @@ -272,10 +273,10 @@ def process_multi_modality(self, stimuli, detected: Set[str], layers, modality) if (owner._region_modality_map and not modality_layers - and owner._recording_regions): + and owner._recorder.recording_regions): continue sub = (owner._process_composite_regions(stimuli, modality) - if owner._composite_recording else + if owner._recorder.composite_recording else owner._extract_for_modality(stimuli, modality, modality_layers)) if not hasattr(sub, 'dims') or 'neuroid' not in sub.dims: raise TypeError( @@ -285,7 +286,7 @@ def process_multi_modality(self, stimuli, detected: Set[str], f"{type(sub).__name__}. Upgrade the preprocessor to a " f"layer-aware extractor (TextWrapper / VLMVisionWrapper)." ) - if owner._is_multi_region and not owner._composite_recording: + if owner._recorder.is_multi_region and not owner._recorder.composite_recording: sub = owner._tag_neuroids_with_regions(sub, modality) n_neuroid = sub.sizes['neuroid'] sub = sub.assign_coords( diff --git a/brainscore_core/extensions.py b/brainscore_core/extensions.py index fc23d457..df3ac158 100644 --- a/brainscore_core/extensions.py +++ b/brainscore_core/extensions.py @@ -8,8 +8,7 @@ def register_channel(entry: CatalogEntry, *, columns=(), replace=False): """Register a schema and optional stimulus columns without editing core. - Existing registrations require explicit replacement. The legacy catalog - register() function retains its historical replacement behavior. + Existing registrations require explicit replacement. """ from .streaming import parse_channel from .brainscore_model import BrainScoreModel @@ -27,7 +26,7 @@ def register_channel(entry: CatalogEntry, *, columns=(), replace=False): existing = {**BrainScoreModel.COLUMN_TO_MODALITY, **io_catalog.stimulus_columns()}.get(column) if existing is not None and existing != entry.name and not replace: raise ValueError(f'Column {column!r} already belongs to {existing!r}') - io_catalog.register(entry) + io_catalog.register(entry, replace=replace) io_catalog._STIMULUS_COLUMNS.update({column: entry.name for column in columns}) return entry diff --git a/brainscore_core/extractor.py b/brainscore_core/extractor.py new file mode 100644 index 00000000..a14b2296 --- /dev/null +++ b/brainscore_core/extractor.py @@ -0,0 +1,24 @@ +"""The extractor protocol: what records activity for one input modality. + +An extractor owns input preparation, the forward pass, hooks, caching and +assembly packaging. ``BrainScoreModel(extractor={modality: extractor})`` +takes one per modality; the keys declare the modalities the subject takes in. +"Wrapper" is the name of the extractors Brain-Score ships (``TextWrapper``, +``VisionWrapper``, ...). Nothing needs to subclass this protocol. +""" + +from typing import Any, Protocol, Sequence + + +class Extractor(Protocol): + """Callable ``(stimuli, layers) -> NeuroidAssembly``. + + ``identifier`` and ``backbone_id`` key the activation cache (registrations + that share weights share ``backbone_id``). An extractor that keeps state + between calls also defines ``reset()``. + """ + + identifier: str + + def __call__(self, stimuli: Any, layers: Sequence[str]) -> Any: + ... diff --git a/brainscore_core/io_catalog.py b/brainscore_core/io_catalog.py index 08502878..341be1e5 100644 --- a/brainscore_core/io_catalog.py +++ b/brainscore_core/io_catalog.py @@ -71,6 +71,7 @@ class CatalogEntry: shape_validator: Optional[ShapeValidator] = None owner: str = "core" materializers: Tuple[str, ...] = () + payload_type: Any = None # type or tuple of types a payload must be; None means unchecked def __post_init__(self): if self.kind not in _DIRECTIONS: @@ -89,8 +90,18 @@ def __post_init__(self): _CATALOG: Dict[str, CatalogEntry] = {} -def register(entry: CatalogEntry) -> None: - """Add or replace a catalog entry, keyed by its family name.""" +def register(entry: CatalogEntry, *, replace: bool = False) -> None: + """Add a catalog entry, keyed by its family name. + + Registering a family that already exists is an error unless + ``replace=True`` (Contract Rules v3, rule 3.5). Re-registering the same + entry object is a no-op. + """ + existing = _CATALOG.get(entry.name) + if existing is not None and existing is not entry and not replace: + raise ValueError( + f"Channel family {entry.name!r} is already registered " + f"(owner {existing.owner!r}); pass replace=True to replace it.") _CATALOG[entry.name] = entry @@ -356,6 +367,12 @@ def validate( meta_keys=RESPONSE_META_KEYS, owner="environment harness", materializers=("EnvironmentResponse",)), + # --- outputs: perturbation record (carried by PerturbationApplied) --- + CatalogEntry("perturbation", OUTPUT, "PerturbationApplied", + "PerturbationApplied record with a handle_id for undoing one change", + "the model's intervention (state_change_fn)", + owner="core", + materializers=("PerturbationApplied",)), # --- outputs: physiological readouts (reserved for human harnesses) --- CatalogEntry("pupil", OUTPUT, "human harness recording", "(t,) float pupil diameter time series, or scalar diameter", @@ -373,6 +390,52 @@ def validate( register(_entry) +def payload_type_notes(label: str, payload) -> List[str]: + """Report a payload whose type does not match its family's ``payload_type``. + + Warn-first (Contract Rules v3, rule 3.2): a mismatch is a note, never a + failure, so sessions that run today keep running while the catalog's + types are checked against the notebooks and examples. Promote to an error + by moving this check into :func:`validate` once that review is done. + """ + entry = get(label) + if entry.payload_type is None or isinstance(payload, entry.payload_type): + return [] + expected = entry.payload_type + names = (expected.__name__ if isinstance(expected, type) + else ', '.join(t.__name__ for t in expected)) + return [f"{label}: payload is {type(payload).__name__}, not {names} " + f"({entry.payload_contract})"] + + +def _declare_payload_types() -> None: + """Attach the payload type of each seeded family. + + Done after seeding, with local imports, so the catalog module itself keeps + no import-time dependency on the event and assembly modules. + """ + from dataclasses import replace as _replace + from pathlib import Path + import numpy as np + from .events import EnvironmentResponse, EnvironmentStep, PerturbationApplied, StateChange + from .supported_data_standards.brainio.assemblies import DataAssembly + + sensory = (str, Path, np.ndarray) + payload_types = { + 'vision': sensory, 'video': sensory, 'audio': sensory, + 'text': str, 'instruction': str, + 'lesion': StateChange, 'stimulation': StateChange, 'pharmacological': StateChange, + 'observation': EnvironmentStep, + 'neural': DataAssembly, 'behavior': DataAssembly, + 'motor': EnvironmentResponse, 'perturbation': PerturbationApplied, + } + for family, payload_type in payload_types.items(): + register(_replace(get(family), payload_type=payload_type), replace=True) + + +_declare_payload_types() + + _STIMULUS_COLUMNS: Dict[str, str] = {} diff --git a/brainscore_core/legacy_facade.py b/brainscore_core/legacy_facade.py new file mode 100644 index 00000000..9b206339 --- /dev/null +++ b/brainscore_core/legacy_facade.py @@ -0,0 +1,284 @@ +"""Run any subject on legacy vision and language benchmarks. + +Legacy benchmarks call ``look_at`` / ``visual_degrees`` / ``start_task`` / +``start_recording`` (vision) or ``digest_text`` / ``start_neural_recording`` / +``start_behavioral_task`` (language). Those benchmarks are never removed, so +this facade is permanent (Contract Rules v3, rule 11.2). It records each +legacy setting, presents the stimuli as a ``BufferedSession`` that carries +those settings (time bins, trial count, task, recording type) and runs the +session through the subject's ``interact``. Nothing here depends on what kind +of subject is wrapped: a ``BrainScoreModel``, a domain adapter and an +interact-only ``Subject`` all take part the same way. + +``subject_for_benchmark`` applies the facade at the scoring boundary to +benchmarks a domain loader marked as legacy. ``BrainScoreModel`` forwards its +own legacy methods to a facade seeded with its current recording and task +state, so direct calls keep working too. +""" + +import hashlib +from typing import Any, Dict, List, Optional, Sequence, Tuple + +from .contract import TaskContext +from .streaming import BufferedSession + +PASSIVE = 'passive' + +LEGACY_API = 'legacy' # calls look_at / digest_text and friends +TYPED_API = 'typed' # calls start_recording + process on the candidate +SESSION_API = 'session' # declares channels and drives interact itself + + +class LegacyBenchmarkFacade: + """Present a subject as a legacy ``BrainModel`` / ``ArtificialSubject``. + + :param subject: any subject with ``interact``, ``reset``, ``identifier`` + and ``out_channels``. + :param visual_degrees: field of view reported to vision benchmarks, or + ``None`` when the subject declares none. + """ + + def __init__(self, subject, visual_degrees: Optional[float] = None) -> None: + self._subject = subject + self._visual_degrees = visual_degrees + self._record: Any = None # region name, list of names or layer path + self._time_bins: Any = None + self._recording_type: Optional[str] = None + self._task: Optional[TaskContext] = None + + @property + def identifier(self) -> str: + return self._subject.identifier + + @property + def subject(self): + return self._subject + + # --- vision: BrainModel --------------------------------------------- + + def visual_degrees(self) -> float: + if self._visual_degrees is None: + raise NotImplementedError( + f"Subject {self.identifier!r} declares no visual field, and " + "legacy vision benchmarks place stimuli by visual degrees. Pass " + "visual_degrees= to LegacyBenchmarkFacade or define " + "visual_degrees() on the subject.") + return self._visual_degrees + + def start_task(self, task, fitting_stimuli=None, **settings) -> None: + """Record the task; it reaches the subject with the next measurement. + + Keyword settings a benchmark passes to ``start_task`` (Malania2007 + sends ``number_of_trials`` and ``require_variance``) are fitting + settings. They travel as ``metadata['task_settings']``, apart from the + measurement settings of each ``look_at``, which may use the same names. + The one TaskContext object is sent with every measurement, so a subject + can tell a repeated task from a new one and keep its fitted state. + """ + if isinstance(task, TaskContext): + self._task = task + return + metadata = {'task_settings': dict(settings)} if settings else {} + self._task = TaskContext(task_type=task, fitting_stimuli=fitting_stimuli, + metadata=metadata) + + def start_recording(self, recording_target, time_bins=None, + recording_type=None) -> None: + self._record = recording_target + self._time_bins = time_bins + self._recording_type = recording_type + + def look_at(self, stimuli, number_of_trials: int = 1, + require_variance: bool = False, **settings): + """Legacy ``BrainModel.look_at``: one session requesting the configured output. + + ``number_of_trials`` and ``require_variance`` travel as the session's + measurement ``settings``. A subject that cannot produce trial-to-trial + variance fails instead of returning identical trials. + """ + if settings: + raise TypeError(f'look_at got unsupported settings {sorted(settings)}') + channel = 'behavior' if self._behavior_requested() else self._neural_channel('look_at') + measured = self._measure(stimuli, [channel], { + 'number_of_trials': number_of_trials, + 'require_variance': require_variance, + }) + return measured[channel] + + # --- language: ArtificialSubject ------------------------------------ + + def start_neural_recording(self, recording_target, recording_type='fMRI') -> None: + self.start_recording(recording_target, recording_type=recording_type) + + def start_behavioral_task(self, task) -> None: + if 'behavior' not in set(self._subject.out_channels): + raise NotImplementedError( + f"Subject {self.identifier!r} has no path for the legacy language " + f"behavioral task {getattr(task, 'value', task)!r} (reading times / " + f"next-word prediction): it declares no 'behavior' output. Legacy " + f"language behavioral benchmarks need a subject that implements " + f"that task.") + self.start_task(task) + + def digest_text(self, text) -> Dict[str, Any]: + """Legacy ``ArtificialSubject.digest_text``: every configured output in one call. + + Returns ``{'neural': ...}`` and/or ``{'behavior': ...}`` according to + what ``start_neural_recording`` / ``start_behavioral_task`` configured. + Plain text is handed to the subject unchanged as ``legacy_text`` in the + task settings, so a legacy adapter can replay the exact original call. + """ + stimuli = text_stimulus_set(text) + channels: Dict[str, str] = {} + if self._record is not None: + channels['neural'] = self._neural_channel('digest_text') + if self._behavior_requested(): + channels['behavior'] = 'behavior' + if not channels: + raise ValueError('Call start_neural_recording or start_behavioral_task ' + 'before digest_text') + settings = {} if stimuli is text else {'legacy_text': text} + measured = self._measure(stimuli, list(channels.values()), settings) + return {name: measured[channel] for name, channel in channels.items()} + + def reset(self) -> None: + self._subject.reset() + self._record = self._time_bins = self._recording_type = self._task = None + self._passive = None + + # --- internals -------------------------------------------------------- + + _passive: Optional[TaskContext] = None + + def _behavior_requested(self) -> bool: + return self._task is not None and self._task.task_type != PASSIVE + + def _task_context(self) -> TaskContext: + """The task sent with each measurement: the recorded one, else one passive context.""" + if self._task is not None: + return self._task + if self._passive is None: + self._passive = TaskContext(task_type=PASSIVE) + return self._passive + + def _neural_channel(self, method: str) -> str: + if self._record is None: + raise ValueError(f'Call start_recording or start_task before {method}') + if not isinstance(self._record, str): + raise ValueError( + f'{method} returns one region; several were requested ' + f'({list(self._record)}). Use neural_response(..., record=[...]) ' + f'or process() for several regions.') + return f'neural:{self._record}' + + def _measure(self, stimuli, channels: List[str], settings: Dict[str, Any]): + """One session: the recorded task object, the time bins and the measurement settings.""" + if self._recording_type is not None: + settings['recording_type'] = self._recording_type + session = BufferedSession(stimulus_set=stimuli, + requested_output_channels=channels, + task_context=self._task_context(), + time_bins=self._time_bins) + session.settings = dict(settings) + self._subject.interact(session) + return {channel: session.collect(channel) for channel in channels} + + +def mark_legacy_benchmark(benchmark): + """Record a benchmark's calling convention at a domain loader. + + A benchmark that declares its own input channels drives sessions itself; + one that declares none calls the legacy methods. A benchmark class may + set ``subject_api`` itself (``'typed'`` for the retained + ``start_recording`` + ``process`` convention). + """ + if getattr(benchmark, 'subject_api', None) is None: + declares_channels = getattr(benchmark, 'required_input_channels', None) is not None + benchmark.subject_api = SESSION_API if declares_channels else LEGACY_API + return benchmark + + +def subject_for_benchmark(subject, benchmark, *, visual_degrees: Optional[float] = None): + """The candidate a benchmark receives: the facade for legacy benchmarks, else the subject.""" + if getattr(benchmark, 'subject_api', None) != LEGACY_API: + return subject + return LegacyBenchmarkFacade(subject, visual_degrees=visual_degrees) + + +def declared_visual_degrees(subject) -> Optional[float]: + """The field of view a vision candidate reports, or ``None`` when it has none. + + ``visual_degrees()`` is part of the legacy vision ``BrainModel`` surface, a + measurement setting of the vision harness rather than an ability; a + subject without it gets no invented value. + """ + degrees = getattr(subject, 'visual_degrees', None) + if degrees is None: + return None + return degrees() if callable(degrees) else degrees + + +def text_stimulus_set(text): + """The stimulus set legacy ``digest_text`` callers mean by ``text``. + + A string or list of strings is one passage, in order. A stimulus set is + passed through unchanged. The identifier hashes the content so different + passages never share a cache entry. + """ + import numpy as np + import pandas as pd + from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet + + if isinstance(text, np.ndarray): + text = text.tolist() + if not isinstance(text, (str, list)): + return text + if isinstance(text, str): + text = [text] + stimuli = StimulusSet(pd.DataFrame({ + 'sentence': text, + 'stimulus_id': list(range(len(text))), + 'context_id': ['passage'] * len(text), + })) + # Length prefixes keep different segmentations of the same characters + # (['a|b', 'c'] and ['a', 'b|c']) from sharing one cache entry. + content = ''.join(f'{len(part)}:{part}' for part in text).encode('utf-8') + content_hash = hashlib.md5(content).hexdigest()[:12] + stimuli.identifier = f'text_stimuli_{len(text)}_{content_hash}' + return stimuli + + +def repeat_over_time_bins(assembly, time_bins: Optional[Sequence[Tuple[int, int]]]): + """Repeat a static response across several requested time bins. + + Matches the legacy vision behavior for models without a time axis: one + requested bin leaves the response unchanged; several bins repeat it along a + ``time_bin`` dimension indexed by ``(time_bin_start, time_bin_end)``. + """ + bins = _time_bin_pairs(time_bins) + if len(bins) < 2 or 'time_bin' in getattr(assembly, 'dims', ()): + return assembly + import pandas as pd + import xarray as xr + + plain = xr.DataArray(assembly) + stacked = xr.concat([plain] * len(bins), dim='time_bin') + stacked = stacked.assign_coords(time_bin=pd.MultiIndex.from_tuples( + bins, names=['time_bin_start', 'time_bin_end'])) + return type(assembly)(stacked) + + +def _time_bin_pairs(time_bins) -> List[Tuple[Any, Any]]: + """Normalize benchmark time bins (lists, arrays, MultiIndex values) to pairs.""" + import numpy as np + if time_bins is None: + return [] + items = time_bins.tolist() if isinstance(time_bins, np.ndarray) else list(time_bins) + if len(items) == 2 and all(np.isscalar(value) for value in items): + return [(items[0], items[1])] # one flat (start, end) pair + pairs = [] + for item in items: + item = item.tolist() if isinstance(item, np.ndarray) else item + start, end = item + pairs.append((start, end)) + return pairs diff --git a/brainscore_core/memory.py b/brainscore_core/memory.py index 3d421590..c4aefe1d 100644 --- a/brainscore_core/memory.py +++ b/brainscore_core/memory.py @@ -26,7 +26,7 @@ from typing import Optional from .execution_plan import ExecutionPlan, _positive_int as _validate_positive_int -from .contract import Subject, UnifiedModel +from .contract import Subject logger = logging.getLogger(__name__) @@ -368,25 +368,28 @@ def _get_execution_plan(benchmark) -> Optional[ExecutionPlan]: def _probe_regions(model, benchmark): """Regions to try recording for the probe, in order. - The benchmark's own ``.region`` if it exposes one; otherwise the model's - ``region_layer_map`` keys tried ONE AT A TIME (never all at once — recording - every region would sum several layers and inflate the feature width, and would - request a modality a single-tower wrapper can't serve). ``[None]`` means the - model has no recording concept (e.g. a behavioral candidate) — probe as-is. + The benchmark's own ``.region`` if it exposes one; otherwise the regions of + the neural output channels the benchmark requests, or that the subject + declares, tried ONE AT A TIME (never all at once — recording every region + would sum several layers and inflate the feature width, and would request a + modality a single-tower wrapper can't serve). A model's ``region_layer_map`` + order is kept when it has one. ``[None]`` means no neural output is declared + (e.g. a behavioral candidate) — nothing to probe. """ region = getattr(benchmark, 'region', None) if region is not None: return [region] region_map = getattr(model, 'region_layer_map', None) + channels = getattr(benchmark, 'requested_output_channels', None) + if channels is None: + channels = model.out_channels + regions = [channel.split(':', 1)[1] for channel in sorted(channels) + if channel.startswith('neural:')] if region_map: - return list(region_map.keys()) - if isinstance(model, Subject) and not isinstance(model, UnifiedModel): - channels = getattr(benchmark, 'requested_output_channels', None) - if channels is None: - channels = model.out_channels - return [channel.split(':', 1)[1] for channel in sorted(channels) - if channel.startswith('neural:')] or [None] - return [None] + declared = set(regions) + regions = ([r for r in region_map if r in declared] + + [r for r in regions if r not in region_map]) + return regions or [None] @@ -439,26 +442,19 @@ def _probe_feature_dim(model, benchmark, stimulus_set, except Exception: one = stimulus_set one = _isolate_probe_identifier(one, stimulus_set) - can_record = hasattr(model, 'start_recording') + from .streaming_helpers import neural_response time_bins = getattr(benchmark, 'timebins', None) regions = ([recording_target] if recording_target is not None else _probe_regions(model, benchmark)) last_error = None for region in regions: - if region is not None and can_record: - try: - model.start_recording(region, time_bins=time_bins) - except Exception as e: - last_error = e - continue + # One session through interact, the same way every subject is + # evaluated; no probing of the subject's methods or class. try: - if isinstance(model, Subject) and not isinstance(model, UnifiedModel): - from .streaming_helpers import neural_response - if region is None: - raise ValueError('Feature-width probe requires a neural output channel') - result = neural_response(model, one, record=region, time_bins=time_bins) - else: - result = model.process(one) + if region is None: + raise ValueError('Feature-width probe requires a neural output ' + 'channel (neural:)') + result = neural_response(model, one, record=region, time_bins=time_bins) except Exception as e: last_error = e continue diff --git a/brainscore_core/model_interface.py b/brainscore_core/model_interface.py index 9baa03cd..e2d0ccab 100644 --- a/brainscore_core/model_interface.py +++ b/brainscore_core/model_interface.py @@ -35,4 +35,5 @@ _promote_to_selector, ) from .contract import Subject, TaskContext, UnifiedModel +from .extractor import Extractor # noqa: F401 (public protocol) from .brainscore_model import BrainScoreModel diff --git a/brainscore_core/session_view.py b/brainscore_core/session_view.py new file mode 100644 index 00000000..32003429 --- /dev/null +++ b/brainscore_core/session_view.py @@ -0,0 +1,86 @@ +"""A session view that delivers only the streams a subject declares. + +Contract Rules v3, rule 5.5: the benchmark sends every stream its human +participants received; a stream the subject cannot take in never enters it. +Every shipped subject (``BrainScoreModel`` and the domain adapters) wraps the +sessions it runs in this view with its ``in_channels``, so ignoring an input +is implicit: no stub, no placeholder, and no change to the experiment, which +keeps its own inputs and responses. A subject written against ``Subject`` +directly reads only what it declares. +""" + +from .streaming import Session, StreamEvent + + +def declared_only(stimulus_set, channels): + """``stimulus_set`` without the columns of input modalities not in ``channels``. + + Returns the same object when nothing has to be dropped, so identity (and + the activation-cache key) survives. Never mutates the input table; a + projection is a new table that keeps the stimulus set's identifier and + paths. ``None`` passes through. + """ + if stimulus_set is None: + return None + from .streaming_helpers import _channel_for_stimulus_column, _preserve_stimulus_set_type + declared = frozenset(channels) + + def is_declared(channel): + return channel in declared or channel.split(':', 1)[0] in declared + + omitted = [] + for column in list(getattr(stimulus_set, 'columns', [])): + channel = _channel_for_stimulus_column(column) + if channel is not None and not is_declared(channel): + omitted.append(column) + if not omitted: + return stimulus_set + return _preserve_stimulus_set_type(stimulus_set, stimulus_set.drop(columns=omitted)) + + +class DeclaredInputSession(Session): + """Expose ``session`` to a subject that declares ``channels``. + + Events whose channel (or channel family) is not declared are skipped. + Non-event inputs, such as typed environment steps, pass through: their + stream is declared and checked at pre-flight. A stimulus table loses the + columns of undeclared input modalities, and keeps its identity when there + is nothing to drop. Measurement settings a facade attached to the session + travel with it. + """ + + def __init__(self, session: Session, channels) -> None: + self._session = session + self._channels = frozenset(channels) + for name in Session.context_fields: + setattr(self, name, getattr(session, name, getattr(Session, name))) + # Typed session subclasses carry their request as an attribute too. + for name in ('record', 'state_change'): + value = getattr(session, name, None) + if value is not None: + setattr(self, name, value) + if self.stimulus_set is not None: + self.stimulus_set = declared_only(self.stimulus_set, self._channels) + + def _declared(self, channel: str) -> bool: + return channel in self._channels or channel.split(':', 1)[0] in self._channels + + @property + def complete(self) -> bool: + return self._session.complete + + @property + def emitted(self): + return getattr(self._session, 'emitted', []) + + def next_input(self): + while (event := self._session.next_input()) is not None: + if not isinstance(event, StreamEvent) or self._declared(event.channel): + return event + return None + + def respond(self, event: StreamEvent) -> None: + self._session.respond(event) + + def collect(self, channel=None): + return self._session.collect(channel) diff --git a/brainscore_core/streaming.py b/brainscore_core/streaming.py index 9f732134..261aa879 100644 --- a/brainscore_core/streaming.py +++ b/brainscore_core/streaming.py @@ -72,8 +72,11 @@ class Session(ABC): task_context: Any = None time_bins: Any = None streaming = False + # Per-measurement settings a legacy facade attaches (trial count, variance + # request, recording type, legacy text). None when the session has none. + settings: Any = None context_fields = ('requested_output_channels', 'stimulus_set', 'task_context', - 'time_bins', 'streaming') + 'time_bins', 'streaming', 'settings') def collect(self, channel: Optional[str] = None) -> Any: """Return emitted events; specialized sources may assemble their payloads.""" @@ -107,6 +110,44 @@ def emit(self, event: StreamEvent) -> None: self.respond(event) +def _validate_inputs(inputs) -> None: + """Check known inputs against the channel catalog before any is delivered. + + Contract Rules v3, rule 3.4: a session whose inputs are known up front + fails at construction, not at the first event the subject reads. A wrong + direction, a bad address or a failed family validator is an error. An + unregistered family (allowed in exploratory experiments; benchmark scoring + rejects it at pre-flight), the catalog's documented ndim/dtype hints and a + payload whose type differs from the family's ``payload_type`` only warn. + """ + import warnings + from . import io_catalog + problems, notes = [], [] + for event in inputs: + if not isinstance(event, StreamEvent): + continue # e.g. replayed EnvironmentSteps; validated by their harness + family = event.channel.split(":", 1)[0] + if not io_catalog.has(family): + notes.append(f"{event.channel}: unregistered channel family") + continue + notes.extend(io_catalog.payload_type_notes(event.channel, event.payload)) + found = io_catalog.validate(event.channel, event.payload, + direction=io_catalog.INPUT) + if not found: + continue + documented = io_catalog.check_payload(event.channel, event.payload) + notes.extend(m for m in found if m in documented) + problems.extend(m for m in found if m not in documented) + if problems: + shown = sorted(set(problems)) + more = f" (+{len(shown) - 5} more)" if len(shown) > 5 else "" + raise ValueError("Session inputs failed catalog validation: " + + "; ".join(shown[:5]) + more) + if notes: + warnings.warn("Session inputs not fully covered by the channel catalog: " + + "; ".join(sorted(set(notes))[:5]), stacklevel=3) + + class BufferedSession(Session): """Deliver fixed inputs and collect responses without environment feedback.""" @@ -123,6 +164,7 @@ def __init__(self, inputs: Iterable[StreamEvent] = (), *, elif task_context is not None and not inputs: stimulus_set = _behavior_stimuli_to_score(task_context) inputs = _behavior_events(task_context, stimulus_set) + _validate_inputs(inputs) if requested_output_channels is not None: if isinstance(requested_output_channels, str): raise TypeError('requested_output_channels must be a sequence of channel names.') diff --git a/brainscore_core/streaming_helpers.py b/brainscore_core/streaming_helpers.py index 67ad6fc6..4dc2071a 100644 --- a/brainscore_core/streaming_helpers.py +++ b/brainscore_core/streaming_helpers.py @@ -12,7 +12,7 @@ import numpy as np -from .contract import Subject, UnifiedModel +from .contract import Subject, UnifiedModel # noqa: F401 (re-exported) from .io_catalog import modalities_to_input_channels from .streaming import BufferedSession, Session, StreamEvent, parse_channel from .events import ( @@ -375,11 +375,7 @@ def neural_response(subject, stimulus_set, record: Union[str, Sequence[str]] = " " subject.start_recording(['V4', 'IT']); assembly = subject.process(stimuli)") session = stimulus_session(stimulus_set, record=record, time_bins=time_bins) - if _has_native_interact(subject): - subject.interact(session) - else: - _drive_via_process(subject, session, stimulus_set, record=record, - time_bins=time_bins) + subject.interact(session) return session.collect(f"neural:{record}") @@ -388,19 +384,13 @@ def behavioral_response(subject, task_context) -> BehavioralAssembly: ``BehavioralAssembly`` -- the model's output, not a Score. """ session = behavior_session(task_context) - if _has_native_interact(subject): - subject.interact(session) - else: - _drive_behavior_via_process(subject, session, task_context) + subject.interact(session) return session.collect("behavior") def apply_state_change(subject, state_change: StateChange): session = state_change_session(state_change) - if _has_native_interact(subject): - subject.interact(session) - else: - _drive_state_change_via_process(subject, session, state_change) + subject.interact(session) return session.collect("perturbation") @@ -422,10 +412,7 @@ def run_environment(subject, environment, *, seed=None, options=None, close() raise with session: - if _has_native_interact(subject): - subject.interact(session) - else: - _drive_environment_via_process(subject, session) + subject.interact(session) if not session.complete: raise RuntimeError("Environment session ended before episode completion") return session.collect("motor") @@ -445,12 +432,19 @@ def _drive_via_process(subject, session: StimulusSetSession, stimulus_set, def _drive_neural_session_via_process( - subject, session, driver: str = "interact" + subject, session, driver: str = "interact", allow_multi_modality: bool = False ) -> None: + """Run a neural session through ``start_recording`` + ``process``. + + ``allow_multi_modality`` is the subject's own statement that its + ``process`` combines several input modalities; when several are present + it is called with ``multi_modality=True``. + """ + _refuse_variance(subject, session) input_events = _drain_stream_events(session) stimuli = _stimuli_from_stream_session(session, input_events) output_t_ms = input_events[-1].t_ms if input_events else 0.0 - use_multi_modality = _should_use_multi_modality(subject, input_events) + use_multi_modality = allow_multi_modality and _spans_several_inputs(input_events) time_bins = getattr(session, "time_bins", None) channel_regions = [] @@ -681,9 +675,7 @@ def _stimuli_from_stream_session(session, events: list[StreamEvent]): return _reconstruct_stimulus_set_from_events(events) -def _should_use_multi_modality(subject, events: list[StreamEvent]) -> bool: - if not hasattr(subject, "_process_multi_modality"): - return False +def _spans_several_inputs(events: list[StreamEvent]) -> bool: families = set() for event in events: family, _ = parse_channel(event.channel) @@ -775,9 +767,40 @@ def _requested_output_channels(session) -> list[str]: ) -def _has_native_interact(subject) -> bool: - interact = getattr(type(subject), "interact", None) - return callable(interact) and interact not in (Subject.interact, UnifiedModel.interact) +def drive_typed_session(subject, session) -> None: + """Run a session through a typed model's ``process`` / ``start_*`` paths. + + Routes on the session's requested output channels: ``behavior``, + ``perturbation``, ``motor``, or any set of ``neural:`` channels. + """ + requested_channels = _requested_output_channels(session) + streaming = getattr(session, "streaming", False) + if requested_channels == ["behavior"]: + if streaming: + _drive_behavior_session_streaming(subject, session) + else: + _drive_behavior_session_via_process(subject, session) + return + if requested_channels == ["perturbation"]: + _drive_state_change_session_via_process(subject, session) + return + if requested_channels == ["motor"]: + _drive_environment_session_via_process(subject, session) + return + if requested_channels and all(channel.startswith("neural:") + for channel in requested_channels): + # A session may ask to be consumed one input at a time (inputs generated + # on demand rather than known up front). Same scores, different arrival. + if streaming: + _drive_neural_session_streaming(subject, session) + else: + _drive_neural_session_via_process(subject, session) + return + raise NotImplementedError( + f"Subject {getattr(subject, 'identifier', type(subject).__name__)!r} " + "runs sessions requesting neural:, behavior, perturbation, or " + f"motor output channels; got {requested_channels!r}." + ) def _drive_environment_via_process(subject, session: EnvironmentSession) -> None: @@ -868,6 +891,28 @@ def _behavior_stimuli_from_session(session, task_context): return _behavior_stimuli_to_score(task_context) +def measurement_settings(session) -> dict: + """Per-measurement settings a legacy facade attached to ``session`` (trial count, variance request).""" + return dict(getattr(session, "settings", None) or {}) + + +def _refuse_variance(subject, session) -> None: + """The typed paths are deterministic: a request for trial-to-trial variance fails. + + A legacy ``look_at(require_variance=True)`` asks for repeated noisy + presentations. Returning identical trials would be a silently corrected + response (rule 6.2), so both the neural and the behavioral typed drivers + refuse it before any compute. + """ + settings = measurement_settings(session) + if settings.get("require_variance"): + raise NotImplementedError( + f"Subject {getattr(subject, 'identifier', type(subject).__name__)!r} " + "cannot produce trial-to-trial variance (look_at(require_variance=True, " + f"number_of_trials={settings.get('number_of_trials', 1)})). Legacy models " + "add it with microsaccades; this subject has no equivalent.") + + def _drive_behavior_session_via_process( subject, session: BehavioralSession, task_context=None, driver: str = "interact" @@ -879,6 +924,7 @@ def _drive_behavior_session_via_process( raise ValueError( "behavior interact requires a session.task_context to drive." ) + _refuse_variance(subject, session) subject.start_task(task_context) stimuli = _behavior_stimuli_from_session(session, task_context) if stimuli is None: diff --git a/tests/test_channel_compatibility.py b/tests/test_channel_compatibility.py index de9a7652..b2959c4a 100644 --- a/tests/test_channel_compatibility.py +++ b/tests/test_channel_compatibility.py @@ -80,7 +80,7 @@ def test_channel_compatible_pair_passes(): check_channel_compatibility(subject, benchmark) -def test_missing_required_input_channel_raises_with_channel_name(): +def test_partial_input_overlap_runs_and_no_overlap_raises(): subject = _Subject( in_channels={"vision"}, out_channels={"neural:IT"}, @@ -90,8 +90,17 @@ def test_missing_required_input_channel_raises_with_channel_name(): requested_output_channels={"neural:IT"}, ) + # Contract v3 rule 5.5: partial matches run; the dropped stream is reported. + match = check_channel_compatibility(subject, benchmark) + assert match.kind == "partial" + assert match.dropped_inputs == {"text"} + + no_overlap = _Benchmark( + required_input_channels={"text"}, + requested_output_channels={"neural:IT"}, + ) with pytest.raises(CompatibilityError, match="text"): - check_channel_compatibility(subject, benchmark) + check_channel_compatibility(subject, no_overlap) def test_missing_requested_output_channel_raises_with_channel_name(): diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index 1479a88d..a77f56b5 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -73,14 +73,25 @@ def test_superset_modalities_passes(self): def test_missing_required_raises(self): model = FakeModel(modalities={'text'}) bench = FakeBenchmark(required_modalities={'vision'}) - with pytest.raises(CompatibilityError, match="does not support modalities required"): + with pytest.raises(CompatibilityError, match="takes in none of the inputs"): check_compatibility(model, bench) - def test_missing_one_of_multiple_required_raises(self): + def test_missing_one_of_multiple_required_runs_as_partial(self): + # Contract v3 rule 5: a subject that takes in some of the inputs runs; + # the rest never enter it, and the match says so. model = FakeModel(modalities={'vision'}) bench = FakeBenchmark(required_modalities={'vision', 'text'}) - with pytest.raises(CompatibilityError, match="'text'"): - check_compatibility(model, bench) + match = check_compatibility(model, bench) + assert match.kind == 'partial' + assert match.consumed_inputs == {'vision'} + assert match.dropped_inputs == {'text'} + + def test_full_match_reports_full(self): + match = check_compatibility(FakeModel(modalities={'vision', 'text'}), + FakeBenchmark(required_modalities={'vision'})) + assert match.kind == 'full' + assert match.dropped_inputs == frozenset() + assert match.as_attrs()['match'] == 'full' def test_no_required_modalities_on_benchmark(self): model = FakeModel(modalities={'vision'}) @@ -99,7 +110,7 @@ def test_any_one_accepted_passes(self): def test_none_accepted_raises(self): bench = FakeBenchmark(accepted_modalities={'video', 'vision'}) - with pytest.raises(CompatibilityError, match="provides none of the input formats"): + with pytest.raises(CompatibilityError, match="takes in none of the inputs"): check_compatibility(FakeModel(modalities={'text'}), bench) def test_required_plus_accepted_combo(self): @@ -108,10 +119,11 @@ def test_required_plus_accepted_combo(self): accepted_modalities={'video', 'vision'}) check_compatibility(FakeModel(modalities={'audio', 'video'}), bench) check_compatibility(FakeModel(modalities={'audio', 'vision'}), bench) - with pytest.raises(CompatibilityError, match="does not support modalities required"): - check_compatibility(FakeModel(modalities={'video'}), bench) # missing audio - with pytest.raises(CompatibilityError, match="provides none of the input formats"): - check_compatibility(FakeModel(modalities={'audio'}), bench) # neither video nor vision + # Missing audio or missing both visual formats: partial, still runs. + assert check_compatibility(FakeModel(modalities={'video'}), bench).dropped_inputs == {'audio'} + assert check_compatibility(FakeModel(modalities={'audio'}), bench).dropped_inputs == {'vision'} + with pytest.raises(CompatibilityError, match="takes in none of the inputs"): + check_compatibility(FakeModel(modalities={'text'}), bench) def test_model_hard_requires_accepted_format_passes(self): # Check-2 fix: a model hard-requiring video isn't rejected when the @@ -235,14 +247,14 @@ def test_full_compatible_model(self): warnings.simplefilter('error') check_compatibility(model, bench) - def test_required_fails_before_region_check(self): - """Required modality failure should be raised even if region is also bad.""" + def test_missing_output_reported_before_inputs(self): + """A missing requested output is reported first, even if inputs also fail.""" model = FakeModel(modalities={'text'}, region_layer_map={}) bench = FakeBenchmark( required_modalities={'vision'}, region='IT', ) - with pytest.raises(CompatibilityError, match="does not support modalities required"): + with pytest.raises(CompatibilityError, match="no layer mapping for region 'IT'"): check_compatibility(model, bench) def test_error_message_includes_identifiers(self): @@ -344,3 +356,31 @@ def test_legacy_model_available_defaults_to_supported(self): def test_legacy_model_required_defaults_to_empty(self): model = FakeModel(modalities={'vision'}) assert model.required_modalities == set() + + +class TestDeclaredChannelsRun: + """Contract v3 rule 2.4: every declared channel passes pre-flight and runs.""" + + def _model(self): + from brainscore_core.model_interface import BrainScoreModel + return BrainScoreModel( + 'all-parts', region_layer_map={'IT': 'layer'}, + extractor={'vision': lambda stimuli, layers: None}, + behavioral_readout_layer='layer', + action_fn=lambda step: None, + state_change_fn=lambda change: (None, lambda: None)) + + def test_every_part_declares_registered_channels(self): + model = self._model() + assert model.out_channels == {'neural:IT', 'behavior', 'motor', 'perturbation'} + assert {'vision', 'observation', 'lesion', 'stimulation'} <= model.in_channels + for output in sorted(model.out_channels): + bench = FakeBenchmark(identifier=output) + bench.requested_output_channels = {output} + bench.required_input_channels = {'vision'} + check_compatibility(model, bench) # registered and produced + + def test_every_declared_output_has_one_session_handler(self): + model = self._model() + for output in sorted(model.out_channels): + assert model.check_session_support({output}) is not None diff --git a/tests/test_constructor_ergonomics.py b/tests/test_constructor_ergonomics.py index 7a8ce95b..d36115f6 100644 --- a/tests/test_constructor_ergonomics.py +++ b/tests/test_constructor_ergonomics.py @@ -42,7 +42,7 @@ def test_positional_capability_tail_warns(): def test_keyword_calls_do_not_warn(): with warnings.catch_warnings(): warnings.simplefilter('error', DeprecationWarning) - BrainScoreModel('id', None, {}, {}, None, action_fn=_f) # must not raise + BrainScoreModel('id', None, {}, {}, None, policy=_f) # must not raise def test_identifier_only_construction_works(): diff --git a/tests/test_contract_drift.py b/tests/test_contract_drift.py index 423ba717..aecdc217 100644 --- a/tests/test_contract_drift.py +++ b/tests/test_contract_drift.py @@ -20,6 +20,7 @@ "lesion": (io_catalog.INPUT, "core"), "motor": (io_catalog.OUTPUT, "environment harness"), "neural": (io_catalog.OUTPUT, "core"), + "perturbation": (io_catalog.OUTPUT, "core"), "observation": (io_catalog.INPUT, "environment harness"), "pharmacological": (io_catalog.INPUT, "core"), "proprioception": (io_catalog.INPUT, "environment harness"), diff --git a/tests/test_dispatch_cache.py b/tests/test_dispatch_cache.py index 15f64b04..71f896cc 100644 --- a/tests/test_dispatch_cache.py +++ b/tests/test_dispatch_cache.py @@ -19,7 +19,9 @@ class _FakeOwner: def __init__(self, perturbed): self._activations_model = _FakeActivationsModel() self._extractors = {'vision': self._activations_model} - self._active_perturbations = {'h1': (lambda: None)} if perturbed else {} + from types import SimpleNamespace + self._perturbations = SimpleNamespace( + active_perturbations={'h1': (lambda: None)} if perturbed else {}) self._preprocessors = {} self._model = None self._recording_layer = None diff --git a/tests/test_environment_contract.py b/tests/test_environment_contract.py index aa7808f1..ea40bacf 100644 --- a/tests/test_environment_contract.py +++ b/tests/test_environment_contract.py @@ -66,9 +66,17 @@ def test_terminal_observation_is_retained_without_a_policy_call(): session.emit(StreamEvent('motor', 1, None)) +def _typed_subject(process): + """A minimal typed subject: process() plus the shared typed session driver.""" + from brainscore_core.streaming_helpers import drive_typed_session + subject = SimpleNamespace(process=process) + subject.interact = lambda session: drive_typed_session(subject, session) + return subject + + def test_terminal_reset_does_not_call_policy(): env = Environment(terminal_on_reset=True) - subject = SimpleNamespace(process=lambda step: pytest.fail('Terminal input reached policy')) + subject = _typed_subject(lambda step: pytest.fail('Terminal input reached policy')) assert run_environment(subject, env) == [] assert env.actions == [] and env.closed == 1 @@ -101,7 +109,7 @@ def test_failed_step_is_unknown_execution_never_success(error): def test_bad_observation_fails_before_inference_and_closes(): env = Environment() env.observation = lambda **kwargs: EnvironmentStep(observation={'image': [1], 'task': 'move'}) - subject = SimpleNamespace(process=lambda step: pytest.fail('Bad observation reached policy')) + subject = _typed_subject(lambda step: pytest.fail('Bad observation reached policy')) with pytest.raises(ValueError, match='image'): run_environment(subject, env) assert env.actions == [] and env.closed == 1 diff --git a/tests/test_io_catalog.py b/tests/test_io_catalog.py index d039c9d2..2c50347b 100644 --- a/tests/test_io_catalog.py +++ b/tests/test_io_catalog.py @@ -94,6 +94,44 @@ def test_register_and_get_roundtrip(self, monkeypatch): payload_contract="(time,) float diameter in mm", handled_by="an eye-tracker harness", ) - io_catalog.register(entry) + io_catalog.register(entry, replace=True) # 'pupil' is a seeded family assert io_catalog.get("pupil") is entry assert io_catalog.get("pupil:left").name == "pupil" + + def test_register_rejects_silent_overwrite(self, monkeypatch): + monkeypatch.setattr(io_catalog, "_CATALOG", dict(io_catalog._CATALOG)) + original = io_catalog.get("pupil") + clash = CatalogEntry( + name="pupil", kind=OUTPUT, carried_by="x", payload_contract="x", + handled_by="another harness", + ) + with pytest.raises(ValueError, match="already registered"): + io_catalog.register(clash) + assert io_catalog.get("pupil") is original + io_catalog.register(original) # same object: allowed, no change + + +class TestPayloadTypes: + """Contract v3 rule 3.2: a family declares its payload type; mismatches warn first.""" + + def test_seeded_families_declare_payload_types(self): + from pathlib import Path + import numpy as np + from brainscore_core.events import EnvironmentStep, StateChange + assert io_catalog.get("text").payload_type is str + assert io_catalog.get("vision").payload_type == (str, Path, np.ndarray) + assert io_catalog.get("lesion").payload_type is StateChange + assert io_catalog.get("observation").payload_type is EnvironmentStep + assert io_catalog.get("pupil").payload_type is None # unchecked families stay unchecked + + def test_payload_type_notes_report_a_mismatch_and_nothing_else(self): + assert io_catalog.payload_type_notes("text", "a sentence") == [] + assert io_catalog.payload_type_notes("vision:left", "/tmp/a.png") == [] + assert io_catalog.payload_type_notes("pupil", object()) == [] + notes = io_catalog.payload_type_notes("text", 42) + assert notes == ["text: payload is int, not str (a Python str)"] + + def test_type_mismatch_never_enters_validate_failures(self): + # Warn-first: validate() stays as it was, so callers that raise on its + # failures do not start rejecting payloads of an unexpected type. + assert io_catalog.validate("text", 42, direction=INPUT) == [] diff --git a/tests/test_io_catalog_preflight.py b/tests/test_io_catalog_preflight.py index a3b30cc1..144ad6c1 100644 --- a/tests/test_io_catalog_preflight.py +++ b/tests/test_io_catalog_preflight.py @@ -11,6 +11,7 @@ from brainscore_core.model_interface import UnifiedModel from brainscore_core.compatibility import ( check_compatibility, + CompatibilityError, CompatibilityWarning, ) @@ -55,19 +56,19 @@ def test_documented_input_no_warning(self): warnings.simplefilter("error", CompatibilityWarning) check_compatibility(m, b) # vision is documented; must not warn - def test_undocumented_input_warns(self): + def test_undocumented_input_is_rejected(self): + # Contract v3 rule 3.2: every channel family is registered. An + # undocumented input is a compatibility error, not a warning. m = _Model({"vision", "telepathy"}) b = _Bench({"telepathy"}) - with pytest.warns(CompatibilityWarning, match="not in the Input/Output Catalog"): + with pytest.raises(CompatibilityError, match="unknown channel 'telepathy'"): check_compatibility(m, b) - def test_catalog_check_is_warn_only(self): - # An undocumented modality must NOT raise CompatibilityError; it only warns. - m = _Model({"telepathy"}) - b = _Bench({"telepathy"}) - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - check_compatibility(m, b) # no exception + def test_catalog_documentation_check_is_warn_only(self): + # check_io_catalog itself only warns; it never raises. + from brainscore_core.compatibility import check_io_catalog + with pytest.warns(CompatibilityWarning, match="not in the Input/Output Catalog"): + check_io_catalog(_Model({"telepathy"}), _Bench({"telepathy"})) def test_neural_region_is_documented(self): m = _Model({"vision"}, region_map={"IT": "layer.10"}) diff --git a/tests/test_legacy_facade.py b/tests/test_legacy_facade.py new file mode 100644 index 00000000..a8f5807e --- /dev/null +++ b/tests/test_legacy_facade.py @@ -0,0 +1,472 @@ +"""Fidelity tests for the legacy-benchmark facade (Contract Rules v3, rule 11.2). + +Each legacy method a vision or language benchmark calls on a candidate is +checked against the legacy behavior it stands in for. +""" +import numpy as np +import pandas as pd +import pytest + +from brainscore_core.legacy_facade import ( + LegacyBenchmarkFacade, + repeat_over_time_bins, + text_stimulus_set, +) +from brainscore_core.model_interface import BrainScoreModel +from brainscore_core.supported_data_standards.brainio.assemblies import NeuroidAssembly +from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet + + +class _Extractor: + """Deterministic features: one neuroid per character of the stimulus id.""" + + def __init__(self): + self.calls = [] + self.seen = [] + + def __call__(self, stimuli, layers=None, **kwargs): + self.calls.append(list(stimuli['stimulus_id'])) + self.seen.append(stimuli) + n = len(stimuli) + return NeuroidAssembly( + np.arange(n * 3, dtype=float).reshape(n, 3), + coords={ + 'stimulus_id': ('presentation', list(stimuli['stimulus_id'])), + 'neuroid_id': ('neuroid', np.arange(3)), + 'layer': ('neuroid', [layers[0]] * 3), + }, + dims=['presentation', 'neuroid'], + ) + + +def _model(modality='vision'): + extractor = _Extractor() + model = BrainScoreModel('facade-test', region_layer_map={'IT': 'layer'}, + extractor={modality: extractor}, visual_degrees=10) + return model, extractor + + +def _images(n=2): + stimuli = StimulusSet(pd.DataFrame({ + 'stimulus_id': [f's{i}' for i in range(n)], + 'image_path': [f'/tmp/s{i}.png' for i in range(n)], + })) + stimuli.identifier = f'images-{n}' + return stimuli + + +def test_look_at_matches_process(): + model, _ = _model() + model.start_recording('IT', time_bins=[(70, 170)]) + via_legacy = model.look_at(_images()) + via_process = model.process(_images()) + np.testing.assert_array_equal(via_legacy.values, via_process.values) + assert 'time_bin' not in via_legacy.dims # one bin: unchanged, as in legacy + + +def test_several_time_bins_repeat_a_static_response(): + model, _ = _model() + bins = [(70, 170), (170, 270)] + model.start_recording('IT', time_bins=bins) + responses = model.look_at(_images()) + assert responses.sizes['time_bin'] == 2 + assert list(responses['time_bin_start'].values) == [70, 170] + assert list(responses['time_bin_end'].values) == [170, 270] + first, second = responses.isel(time_bin=0), responses.isel(time_bin=1) + np.testing.assert_array_equal(first.values, second.values) + + +def test_number_of_trials_alone_does_not_change_deterministic_responses(): + model, _ = _model() + model.start_recording('IT', time_bins=[(70, 170)]) + once = model.look_at(_images(), number_of_trials=1) + many = model.look_at(_images(), number_of_trials=5) + np.testing.assert_array_equal(once.values, many.values) + + +def test_variance_request_fails_instead_of_returning_identical_trials(): + model, extractor = _model() + model.start_recording('IT', time_bins=[(70, 170)]) + with pytest.raises(NotImplementedError, match='trial-to-trial variance'): + model.look_at(_images(), number_of_trials=3, require_variance=True) + assert extractor.calls == [] + + +def test_visual_degrees_reported(): + model, _ = _model() + assert model.visual_degrees() == 10 + + +def test_start_behavioral_task_fails_clearly(): + model, _ = _model('text') + with pytest.raises(NotImplementedError, match="'reading_times'"): + model.start_behavioral_task('reading_times') + + +def test_digest_text_returns_neural_part_for_a_passage(): + model, extractor = _model('text') + model.start_neural_recording('IT', 'fMRI') + result = model.digest_text(['the quick brown', 'fox jumps']) + assert set(result) == {'neural'} + assert result['neural'].sizes['presentation'] == 2 + assert extractor.calls == [[0, 1]] + + +def test_text_stimulus_set_keys_cache_by_content(): + first = text_stimulus_set(['a passage']) + second = text_stimulus_set(['another passage']) + assert first.identifier != second.identifier + assert list(first['context_id']) == ['passage'] + stimuli = _images() + assert text_stimulus_set(stimuli) is stimuli + + +def test_facade_wraps_a_subject_directly(): + model, _ = _model() + facade = LegacyBenchmarkFacade(model, visual_degrees=4) + facade.start_recording('IT', time_bins=[(70, 170)]) + assert facade.identifier == 'facade-test' + assert facade.visual_degrees() == 4 + assert facade.look_at(_images()).sizes['presentation'] == 2 + + +def test_repeat_over_time_bins_keeps_existing_time_axis(): + model, _ = _model() + model.start_recording('IT', time_bins=[(0, 10), (10, 20)]) + timed = model.process(_images()) + assert repeat_over_time_bins(timed, [(0, 10), (10, 20)]) is timed + + +@pytest.mark.parametrize('time_bins', [ + [(70, 170)], + np.array([[70, 170]]), + (70, 170), + np.array([(70, 170)], dtype=object), + pd.MultiIndex.from_tuples([(70, 170)]).values, +]) +def test_single_time_bin_in_any_benchmark_form_leaves_response_unchanged(time_bins): + model, _ = _model() + model.start_recording('IT', time_bins=time_bins) + assert 'time_bin' not in model.process(_images()).dims + + +@pytest.mark.parametrize('kwargs', [ + {'activations_model': {'vision': _Extractor()}}, + {'extractor': {'vision': _Extractor()}, 'generation_fn': lambda *a: None}, + {'extractor': {'vision': _Extractor()}, 'action_fn': lambda step: None}, + {'extractor': {'vision': _Extractor()}, 'state_change_fn': lambda c: (None, lambda: None)}, +]) +def test_older_constructor_names_warn_and_keep_working(kwargs): + with pytest.warns(DeprecationWarning, match='is deprecated; use'): + model = BrainScoreModel('old-names', region_layer_map={'IT': 'layer'}, **kwargs) + assert 'neural:IT' in model.out_channels + + +def test_preferred_names_do_not_warn(): + import warnings + with warnings.catch_warnings(): + warnings.simplefilter('error', DeprecationWarning) + BrainScoreModel('new-names', region_layer_map={'IT': 'layer'}, + extractor={'vision': _Extractor()}, + answer=lambda *a: None, policy=lambda step: None, + intervention=lambda c: (None, lambda: None)) + + +# --- the facade runs any subject through interact ------------------------ + +from brainscore_core.model_interface import Subject +from brainscore_core.streaming import StreamEvent +from brainscore_core.supported_data_standards.brainio.assemblies import BehavioralAssembly + + +class _Recorder(Subject): + """An interact-only subject: no process, look_at or start_recording.""" + + identifier = 'recording-subject' + in_channels = {'text', 'vision'} + out_channels = {'neural:IT', 'behavior'} + required_channels = set() + + def __init__(self): + self.sessions = [] + self.resets = 0 + + def interact(self, session): + self.sessions.append(session) + while session.next_input() is not None: + pass + n = len(session.stimulus_set) + for channel in session.requested_output_channels: + neural = channel.startswith('neural:') + cls = NeuroidAssembly if neural else BehavioralAssembly + payload = cls(np.arange(n)[:, None], + dims=['presentation', 'neuroid' if neural else 'choice']) + session.respond(StreamEvent(channel, payload, None)) + + def reset(self): + self.resets += 1 + + +def test_facade_runs_an_interact_only_subject_on_vision_calls(): + subject = _Recorder() + facade = LegacyBenchmarkFacade(subject, visual_degrees=12) + facade.start_recording('IT', time_bins=[(70, 170), (170, 270)]) + out = facade.look_at(_images(), number_of_trials=7) + np.testing.assert_array_equal(out.values, [[0], [1]]) + session, = subject.sessions + assert session.requested_output_channels == ('neural:IT',) + assert session.time_bins == [(70, 170), (170, 270)] + assert session.settings['number_of_trials'] == 7 + assert facade.visual_degrees() == 12 + assert facade.identifier == 'recording-subject' + + +def test_facade_runs_language_neural_and_behavior_in_one_session(): + subject = _Recorder() + facade = LegacyBenchmarkFacade(subject) + facade.start_neural_recording('IT', recording_type='spikes') + facade.start_behavioral_task('reading_times') + text = np.array(['one', 'two']) + out = facade.digest_text(text) + assert set(out) == {'neural', 'behavior'} + session, = subject.sessions + assert set(session.requested_output_channels) == {'neural:IT', 'behavior'} + assert session.task_context.task_type == 'reading_times' + assert session.settings['recording_type'] == 'spikes' + assert session.settings['legacy_text'] is text + + +def test_facade_passes_a_text_stimulus_set_unchanged(): + subject = _Recorder() + facade = LegacyBenchmarkFacade(subject) + facade.start_neural_recording('IT') + text = StimulusSet({'sentence': ['a', 'b'], 'stimulus_id': ['x', 'y'], + 'context_id': ['p', 'p']}) + facade.digest_text(text) + session, = subject.sessions + assert session.stimulus_set is text + assert 'legacy_text' not in session.task_context.metadata + + +def test_behavioral_task_needs_a_declared_behavior_output(): + class Neural(_Recorder): + out_channels = {'neural:IT'} + + with pytest.raises(NotImplementedError, match="declares no 'behavior' output"): + LegacyBenchmarkFacade(Neural()).start_behavioral_task('reading_times') + + +def test_facade_without_visual_field_fails_instead_of_inventing_one(): + with pytest.raises(NotImplementedError, match='declares no visual field'): + LegacyBenchmarkFacade(_Recorder()).visual_degrees() + + +def test_look_at_needs_a_recording_and_one_region(): + facade = LegacyBenchmarkFacade(_Recorder()) + with pytest.raises(ValueError, match='start_recording'): + facade.look_at(_images()) + facade.start_recording(['IT', 'V4']) + with pytest.raises(ValueError, match='one region'): + facade.look_at(_images()) + with pytest.raises(TypeError, match='unsupported settings'): + facade.look_at(_images(), fixation=True) + + +def test_facade_reset_clears_settings_and_resets_subject(): + subject = _Recorder() + facade = LegacyBenchmarkFacade(subject) + facade.start_recording('IT') + facade.reset() + assert subject.resets == 1 + with pytest.raises(ValueError, match='start_recording'): + facade.look_at(_images()) + + +def test_typed_look_at_runs_through_interact(): + model, extractor = _model() + sessions = [] + original = model.interact + model.interact = lambda session: (sessions.append(session), original(session))[1] + model.start_recording('IT', time_bins=[(70, 170)]) + via_legacy = model.look_at(_images()) + assert len(sessions) == 1 + assert sessions[0].requested_output_channels == ('neural:IT',) + np.testing.assert_array_equal(via_legacy.values, model.process(_images()).values) + + +def test_typed_behavioral_look_at_matches_process(): + extractor = _Extractor() + model = BrainScoreModel('facade-behavior', region_layer_map={'IT': 'layer'}, + extractor={'vision': extractor}, + behavioral_readout_layer='layer') + fitting = StimulusSet(pd.DataFrame({ + 'stimulus_id': ['f0', 'f1', 'f2', 'f3'], + 'image_path': ['/tmp/f0.png', '/tmp/f1.png', '/tmp/f2.png', '/tmp/f3.png'], + 'image_label': ['cat', 'dog', 'cat', 'dog'], + })) + fitting.identifier = 'fitting' + model.start_task('probabilities', fitting_stimuli=fitting) + via_legacy = model.look_at(_images()) + via_process = model.process(_images()) + assert isinstance(via_legacy, BehavioralAssembly) + np.testing.assert_array_equal(via_legacy.values, via_process.values) + + +# --- applying the facade at the scoring boundary -------------------------- + +def test_domain_loaders_mark_legacy_benchmarks_and_scoring_wraps_them(): + from types import SimpleNamespace + from brainscore_core.legacy_facade import mark_legacy_benchmark, subject_for_benchmark + + legacy = SimpleNamespace(identifier='legacy') + assert mark_legacy_benchmark(legacy).subject_api == 'legacy' + session_native = SimpleNamespace(identifier='native', required_input_channels={'vision'}) + assert mark_legacy_benchmark(session_native).subject_api == 'session' + typed = SimpleNamespace(identifier='typed', subject_api='typed') + assert mark_legacy_benchmark(typed).subject_api == 'typed' + + subject = _Recorder() + facade = subject_for_benchmark(subject, legacy, visual_degrees=8) + assert isinstance(facade, LegacyBenchmarkFacade) + assert facade.subject is subject and facade.visual_degrees() == 8 + assert subject_for_benchmark(subject, session_native) is subject + assert subject_for_benchmark(subject, typed) is subject + + +def test_declared_visual_degrees_reads_the_legacy_vision_setting_or_none(): + from brainscore_core.legacy_facade import declared_visual_degrees + model, _ = _model() + assert declared_visual_degrees(model) == 10 + assert declared_visual_degrees(_Recorder()) is None + + +# --- setup once, measure many times (review findings 1 to 4) --------------- + +from brainscore_core.model_interface import TaskContext + +def _labelled_images(n=4, with_text=False): + frame = {'stimulus_id': [f'f{i}' for i in range(n)], + 'image_path': [f'/tmp/f{i}.png' for i in range(n)], + 'image_label': ['cat', 'dog'] * (n // 2)} + if with_text: + frame['sentence'] = [f'caption {i}' for i in range(n)] + stimuli = StimulusSet(pd.DataFrame(frame)) + stimuli.identifier = 'fitting-images' + return stimuli + + +def _readout_model(): + extractor = _Extractor() + model = BrainScoreModel('readout', region_layer_map={'IT': 'layer'}, + extractor={'vision': extractor}, + behavioral_readout_layer='layer') + return model, extractor + + +def _fits(extractor): + """How many times the fitting stimuli were extracted (one per readout fit).""" + return sum(1 for ids in extractor.calls if ids and str(ids[0]).startswith('f')) + + +def test_fitting_settings_stay_apart_from_measurement_settings(): + subject = _Recorder() + facade = LegacyBenchmarkFacade(subject) + fitting = _labelled_images() + facade.start_task('probabilities', fitting_stimuli=fitting, + number_of_trials=2, require_variance=True) + facade.look_at(_images(), number_of_trials=1) + facade.look_at(_images(), number_of_trials=3) + first, second = subject.sessions + assert first.task_context is second.task_context # one task object, sent each time + assert first.task_context.metadata['task_settings'] == { + 'number_of_trials': 2, 'require_variance': True} + assert first.task_context.fitting_stimuli is fitting + assert first.settings == {'number_of_trials': 1, 'require_variance': False} + assert second.settings == {'number_of_trials': 3, 'require_variance': False} + + +def test_typed_behavioral_look_at_fits_the_readout_once(): + model, extractor = _readout_model() + model.start_task('probabilities', fitting_stimuli=_labelled_images()) + assert _fits(extractor) == 1 + model.look_at(_images()) + model.look_at(_images()) + assert _fits(extractor) == 1 + + +def test_scored_facade_fits_the_readout_once_across_measurements(): + model, extractor = _readout_model() + facade = LegacyBenchmarkFacade(model, visual_degrees=8) + facade.start_task('probabilities', fitting_stimuli=_labelled_images()) + facade.look_at(_images()) + facade.look_at(_images()) + assert _fits(extractor) == 1 + facade.start_task('probabilities', fitting_stimuli=_labelled_images()) # a new task + facade.look_at(_images()) + assert _fits(extractor) == 2 + + +def test_repeated_task_refits_after_a_state_change(): + from brainscore_core.events import PerturbationApplied, Perturbation, Selection, StateChange + from brainscore_core.streaming_helpers import apply_state_change, behavioral_response + extractor = _Extractor() + model = BrainScoreModel( + 'lesioned', region_layer_map={'IT': 'layer'}, extractor={'vision': extractor}, + behavioral_readout_layer='layer', + intervention=lambda change: ( + PerturbationApplied(handle_id='h', target=change.target, + perturbation=change.perturbation), + lambda: None)) + task = TaskContext(task_type='probabilities', fitting_stimuli=_labelled_images(), + metadata={'stimulus_set': _images()}) + behavioral_response(model, task) + behavioral_response(model, task) + assert _fits(extractor) == 1 + apply_state_change(model, StateChange(kind='lesion', target=Selection(layer='layer'), + perturbation=Perturbation(kind='zero'))) + behavioral_response(model, task) + assert _fits(extractor) == 2 # the model changed under the readout + + +def test_fitting_stimuli_reach_the_readout_with_declared_inputs_only(): + model, extractor = _readout_model() + fitting = _labelled_images(with_text=True) + model.start_task(TaskContext(task_type='probabilities', fitting_stimuli=fitting)) + seen = extractor.seen[-1] + assert 'sentence' not in seen.columns + assert list(seen['image_label']) == list(fitting['image_label']) + assert 'sentence' in fitting.columns # the benchmark's table is untouched + assert seen.identifier == fitting.identifier + + +def test_typed_behavioral_variance_request_fails_before_compute(): + model, extractor = _readout_model() + model.start_task('probabilities', fitting_stimuli=_labelled_images()) + calls = len(extractor.calls) + with pytest.raises(NotImplementedError, match='trial-to-trial variance'): + model.look_at(_images(), number_of_trials=3, require_variance=True) + assert len(extractor.calls) == calls + + +def test_text_identifiers_do_not_collide_on_the_delimiter(): + first = text_stimulus_set(['a|b', 'c']) + second = text_stimulus_set(['a', 'b|c']) + assert first.identifier != second.identifier + + +def test_changed_fitting_data_in_the_same_task_object_refits(): + """TaskContext is mutable: an explicit start_task after a change fits again.""" + model, extractor = _readout_model() + task = TaskContext(task_type='probabilities', fitting_stimuli=_labelled_images()) + model.start_task(task) + model.start_task(task) + assert _fits(extractor) == 1 # unchanged task sent again: no refit + task.fitting_stimuli = _labelled_images() # a new table + task.fitting_stimuli['image_label'] = ['dog', 'cat', 'dog', 'cat'] + model.start_task(task) + assert _fits(extractor) == 2 + task.fitting_stimuli['image_label'] = ['cat', 'cat', 'dog', 'dog'] # relabelled in place + model.start_task(task) + assert _fits(extractor) == 3 + model.look_at(_images()) # a measurement on the unchanged task: no refit + assert _fits(extractor) == 3 diff --git a/tests/test_memory.py b/tests/test_memory.py index 4b69cc86..aad0dd3f 100644 --- a/tests/test_memory.py +++ b/tests/test_memory.py @@ -17,15 +17,20 @@ SAFETY_FACTORS, ) from brainscore_core.model_interface import UnifiedModel +from brainscore_core.supported_data_standards.brainio.assemblies import NeuroidAssembly +from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet # ── Helpers ────────────────────────────────────────────────────────── -class FakeResult: - """Mimics a DataAssembly with nbytes and shape.""" - def __init__(self, nbytes=8000, n_features=1000): - self.nbytes = nbytes - self.shape = (1, n_features) +def FakeResult(nbytes=8000, n_features=1000): + """A one-presentation NeuroidAssembly of the given width (nbytes is unused).""" + import numpy as np + del nbytes + return NeuroidAssembly(np.zeros((1, n_features), dtype='float32'), + dims=['presentation', 'neuroid'], + coords={'stimulus_id': ('presentation', [0]), + 'neuroid_id': ('neuroid', list(range(n_features)))}) class FakeModel(UnifiedModel): @@ -35,8 +40,10 @@ def __init__(self, identifier='test-model', modalities=None, activation_nbytes=8000, n_features=1000): self._id = identifier self._modalities = modalities or {'vision'} - self._rlm = region_layer_map or {} - self._process_result = process_result or FakeResult(nbytes=activation_nbytes, n_features=n_features) + # A model that records somewhere; the probe reads it from the declared output. + self._rlm = region_layer_map if region_layer_map is not None else {'IT': 'layer'} + self._process_result = (process_result if process_result is not None + else FakeResult(nbytes=activation_nbytes, n_features=n_features)) @property def identifier(self) -> str: @@ -90,24 +97,16 @@ def process(self, stimuli): raise RuntimeError("recording not started") if 'A1' in self._recorded: raise RuntimeError("audio layer not available on this wrapper") - return type('R', (), {'shape': (1, self._per * len(self._recorded))})() + return FakeResult(n_features=self._per * len(self._recorded)) -class FakeStimulusSet: - """Minimal stimulus set with iloc support.""" - - def __init__(self, n=10): - self._n = n - - def __len__(self): - return self._n - - @property - def iloc(self): - return self - - def __getitem__(self, key): - return FakeStimulusSet(n=1) +def FakeStimulusSet(n=10): + """A real n-row stimulus table, since the probe runs through a session.""" + import pandas as pd + stimuli = StimulusSet(pd.DataFrame({'stimulus_id': [f's{i}' for i in range(n)], + 'image_path': [f'/tmp/s{i}.png' for i in range(n)]})) + stimuli.identifier = f'fake-{n}' + return stimuli class FakeAssembly: @@ -544,19 +543,22 @@ def test_plan_probe_uses_declared_target_and_stimuli(self): from brainscore_core.execution_plan import ExecutionPlan recorded = {} + one_frame = FakeStimulusSet(n=1) + one_frame.identifier = 'one-frame' + class _FrameOnlyModel(FakeModel): def start_recording(self, target, time_bins=None, recording_type=None): recorded['target'] = target def process(self, stimuli): - if stimuli != 'ONE_FRAME': + if getattr(stimuli, 'identifier', None) != 'one-frame': raise RuntimeError("image-only model can't process a video row") - return type('R', (), {'shape': (1, 4321)})() + return FakeResult(n_features=4321) model = _FrameOnlyModel(region_layer_map={'V1': 'a', 'IT': 'b'}) bench = FakeBenchmark(identifier='test-ridge', n_stimuli=10) # raw = videos bench.execution_plan = ExecutionPlan( n_extraction_presentations=100000, - recording_target='IT', probe_stimuli='ONE_FRAME') + recording_target='IT', probe_stimuli=one_frame) with patch('brainscore_core.memory.get_host_available_memory', return_value=1_000_000_000), \ _mock_memory(200_000_000): with pytest.raises(MemoryError, match=r"n_features: 4321"): @@ -622,7 +624,7 @@ def test_probed_zero_width_is_not_accepted(self, caplog): import logging # A degenerate (1, 0) probe result must be treated as unresolved (skip # loudly), not a zero-width, zero-memory pass. - model = FakeModel(process_result=type('R', (), {'shape': (1, 0)})()) + model = FakeModel(process_result=FakeResult(n_features=0)) bench = FakeBenchmark(identifier='test-ridge', n_stimuli=10) with patch('brainscore_core.memory.get_host_available_memory', return_value=16_000_000_000), \ _mock_memory(200_000_000): @@ -702,8 +704,12 @@ def test_non_plan_execution_plan_rejected(self): def test_shapeless_probe_result_warns_and_skips(self, caplog): import logging - # process() returns a 1-D result -> feature width unreadable -> skip loudly. - model = FakeModel(process_result=type('R', (), {'shape': (5,)})()) + # process() returns a response with no presentation axis -> the session + # cannot carry it as a neural response -> feature width unreadable -> skip loudly. + import numpy as np + model = FakeModel(process_result=NeuroidAssembly( + np.zeros((5, 3)), dims=['time', 'neuroid'], + coords={'time': ('time', list(range(5))), 'neuroid_id': ('neuroid', [0, 1, 2])})) bench = FakeBenchmark(identifier='test-ridge', n_stimuli=10) with patch('brainscore_core.memory.get_host_available_memory', return_value=16_000_000_000), \ _mock_memory(200_000_000): @@ -829,3 +835,75 @@ def test_unsliceable_stimulus_set_is_not_renamed(self): from brainscore_core.memory import _isolate_probe_identifier parent = self._stimulus_set() assert _isolate_probe_identifier(parent, parent).identifier == 'hvm-public' + + +class TestProbeThroughInteract: + """The feature-width probe runs one session through interact for every subject.""" + + def _stimuli(self): + import pandas as pd + from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet + stimuli = StimulusSet(pd.DataFrame({'stimulus_id': ['a', 'b'], + 'image_path': ['/tmp/a.png', '/tmp/b.png']})) + stimuli.identifier = 'probe-images' + return stimuli + + def test_interact_only_subject_is_probed_from_its_declarations(self): + from types import SimpleNamespace + import numpy as np + from brainscore_core.memory import _probe_feature_dim, _probe_regions + from brainscore_core.model_interface import Subject + from brainscore_core.streaming import StreamEvent + from brainscore_core.supported_data_standards.brainio.assemblies import NeuroidAssembly + + class Native(Subject): + identifier = 'native' + in_channels = {'vision'} + out_channels = {'neural:IT'} + required_channels = set() + + def interact(self, session): + n = 0 + while session.next_input() is not None: + n += 1 + session.respond(StreamEvent('neural:IT', NeuroidAssembly( + np.zeros((n, 7)), dims=['presentation', 'neuroid'], + coords={'stimulus_id': ('presentation', list(range(n))), + 'neuroid_id': ('neuroid', list(range(7)))}), None)) + + benchmark = SimpleNamespace(identifier='bench') + assert _probe_regions(Native(), benchmark) == ['IT'] + width, region = _probe_feature_dim(Native(), benchmark, self._stimuli()) + assert (width, region) == (7, 'IT') + + def test_typed_model_is_probed_the_same_way(self): + from types import SimpleNamespace + import numpy as np + from brainscore_core.memory import _probe_feature_dim, _probe_regions + from brainscore_core.model_interface import BrainScoreModel + from brainscore_core.supported_data_standards.brainio.assemblies import NeuroidAssembly + + def extractor(stimuli, layers=None, **kwargs): + n = len(stimuli) + return NeuroidAssembly(np.zeros((n, 5)), dims=['presentation', 'neuroid'], + coords={'stimulus_id': ('presentation', list(stimuli['stimulus_id'])), + 'neuroid_id': ('neuroid', list(range(5))), + 'layer': ('neuroid', [layers[0]] * 5)}) + + model = BrainScoreModel('typed', region_layer_map={'V4': 'early', 'IT': 'late'}, + extractor={'vision': extractor}) + benchmark = SimpleNamespace(identifier='bench') + assert _probe_regions(model, benchmark) == ['V4', 'IT'] # region map order kept + width, region = _probe_feature_dim(model, benchmark, self._stimuli()) + assert (width, region) == (5, 'V4') + width, region = _probe_feature_dim(model, SimpleNamespace(identifier='it', region='IT'), + self._stimuli()) + assert (width, region) == (5, 'IT') + + def test_no_neural_output_means_no_probe(self): + from types import SimpleNamespace + from brainscore_core.memory import _probe_feature_dim, _probe_regions + from brainscore_core.model_interface import BrainScoreModel + behavioral = BrainScoreModel('policy', policy=lambda step: None) + assert _probe_regions(behavioral, SimpleNamespace(identifier='b')) == [None] + assert _probe_feature_dim(behavioral, SimpleNamespace(identifier='b'), self._stimuli()) == (None, None) diff --git a/tests/test_model_interface_public_api.py b/tests/test_model_interface_public_api.py index 8b9439d0..eb9fac9a 100644 --- a/tests/test_model_interface_public_api.py +++ b/tests/test_model_interface_public_api.py @@ -18,6 +18,7 @@ 'Dict', 'EnvironmentResponse', 'EnvironmentStep', + 'Extractor', 'FunctionalSelection', 'IndexSelection', 'InputEvent', diff --git a/tests/test_out_channels.py b/tests/test_out_channels.py index 90415fbd..807d096e 100644 --- a/tests/test_out_channels.py +++ b/tests/test_out_channels.py @@ -10,7 +10,9 @@ def test_motor_advertised_when_action_fn_present(): def test_motor_absent_without_action_fn(): - m = BrainScoreModel('feat', None, {'IT': 'layer4'}, {}) + # A neural output is declared only with a wired extractor (rule 2.4). + m = BrainScoreModel('feat', None, {'IT': 'layer4'}, + extractor={'vision': lambda stimuli, layers=None, **k: None}) assert 'motor' not in m.out_channels assert 'neural:IT' in m.out_channels diff --git a/tests/test_partial_inputs.py b/tests/test_partial_inputs.py new file mode 100644 index 00000000..8f179482 --- /dev/null +++ b/tests/test_partial_inputs.py @@ -0,0 +1,54 @@ +"""Contract v3 rule 5.5: streams a subject cannot take in never enter it.""" +import numpy as np +import pandas as pd + +from brainscore_core.compatibility import check_compatibility +from brainscore_core.model_interface import BrainScoreModel +from brainscore_core.streaming_helpers import neural_response +from brainscore_core.supported_data_standards.brainio.assemblies import NeuroidAssembly +from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet + + +class _VisionExtractor: + def __init__(self): + self.seen_columns = [] + + def __call__(self, stimuli, layers=None, **kwargs): + self.seen_columns.append(sorted(stimuli.columns)) + n = len(stimuli) + return NeuroidAssembly( + np.ones((n, 2)), dims=['presentation', 'neuroid'], + coords={'stimulus_id': ('presentation', list(stimuli['stimulus_id'])), + 'neuroid_id': ('neuroid', [0, 1]), + 'layer': ('neuroid', [layers[0]] * 2)}) + + +class _Benchmark: + identifier = 'movie-fmri' + required_modalities = {'vision', 'text', 'audio'} + region = 'IT' + + +def _movie(): + stimuli = StimulusSet(pd.DataFrame({ + 'stimulus_id': ['clip0', 'clip1'], + 'image_path': ['/tmp/0.png', '/tmp/1.png'], + 'sentence': ['subtitle zero', 'subtitle one'], + 'audio_path': ['/tmp/0.wav', '/tmp/1.wav'], + })) + stimuli.identifier = 'movie-clips' + return stimuli + + +def test_vision_only_subject_runs_audiovisual_benchmark_on_vision_alone(): + extractor = _VisionExtractor() + model = BrainScoreModel('vision-only', region_layer_map={'IT': 'layer'}, + extractor={'vision': extractor}) + + match = check_compatibility(model, _Benchmark()) + assert match.kind == 'partial' + assert match.dropped_inputs == {'text', 'audio'} + + response = neural_response(model, _movie(), record='IT') + assert response.sizes['presentation'] == 2 + assert len(extractor.seen_columns) == 1 # one extraction, vision only diff --git a/tests/test_score_status.py b/tests/test_score_status.py index dd89a81e..60305a61 100644 --- a/tests/test_score_status.py +++ b/tests/test_score_status.py @@ -7,10 +7,11 @@ class _Model: + # Compatibility reads declarations only (Contract v3 rule 2.1). identifier = "text-model" - available_modalities = {"text"} - required_modalities = set() - region_layer_map = {} + in_channels = {"text"} + out_channels = set() + required_channels = set() class _Benchmark: diff --git a/tests/test_session_view.py b/tests/test_session_view.py new file mode 100644 index 00000000..76618ce5 --- /dev/null +++ b/tests/test_session_view.py @@ -0,0 +1,106 @@ +"""Contract v3 rule 5.5: a stream the subject cannot take in never enters it.""" +import numpy as np +import pandas as pd +import pytest + +from brainscore_core.events import EnvironmentResponse, EnvironmentStep +from brainscore_core.model_interface import BrainScoreModel +from brainscore_core.session_view import DeclaredInputSession +from brainscore_core.streaming import BufferedSession, StreamEvent +from brainscore_core.streaming_helpers import neural_response +from brainscore_core.supported_data_standards.brainio.assemblies import NeuroidAssembly +from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet + + +class _VisionExtractor: + def __init__(self): + self.seen = [] + + def __call__(self, stimuli, layers=None, **kwargs): + self.seen.append(stimuli) + n = len(stimuli) + return NeuroidAssembly( + np.ones((n, 2)), dims=['presentation', 'neuroid'], + coords={'stimulus_id': ('presentation', list(stimuli['stimulus_id'])), + 'neuroid_id': ('neuroid', [0, 1]), + 'layer': ('neuroid', [layers[0]] * 2)}) + + +def _model(): + extractor = _VisionExtractor() + model = BrainScoreModel('vision-only', region_layer_map={'IT': 'layer'}, + extractor={'vision': extractor}) + return model, extractor + + +def _movie(): + stimuli = StimulusSet(pd.DataFrame({ + 'stimulus_id': ['clip0', 'clip1'], + 'image_path': ['/tmp/0.png', '/tmp/1.png'], + 'sentence': ['subtitle zero', 'subtitle one'], + 'audio_path': ['/tmp/0.wav', '/tmp/1.wav'], + 'object_name': ['cat', 'dog'], + })) + stimuli.identifier = 'movie-clips' + return stimuli + + +def test_undeclared_modalities_never_reach_the_extractor(): + model, extractor = _model() + response = neural_response(model, _movie(), record='IT') + assert response.sizes['presentation'] == 2 + seen, = extractor.seen + assert sorted(seen.columns) == ['image_path', 'object_name', 'stimulus_id'] + assert seen.identifier == 'movie-clips' # cache identity survives the projection + + +def test_stimulus_set_keeps_its_identity_when_nothing_is_dropped(): + model, extractor = _model() + stimuli = StimulusSet(pd.DataFrame({'stimulus_id': ['a'], 'image_path': ['/tmp/a.png']})) + stimuli.identifier = 'one-image' + neural_response(model, stimuli, record='IT') + assert extractor.seen[0] is stimuli + + +def test_undeclared_events_are_skipped_and_declared_ones_delivered(): + model, extractor = _model() + session = BufferedSession(inputs=[ + StreamEvent('text', 'not for this subject', 0.0, + meta={'stimulus_id': 't', 'column': 'sentence'}), + StreamEvent('vision', '/tmp/a.png', 0.0, + meta={'stimulus_id': 'a', 'column': 'image_path'}), + ], requested_output_channels=['neural:IT']) + model.interact(session) + seen, = extractor.seen + assert list(seen['stimulus_id']) == ['a'] + assert session.collect('neural:IT').sizes['presentation'] == 1 + + +def test_environment_steps_pass_through_to_the_policy(): + steps = [] + model = BrainScoreModel('policy', policy=lambda step: steps.append(step) or EnvironmentResponse(action=1)) + step = EnvironmentStep(observation={'value': 1}) + session = BufferedSession([step], requested_output_channels=['motor']) + model.interact(session) + assert steps == [step] + assert session.emitted[0].payload.action == 1 + + +def test_view_forwards_completion_responses_and_collection(): + inner = BufferedSession([StreamEvent('vision', 'x', 0.0), StreamEvent('audio', 'y', 0.0)], + requested_output_channels=['neural:IT'], time_bins=[(0, 10)]) + view = DeclaredInputSession(inner, {'vision'}) + assert view.complete is inner.complete + assert view.time_bins == [(0, 10)] + assert view.requested_output_channels == ('neural:IT',) + assert view.next_input().channel == 'vision' + assert view.next_input() is None + view.respond(StreamEvent('neural:IT', 'out', 0.0)) + assert inner.emitted[0].payload == 'out' + assert view.collect('neural:IT') == inner.collect('neural:IT') + + +def test_view_carries_measurement_settings(): + inner = BufferedSession([StreamEvent('vision', 'x', 0.0)], requested_output_channels=['neural:IT']) + inner.settings = {'number_of_trials': 3} + assert DeclaredInputSession(inner, {'vision'}).settings == {'number_of_trials': 3} diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 9a09f891..febd686b 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -94,3 +94,39 @@ def test_in_memory_session_drives_inputs_and_records_emits(): session.emit(output) assert session.emitted == [output] + + +def test_buffered_session_rejects_wrong_direction_at_construction(): + """Contract v3 rule 3.4: known inputs are validated before delivery.""" + import pytest + from brainscore_core.streaming import BufferedSession, StreamEvent + with pytest.raises(ValueError, match="registry direction"): + BufferedSession([StreamEvent("behavior", "x", 0.0)]) + + +def test_buffered_session_warns_on_unregistered_family(): + import pytest + from brainscore_core.streaming import BufferedSession, StreamEvent + with pytest.warns(UserWarning, match="unregistered channel family"): + BufferedSession([StreamEvent("not_a_family", 1, 0.0)]) + + +def test_buffered_session_warns_on_payload_type_mismatch_and_still_delivers(): + """Payload types are checked warn-first: a mismatch is reported, not rejected.""" + import pytest + from brainscore_core.streaming import BufferedSession, StreamEvent + with pytest.warns(UserWarning, match="payload is int, not str"): + session = BufferedSession([StreamEvent("text", 42, 0.0)]) + assert session.next_input().payload == 42 + + +def test_measurement_settings_are_a_saved_session_context_field(): + """A recorder that saves Session.context_fields keeps per-measurement settings.""" + from brainscore_core.streaming import BufferedSession, Session, StreamEvent + assert 'settings' in Session.context_fields + assert Session.settings is None + session = BufferedSession([StreamEvent('text', 'hello', None)], requested_output_channels=['behavior']) + session.settings = {'number_of_trials': 7, 'require_variance': True} + saved = {name: getattr(session, name) for name in Session.context_fields + if getattr(session, name, None) is not None} + assert saved['settings'] == {'number_of_trials': 7, 'require_variance': True} diff --git a/tests/test_subject_channels.py b/tests/test_subject_channels.py index a301945d..29302175 100644 --- a/tests/test_subject_channels.py +++ b/tests/test_subject_channels.py @@ -31,7 +31,9 @@ def test_subject_derives_channel_identity_from_v1_contract(): subject = _ChannelSubject() assert subject.in_channels == {"vision", "text"} - assert subject.out_channels == {"neural:V1", "neural:IT", "behavior"} + # Contract v3 rule 2.2: outputs come from declarations, never from private + # attribute names. The fixture's ``_behavioral_readout_layer`` declares nothing. + assert subject.out_channels == {"neural:V1", "neural:IT"} assert subject.required_channels == {"text"} @@ -56,11 +58,12 @@ def required_channels(self): assert subject.required_channels == {"instruction"} -def test_legacy_interact_default_is_non_abstract_and_clear(): - subject = _ChannelSubject() +def test_typed_interact_default_runs_sessions_and_rejects_unknown_outputs(): + from brainscore_core.streaming import BufferedSession - with pytest.raises(NotImplementedError, match="no v2 interact\\(\\) path yet"): - subject.interact(session=None) + subject = _ChannelSubject() + with pytest.raises(NotImplementedError, match="runs sessions requesting"): + subject.interact(BufferedSession(requested_output_channels=["pupil"])) def test_legacy_subclass_still_instantiates_without_interact(): @@ -103,3 +106,47 @@ def test_brainscore_model_gets_channel_identity_from_preprocessors(): def test_unifiedmodel_remains_a_subject(): assert issubclass(UnifiedModel, Subject) assert UnifiedModel is not Subject + + +def test_typed_model_without_multi_modality_runs_multimodal_session(): + """The driver asks for multi-modality only when the subject says it combines + inputs; no private method is probed (Contract v3 rule 2.2).""" + import pandas as pd + from brainscore_core.streaming_helpers import neural_response + from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet + + class TwoInputModel(_ChannelSubject): + def __init__(self): + super().__init__() + self.calls = [] + + def start_recording(self, recording_target, time_bins=None, recording_type=None): + pass + + def process(self, input_event): # no multi_modality keyword + self.calls.append(len(input_event)) + import numpy as np + from brainscore_core.supported_data_standards.brainio.assemblies import NeuroidAssembly + return NeuroidAssembly(np.zeros((len(input_event), 2)), dims=['presentation', 'neuroid'], + coords={'stimulus_id': ('presentation', list(input_event['stimulus_id'])), + 'neuroid_id': ('neuroid', [0, 1])}) + + stimuli = StimulusSet(pd.DataFrame({'stimulus_id': ['a'], 'image_path': ['/tmp/a.png'], + 'sentence': ['a sentence']})) + stimuli.identifier = 'two-inputs' + model = TwoInputModel() + neural_response(model, stimuli, record='IT') + assert model.calls == [1] + + +def test_region_map_without_extractor_declares_no_neural_output(): + """Rule 2.4: a declared channel must work; a map alone cannot record.""" + from brainscore_core.model_interface import BrainScoreModel + model = BrainScoreModel('unwired', region_layer_map={'IT': 'layer'}) + assert 'neural:IT' not in model.out_channels + with pytest.raises(NotImplementedError, match='No session handler'): + model.check_session_support({'neural:IT'}) + wired = BrainScoreModel('wired', region_layer_map={'IT': 'layer'}, + extractor={'vision': lambda stimuli, layers=None, **k: None}) + assert 'neural:IT' in wired.out_channels + diff --git a/tests/test_typed_convenience.py b/tests/test_typed_convenience.py new file mode 100644 index 00000000..144ce5c9 --- /dev/null +++ b/tests/test_typed_convenience.py @@ -0,0 +1,30 @@ +"""process() is a typed convenience of UnifiedModel, not a requirement on it.""" +import pandas as pd +import pytest + +from brainscore_core.model_interface import UnifiedModel +from brainscore_core.streaming import BufferedSession +from brainscore_core.supported_data_standards.brainio.stimuli import StimulusSet + + +class NoProcess(UnifiedModel): + @property + def identifier(self): + return 'no-process' + + @property + def region_layer_map(self): + return {'IT': 'layer'} + + @property + def supported_modalities(self): + return {'vision'} + + +def test_typed_subclass_instantiates_without_process_and_fails_clearly(): + subject = NoProcess() # no abstract method stops instantiation + stimuli = StimulusSet(pd.DataFrame({'stimulus_id': ['a'], 'image_path': ['/tmp/a.png']})) + stimuli.identifier = 'one' + session = BufferedSession(stimulus_set=stimuli, requested_output_channels=['neural:IT']) + with pytest.raises(NotImplementedError, match='defines no process'): + subject.interact(session)