brinkhaustools 0.2.2

Common tooling for Brinkhaus industrial applications

brinkhaustools

Common infrastructure toolkit for Brinkhaus industrial applications.

What is brinkhaustools?

brinkhaustools is a Python package providing reusable components for industrial monitoring and control applications. Zero external dependencies for the core — just the Python standard library.

It bundles:

• App -- central wiring object that connects all components
• ShutdownHandler -- graceful shutdown with signal handling and cleanup hooks
• LoggingHelper -- rotating file logs with an in-memory ring buffer
• Settings -- thread-safe JSON settings with hierarchical key access
• SelfDiagnosisEngine -- error/warning tracking with numeric codes and timeouts
• StatusEngine -- periodic status collection from pluggable sources
• HeartbeatEngine -- watchdog for detecting hung threads
• VersionInformation -- CI-generated version metadata
• Fleet management -- HTTPS-based reporting to BrinkhausFleetManager
• RestServer -- Flask + Waitress HTTP server with standard endpoints

Quick Example

from brinkhaustools.common import App

app = App(
    name="my-service",
    settings_path="data/settings/settings_main.json",
    fleet_config_path="fleetManagementData/config.json",
    version_file="helper/version.json",
)
app.start()
app.wait()  # blocks until shutdown

Installation

pip install brinkhaustools

Optional Extras

Extra	What it adds
rest	Flask, Waitress, bcrypt -- needed for RestServer
dev	pytest, mypy -- development tools

pip install brinkhaustools[rest]

Requirements: Python >= 3.10. The core has no external dependencies.

Getting Started

The App Object

App wires together all components with sensible defaults:

Attribute	Component	Default config
app.shutdown	ShutdownHandler	always created
app.logging	LoggingHelper	data/logs/
app.settings	Settings	data/settings/settings_main.json
app.diagnosis	SelfDiagnosisEngine	always created
app.status	StatusEngine	always created
app.heartbeat	HeartbeatEngine	300s default timeout
app.version	VersionInformation	helper/version.json
app.fleet	StatusMonitor	fleetManagementData/config.json

Pass None to skip any component:

app = App(name="my-service", settings_path=None, fleet_config_path=None)

Settings

timeout = app.settings.get(["expert", "timeout"], default=300)
# or with dot-notation:
timeout = app.settings.get("expert.timeout", default=300)

Diagnostics

app.diagnosis.notify(code=1001, message="Database unreachable", critical=True)
app.diagnosis.clear(code=1001)

Heartbeat Watchdog

app.heartbeat.register("main-loop", timeout_sec=60)

while not app.shutdown.is_shutdown():
    app.heartbeat.register("main-loop", timeout_sec=60)  # refresh
    do_work()

Fleet Management

Report status and diagnostics to BrinkhausFleetManager by placing a fleetManagementData/config.json:

{
    "fleetmanagement": {
        "customer_name": "my-customer",
        "machine": "machine-01",
        "broker": "fleet.brinkhaus-gmbh.de",
        "heartbeat_interval_sec": 60,
        "token": "fmt_..."
    }
}

Diagnostics are automatically forwarded to the FleetManager when App is started with a fleet config.

REST Server

Install the rest extra, then:

from brinkhaustools.rest import RestServer

server = RestServer(app=app, port=8080)
server.start()
app.wait()

Graceful Shutdown

app.wait()                          # block until SIGTERM/SIGINT
app.shutdown.trigger()              # or trigger programmatically

Documentation

Full API reference, guides, and configuration details are available on the documentation site:

brinkhaus.gitlab.io/brinkhaustools

License

MIT