# ❤️ PyHeart - Healthcare Interoperability & Workflow Engine
[](https://www.python.org/downloads/)
[](LICENSE)
[](https://pypi.org/project/pyheart/)
PyHeart is the integration layer of the BrainSAIT Healthcare Unification Platform, providing universal healthcare system connectivity, workflow orchestration, and secure data exchange.
## 🚀 Features
- **Universal API Gateway**: Single interface for all healthcare system integrations
- **Event-Driven Architecture**: Real-time data streaming and processing
- **Microservices Framework**: Modular, scalable healthcare services
- **Security & Compliance Engine**: HIPAA, GDPR, and regional compliance automation
- **Workflow Orchestration**: Complex healthcare process automation
## 📦 Installation
```bash
pip install pyheart
```
For development:
```bash
pip install pyheart[dev]
```
For legacy system support:
```bash
pip install pyheart[legacy]
```
## 🔧 Quick Start
### FHIR Client Usage
```python
from pyheart import FHIRClient
# Connect to FHIR server
client = FHIRClient("https://fhir.example.com")
# Get patient
patient = client.get_patient("12345")
print(f"Patient: {patient}")
# Search patients
bundle = client.search("Patient", {"family": "Smith", "birthdate": "ge1970"})
for entry in bundle.get("entry", []):
print(f"Found: {entry['resource']['id']}")
# Create patient
new_patient = {
"resourceType": "Patient",
"name": [{"given": ["John"], "family": "Doe"}],
"gender": "male",
"birthDate": "1990-01-01"
}
created = client.create(new_patient)
```
### Workflow Engine
```python
from pyheart import WorkflowEngine, ProcessDefinition, Task
# Define a clinical process
process = ProcessDefinition(
id="medication-reconciliation",
name="Medication Reconciliation Process",
tasks=[
Task(
id="fetch-current-meds",
name="Fetch Current Medications",
type="api_call",
config={
"url": "${fhir_server}/MedicationRequest?patient=${patient_id}",
"method": "GET"
}
),
Task(
id="analyze-interactions",
name="Analyze Drug Interactions",
type="transformation",
dependencies=["fetch-current-meds"],
config={
"transform": {"type": "drug_interaction_check"}
}
)
]
)
# Execute workflow
engine = WorkflowEngine()
engine.register_process(process)
instance_id = await engine.start_process(
"medication-reconciliation",
{"patient_id": "12345", "physician_email": "dr.smith@example.com"}
)
```
### Healthcare System Integration
```python
from pyheart import IntegrationHub, FHIRAdapter, HL7Adapter
# Setup integration hub
hub = IntegrationHub()
# Register adapters for different systems
hub.register_adapter("hospital_a", FHIRAdapter("hospital_a"))
hub.register_adapter("lab_system", HL7Adapter("lab_system"))
# Connect to systems
await hub.connect_system("hospital_a", {"base_url": "https://hospital-a.com/fhir"})
await hub.connect_system("lab_system", {"host": "lab.hospital.com", "port": 2575})
# Fetch data from all systems
patient_data = await hub.fetch_from_all_systems("Patient", {"id": "12345"})
```
### CLI Usage
```bash
# FHIR operations
pyheart fhir -s https://fhir.example.com -r Patient -i 12345
pyheart fhir -s https://fhir.example.com -r Observation -q '{"patient": "12345"}'
# Run workflows
pyheart workflow -f medication-check.json -v '{"patient_id": "12345"}'
# Start server
pyheart serve --port 8000
# Sync data between systems
pyheart sync -s https://old.example.com -t https://new.example.com -r Patient
# System diagnostics
pyheart doctor
```
## 🏗️ Architecture
PyHeart provides a layered architecture for healthcare integration:
```
pyheart/
├── core/
│ ├── client/ # FHIR and legacy clients
│ ├── server/ # API gateway and FHIR server
│ ├── workflow/ # Process orchestration
│ ├── integration/ # System adapters
│ └── security/ # Auth and encryption
├── adapters/ # Legacy system adapters
├── messaging/ # Event streaming
└── api/ # REST/GraphQL APIs
```
## 🤝 Integration with PyBrain
PyHeart and PyBrain work together seamlessly:
```python
from pyheart import FHIRClient, WorkflowEngine
from pybrain import AIEngine, DataHarmonizer
# Use PyHeart for data access
client = FHIRClient("https://fhir.example.com")
observations = client.search("Observation", {"patient": "12345"})
# Use PyBrain for intelligence
ai = AIEngine()
harmonizer = DataHarmonizer()
# Process observations with AI
for entry in observations.get("entry", []):
obs = entry["resource"]
# Harmonize if needed
if obs.get("meta", {}).get("source") != "unified":
obs = harmonizer.harmonize_to_fhir(obs, "custom", "Observation")
# AI analysis
risk = ai.predict_risk_score({"patient_id": "12345"})
# Trigger workflow if high risk
if risk > 0.8:
engine = WorkflowEngine()
await engine.start_process("high-risk-intervention", {
"patient_id": "12345",
"risk_score": risk
})
```
## 🔒 Security & Compliance
PyHeart includes comprehensive security features:
```python
from pyheart import SecurityManager, AuthProvider
# Configure security
security = SecurityManager()
security.enable_encryption("AES-256-GCM")
security.enable_audit_logging()
security.configure_compliance(["HIPAA", "GDPR"])
# OAuth2/SMART on FHIR authentication
auth = AuthProvider("oauth2")
token = await auth.get_token(
client_id="app-123",
scope="patient/*.read launch"
)
# Use authenticated client
client = FHIRClient(
base_url="https://fhir.example.com",
auth_token=token
)
```
## 🌟 Advanced Features
### Multi-System Client
```python
from pyheart import HealthcareClient
# Query multiple systems simultaneously
client = HealthcareClient()
client.add_fhir_system("hospital_a", FHIRClient("https://a.example.com"))
client.add_fhir_system("hospital_b", FHIRClient("https://b.example.com"))
# Unified patient record
unified_patient = await client.get_unified_patient("12345")
```
### Workflow Orchestration
- Visual workflow designer compatible
- Event-driven triggers
- Human task management
- Error handling and retries
- Parallel and sequential execution
### Legacy System Support
- HL7v2 messaging
- DICOM integration
- X12 transactions
- Custom adapter framework
## 📚 Documentation
Full documentation available at: https://pyheart.readthedocs.io
### Quick Links
- API Reference
- Workflow Guide
- Integration Patterns
- Security Best Practices
## 🧪 Testing
```bash
# Run tests
pytest
# With coverage
pytest --cov=pyheart
# Integration tests
pytest tests/integration --integration
```
## 🚀 Deployment
### Docker
```bash
docker run -p 8000:8000 brainsait/pyheart:latest
```
### Kubernetes
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: pyheart
spec:
replicas: 3
selector:
matchLabels:
app: pyheart
template:
spec:
containers:
- name: pyheart
image: brainsait/pyheart:latest
ports:
- containerPort: 8000
```
## 🤝 Contributing
We welcome contributions! Please see our Contributing Guide for details.
## 📄 License
PyHeart is licensed under the Apache License 2.0. See LICENSE for details.
## 🌟 Acknowledgments
Built with ❤️ by the BrainSAIT Healthcare Innovation Lab
Special thanks to:
- The FHIR community for excellent standards
- FastAPI for the amazing web framework
- All our contributors and users
---
**Together with PyBrain, PyHeart is building the future of connected healthcare.**
Raw data
{
"_id": null,
"home_page": null,
"name": "brainsait-pyheart",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "BrainSAIT Healthcare Innovation Lab <healthcare@brainsait.com>",
"keywords": "healthcare, fhir, hl7, interoperability, integration, workflow, orchestration, api-gateway, health-informatics, medical-integration, healthcare-api",
"author": null,
"author_email": "\"Dr. Fadil\" <fadil@brainsait.com>, BrainSAIT Team <team@brainsait.com>",
"download_url": "https://files.pythonhosted.org/packages/02/19/53eb2a00e2f39eb60a155e5f0e4c89ed6ce86331467b62e7076891739a06/brainsait_pyheart-0.1.0.tar.gz",
"platform": null,
"description": "# \u2764\ufe0f PyHeart - Healthcare Interoperability & Workflow Engine\n\n[](https://www.python.org/downloads/)\n[](LICENSE)\n[](https://pypi.org/project/pyheart/)\n\nPyHeart is the integration layer of the BrainSAIT Healthcare Unification Platform, providing universal healthcare system connectivity, workflow orchestration, and secure data exchange.\n\n## \ud83d\ude80 Features\n\n- **Universal API Gateway**: Single interface for all healthcare system integrations\n- **Event-Driven Architecture**: Real-time data streaming and processing\n- **Microservices Framework**: Modular, scalable healthcare services\n- **Security & Compliance Engine**: HIPAA, GDPR, and regional compliance automation\n- **Workflow Orchestration**: Complex healthcare process automation\n\n## \ud83d\udce6 Installation\n\n```bash\npip install pyheart\n```\n\nFor development:\n```bash\npip install pyheart[dev]\n```\n\nFor legacy system support:\n```bash\npip install pyheart[legacy]\n```\n\n## \ud83d\udd27 Quick Start\n\n### FHIR Client Usage\n\n```python\nfrom pyheart import FHIRClient\n\n# Connect to FHIR server\nclient = FHIRClient(\"https://fhir.example.com\")\n\n# Get patient\npatient = client.get_patient(\"12345\")\nprint(f\"Patient: {patient}\")\n\n# Search patients\nbundle = client.search(\"Patient\", {\"family\": \"Smith\", \"birthdate\": \"ge1970\"})\nfor entry in bundle.get(\"entry\", []):\n print(f\"Found: {entry['resource']['id']}\")\n\n# Create patient\nnew_patient = {\n \"resourceType\": \"Patient\",\n \"name\": [{\"given\": [\"John\"], \"family\": \"Doe\"}],\n \"gender\": \"male\",\n \"birthDate\": \"1990-01-01\"\n}\ncreated = client.create(new_patient)\n```\n\n### Workflow Engine\n\n```python\nfrom pyheart import WorkflowEngine, ProcessDefinition, Task\n\n# Define a clinical process\nprocess = ProcessDefinition(\n id=\"medication-reconciliation\",\n name=\"Medication Reconciliation Process\",\n tasks=[\n Task(\n id=\"fetch-current-meds\",\n name=\"Fetch Current Medications\",\n type=\"api_call\",\n config={\n \"url\": \"${fhir_server}/MedicationRequest?patient=${patient_id}\",\n \"method\": \"GET\"\n }\n ),\n Task(\n id=\"analyze-interactions\",\n name=\"Analyze Drug Interactions\",\n type=\"transformation\",\n dependencies=[\"fetch-current-meds\"],\n config={\n \"transform\": {\"type\": \"drug_interaction_check\"}\n }\n )\n ]\n)\n\n# Execute workflow\nengine = WorkflowEngine()\nengine.register_process(process)\n\ninstance_id = await engine.start_process(\n \"medication-reconciliation\",\n {\"patient_id\": \"12345\", \"physician_email\": \"dr.smith@example.com\"}\n)\n```\n\n### Healthcare System Integration\n\n```python\nfrom pyheart import IntegrationHub, FHIRAdapter, HL7Adapter\n\n# Setup integration hub\nhub = IntegrationHub()\n\n# Register adapters for different systems\nhub.register_adapter(\"hospital_a\", FHIRAdapter(\"hospital_a\"))\nhub.register_adapter(\"lab_system\", HL7Adapter(\"lab_system\"))\n\n# Connect to systems\nawait hub.connect_system(\"hospital_a\", {\"base_url\": \"https://hospital-a.com/fhir\"})\nawait hub.connect_system(\"lab_system\", {\"host\": \"lab.hospital.com\", \"port\": 2575})\n\n# Fetch data from all systems\npatient_data = await hub.fetch_from_all_systems(\"Patient\", {\"id\": \"12345\"})\n```\n\n### CLI Usage\n\n```bash\n# FHIR operations\npyheart fhir -s https://fhir.example.com -r Patient -i 12345\npyheart fhir -s https://fhir.example.com -r Observation -q '{\"patient\": \"12345\"}'\n\n# Run workflows\npyheart workflow -f medication-check.json -v '{\"patient_id\": \"12345\"}'\n\n# Start server\npyheart serve --port 8000\n\n# Sync data between systems\npyheart sync -s https://old.example.com -t https://new.example.com -r Patient\n\n# System diagnostics\npyheart doctor\n```\n\n## \ud83c\udfd7\ufe0f Architecture\n\nPyHeart provides a layered architecture for healthcare integration:\n\n```\npyheart/\n\u251c\u2500\u2500 core/\n\u2502 \u251c\u2500\u2500 client/ # FHIR and legacy clients\n\u2502 \u251c\u2500\u2500 server/ # API gateway and FHIR server\n\u2502 \u251c\u2500\u2500 workflow/ # Process orchestration\n\u2502 \u251c\u2500\u2500 integration/ # System adapters\n\u2502 \u2514\u2500\u2500 security/ # Auth and encryption\n\u251c\u2500\u2500 adapters/ # Legacy system adapters\n\u251c\u2500\u2500 messaging/ # Event streaming\n\u2514\u2500\u2500 api/ # REST/GraphQL APIs\n```\n\n## \ud83e\udd1d Integration with PyBrain\n\nPyHeart and PyBrain work together seamlessly:\n\n```python\nfrom pyheart import FHIRClient, WorkflowEngine\nfrom pybrain import AIEngine, DataHarmonizer\n\n# Use PyHeart for data access\nclient = FHIRClient(\"https://fhir.example.com\")\nobservations = client.search(\"Observation\", {\"patient\": \"12345\"})\n\n# Use PyBrain for intelligence\nai = AIEngine()\nharmonizer = DataHarmonizer()\n\n# Process observations with AI\nfor entry in observations.get(\"entry\", []):\n obs = entry[\"resource\"]\n \n # Harmonize if needed\n if obs.get(\"meta\", {}).get(\"source\") != \"unified\":\n obs = harmonizer.harmonize_to_fhir(obs, \"custom\", \"Observation\")\n \n # AI analysis\n risk = ai.predict_risk_score({\"patient_id\": \"12345\"})\n \n # Trigger workflow if high risk\n if risk > 0.8:\n engine = WorkflowEngine()\n await engine.start_process(\"high-risk-intervention\", {\n \"patient_id\": \"12345\",\n \"risk_score\": risk\n })\n```\n\n## \ud83d\udd12 Security & Compliance\n\nPyHeart includes comprehensive security features:\n\n```python\nfrom pyheart import SecurityManager, AuthProvider\n\n# Configure security\nsecurity = SecurityManager()\nsecurity.enable_encryption(\"AES-256-GCM\")\nsecurity.enable_audit_logging()\nsecurity.configure_compliance([\"HIPAA\", \"GDPR\"])\n\n# OAuth2/SMART on FHIR authentication\nauth = AuthProvider(\"oauth2\")\ntoken = await auth.get_token(\n client_id=\"app-123\",\n scope=\"patient/*.read launch\"\n)\n\n# Use authenticated client\nclient = FHIRClient(\n base_url=\"https://fhir.example.com\",\n auth_token=token\n)\n```\n\n## \ud83c\udf1f Advanced Features\n\n### Multi-System Client\n```python\nfrom pyheart import HealthcareClient\n\n# Query multiple systems simultaneously\nclient = HealthcareClient()\nclient.add_fhir_system(\"hospital_a\", FHIRClient(\"https://a.example.com\"))\nclient.add_fhir_system(\"hospital_b\", FHIRClient(\"https://b.example.com\"))\n\n# Unified patient record\nunified_patient = await client.get_unified_patient(\"12345\")\n```\n\n### Workflow Orchestration\n- Visual workflow designer compatible\n- Event-driven triggers\n- Human task management\n- Error handling and retries\n- Parallel and sequential execution\n\n### Legacy System Support\n- HL7v2 messaging\n- DICOM integration\n- X12 transactions\n- Custom adapter framework\n\n## \ud83d\udcda Documentation\n\nFull documentation available at: https://pyheart.readthedocs.io\n\n### Quick Links\n- API Reference\n- Workflow Guide\n- Integration Patterns\n- Security Best Practices\n\n## \ud83e\uddea Testing\n\n```bash\n# Run tests\npytest\n\n# With coverage\npytest --cov=pyheart\n\n# Integration tests\npytest tests/integration --integration\n```\n\n## \ud83d\ude80 Deployment\n\n### Docker\n```bash\ndocker run -p 8000:8000 brainsait/pyheart:latest\n```\n\n### Kubernetes\n```yaml\napiVersion: apps/v1\nkind: Deployment\nmetadata:\n name: pyheart\nspec:\n replicas: 3\n selector:\n matchLabels:\n app: pyheart\n template:\n spec:\n containers:\n - name: pyheart\n image: brainsait/pyheart:latest\n ports:\n - containerPort: 8000\n```\n\n## \ud83e\udd1d Contributing\n\nWe welcome contributions! Please see our Contributing Guide for details.\n\n## \ud83d\udcc4 License\n\nPyHeart is licensed under the Apache License 2.0. See LICENSE for details.\n\n## \ud83c\udf1f Acknowledgments\n\nBuilt with \u2764\ufe0f by the BrainSAIT Healthcare Innovation Lab\n\nSpecial thanks to:\n- The FHIR community for excellent standards\n- FastAPI for the amazing web framework\n- All our contributors and users\n\n---\n\n**Together with PyBrain, PyHeart is building the future of connected healthcare.**\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Healthcare Interoperability & Workflow Engine - Universal integration platform for healthcare systems",
"version": "0.1.0",
"project_urls": {
"Changelog": "https://github.com/brainsait/pyheart/blob/main/CHANGELOG.md",
"Documentation": "https://pyheart.readthedocs.io",
"Homepage": "https://github.com/brainsait/pyheart",
"Issues": "https://github.com/brainsait/pyheart/issues",
"Repository": "https://github.com/brainsait/pyheart"
},
"split_keywords": [
"healthcare",
" fhir",
" hl7",
" interoperability",
" integration",
" workflow",
" orchestration",
" api-gateway",
" health-informatics",
" medical-integration",
" healthcare-api"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "e1036089c9ea293a249dfde5cf4788df94125d8b395e6606d7083978dd113655",
"md5": "014464d2d863c4e55fadbf55d37cbb23",
"sha256": "c4921410baeb69a94fd71db668115969e57c54b64b9e60652807d2d9f4bf149f"
},
"downloads": -1,
"filename": "brainsait_pyheart-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "014464d2d863c4e55fadbf55d37cbb23",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 21289,
"upload_time": "2025-08-02T17:16:16",
"upload_time_iso_8601": "2025-08-02T17:16:16.616710Z",
"url": "https://files.pythonhosted.org/packages/e1/03/6089c9ea293a249dfde5cf4788df94125d8b395e6606d7083978dd113655/brainsait_pyheart-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "021953eb2a00e2f39eb60a155e5f0e4c89ed6ce86331467b62e7076891739a06",
"md5": "cfc261650fc08a13f996e03093ecff09",
"sha256": "d8c4a43f5e4934a4445c6cf55d1f51d73153197e52ffb4366c282f34c6cab985"
},
"downloads": -1,
"filename": "brainsait_pyheart-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "cfc261650fc08a13f996e03093ecff09",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 21946,
"upload_time": "2025-08-02T17:16:19",
"upload_time_iso_8601": "2025-08-02T17:16:19.446412Z",
"url": "https://files.pythonhosted.org/packages/02/19/53eb2a00e2f39eb60a155e5f0e4c89ed6ce86331467b62e7076891739a06/brainsait_pyheart-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-02 17:16:19",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "brainsait",
"github_project": "pyheart",
"github_not_found": true,
"lcname": "brainsait-pyheart"
}
JSON
