class DataGenerator:
    @ether_start() 
    @ether_pub(topic="data")
    def generate_data(self, data: int = 42) -> dict:
        print(f"DataGenerator[PID {os.getpid()}]: Generating data - {data}")
        return {"data": data}

class DataProcessor:
    def __init__(self, multiplier: int = 2):
        self.multiplier = multiplier
    
    @ether_sub(topic="data")
    @ether_pub(topic="result")
    def process_data(self, data: int = 0) -> dict:
        print(f"DataProcessor[PID {os.getpid()}]: Processing input data - {data}")
        result = data * self.multiplier
        print(f"DataProcessor[PID {os.getpid()}]: Processed output result - {result}")
        return {"result": result}

class DataCollector:
    results = []
    
    @ether_sub(topic="result")
    @ether_pub(topic="summarize")
    def collect_result(self, result: int) -> None:
        self.results.append(result)
        print(f"DataCollector[PID {os.getpid()}]: Collected result - {result}")

    @ether_sub(topic="summarize")
    def summarize(self) -> None:
        mean = sum(self.results) / len(self.results)
        max_val = max(self.results)
        min_val = min(self.results)
        print(f"DataCollector[PID {os.getpid()}]: Summarizing results - mean: {mean}, max: {max_val}, min: {min_val}")

import os
import time
from ether import ether

# Get path to config file
def _get_config_path(config_name: str) -> str:
    return os.path.join(os.path.dirname(__file__), "config", f"{config_name}.yaml")

# Tap in to the ether with config
config_path = _get_config_path("dual_processors")
ether.tap(config=config_path)

# Start the distributed system
ether.start()
time.sleep(1.0)  # Let the system run for a bit

# Shut down the distributed system
ether.shutdown()

# After shutdown, you can still use components normally in a single process
generator = DataGenerator()
processor2x = DataProcessor(multiplier=2)
processor4x = DataProcessor(multiplier=4)
collector = DataCollector()

# Manual usage example
generated_data = generator.generate_data(data=4242)
processed2x_result = processor2x.process_data(**generated_data)
processed4x_result = processor4x.process_data(**generated_data)
collector.collect_result(**processed2x_result)
collector.collect_result(**processed4x_result)
collector.summarize()

from typing import List
from ether import ether
from ether.config import EtherConfig, EtherInstanceConfig
from ether import ether_save, ether_get
from ether.utils import get_ether_logger

class DataService:
    def __init__(self):
        self.data = {}
    
    @ether_save()
    def save_item(self, id: int, name: str, tags: List[str]) -> dict:
        """Save an item to the data store"""
        if id in self.data:
            raise ValueError(f"Item {id} already exists")
            
        self.data[id] = {
            "name": name,
            "tags": tags
        }
        return self.data[id]  # Return saved item

    @ether_get
    def get_item(self, id: int) -> dict:
        """Get a single item by ID"""
        if id not in self.data:
            raise KeyError(f"Item {id} not found")
        return self.data[id]

if __name__ == "__main__":
    logger = get_ether_logger("EtherDataServiceMain")
    
    config = EtherConfig(
        instances={
            "data_service": EtherInstanceConfig(
                class_path="test_ether_save.DataService",
                kwargs={"ether_name": "data_service"}
            )
        }
    )
    
    ether.tap(config=config)
    
    try:
        
        # Test saving new item
        reply = ether.save(
            "DataService", 
            "save_item", 
            params={
                "id": 1,
                "name": "Test Item",
                "tags": ["test", "new"]
            },
        )
        logger.info(f"Save reply: {reply}")
        assert reply["status"] == "success"
        result = reply["result"]
        assert result["name"] == "Test Item"
        
        # Verify item was saved
        item = ether.get("DataService", "get_item", params={"id": 1})
        logger.info(f"Get item: {item}")
        assert item["name"] == "Test Item"
        assert item["tags"] == ["test", "new"]
        
        # Test saving duplicate item

        reply = ether.save(
            "DataService", 
            "save_item", 
            params={
                "id": 1,
                "name": "Duplicate",
                "tags": []
            },
        )
        logger.info(f"Duplicate save reply: {reply}")
        assert reply["status"] == "error"
        assert "already exists" in str(reply["error"])
        
    finally:
        ether.shutdown()

req_id = ether.get_later('my_service',...)

response = ether.collect(req_id)

class HardwareComponentA:

    @ether.sync(topic='prepare_trial')
    def set_state(config):
        ...

class HardwareComponentB:

    @ether.sync(topic='prepare_trial')
    def set_state(config):
        ...

# this call will wait until all subscribers to the sync topic have finished
ether.sync('prepare_trial')

run_trial()

class TemperatureSensor:
    @ether_pub(topic="DataLogger.log_temperature")
    def read_temperature(self) -> dict:
        temp = self._hardware.get_temperature()
        return {"temperature": temp, "timestamp": time.time()}

class DataLogger:
    @ether_sub()
    @ether_pub(topic="ExperimentController.check_temperature")
    def log_temperature(self, temperature: float, timestamp: float) -> dict:
        self._db.save_temperature(temperature, timestamp)
        return {"temperature": temperature, "timestamp": timestamp}

class ExperimentController:
    @ether_sub()
    def check_temperature(self, temperature: float, timestamp: float):
        if temperature > self.max_temp:
            self._safety_shutdown()