Writer API¶
The TeehistorianWriter class provides a Pythonic interface for creating teehistorian files programmatically. It follows Python best practices with context manager support, method chaining, and a clean API.
Quick Start¶
import teehistorian_py as th
# Basic usage with context manager (recommended)
with th.create() as writer:
writer.write(th.Join(0))
writer.write(th.PlayerName(0, "Alice"))
writer.write(th.PlayerNew(0, 100, 200))
writer.save("game.teehistorian")
# EOS chunk is automatically written when exiting context
Creating a Writer¶
th.create(**headers) → TeehistorianWriter¶
Creates a new teehistorian writer instance.
Parameters:
- **headers: Optional header fields to set immediately
Returns: A new TeehistorianWriter instance
# Basic creation
writer = th.create()
# With initial headers
writer = th.create(
server_name="My Server",
comment="Generated by my script",
map_name="dm1"
)
TeehistorianWriter Class¶
Context Manager Support¶
The writer supports Python's context manager protocol and automatically writes an Eos() chunk when exiting the context:
Method Chaining¶
All writer methods return self, enabling fluent method chaining:
writer = (th.create()
.set_header("server_name", "My Server")
.write(th.Join(0))
.write(th.PlayerName(0, "Player")))
Writing Chunks¶
write(chunk) → TeehistorianWriter¶
Writes a single chunk to the teehistorian file.
Parameters:
- chunk: A chunk object (Join, Drop, PlayerNew, etc.)
Returns: Self for method chaining
write_all(chunks) → TeehistorianWriter¶
Writes multiple chunks at once.
Parameters:
- chunks: Iterable of chunk objects
Returns: Self for method chaining
chunks = [
th.Join(0),
th.PlayerName(0, "Alice"),
th.PlayerNew(0, 100, 200),
th.Eos()
]
writer.write_all(chunks)
Header Management¶
set_header(key, value) → TeehistorianWriter¶
Sets a single header field.
Parameters:
- key (str): Header field name
- value (str): Header field value
Returns: Self for method chaining
Note: Headers can only be modified before writing the first chunk.
writer.set_header("server_name", "My Custom Server")
writer.set_header("comment", "Generated by Python script")
update_headers(headers) → TeehistorianWriter¶
Updates multiple header fields at once.
Parameters:
- headers (dict): Dictionary of header field name-value pairs
Returns: Self for method chaining
writer.update_headers({
"server_name": "My Server",
"comment": "Automated recording",
"map_name": "ctf5"
})
Saving and Output¶
save(path) → None¶
Saves the teehistorian file to disk.
Parameters:
- path (str | Path): File path to save to
getvalue() → bytes¶
Gets the complete teehistorian data as bytes.
Returns: Complete teehistorian file as bytes
writeto(file) → None¶
Writes the teehistorian data to a file-like object.
Parameters:
- file: File-like object with a write() method
with open("output.teehistorian", "wb") as f:
writer.writeto(f)
# Or with BytesIO
from io import BytesIO
buffer = BytesIO()
writer.writeto(buffer)
Properties¶
size → int¶
Gets the current size of the teehistorian data in bytes.
is_empty → bool¶
Checks if any data has been written to the teehistorian.
Utility Methods¶
reset() → None¶
Resets the writer to its initial empty state, clearing all data and headers.
writer.write(th.Join(0))
print(writer.size) # > 0
writer.reset()
print(writer.size) # 0
print(writer.is_empty) # True
__repr__() → str¶
Returns a string representation of the writer showing its current state.
writer = th.create()
print(writer) # TeehistorianWriter(size=0, status=empty)
writer.write(th.Join(0))
print(writer) # TeehistorianWriter(size=42, status=active)
Common Usage Patterns¶
Recording a Complete Game¶
with th.create(server_name="My Server", map_name="dm1") as writer:
# Players join
writer.write(th.Join(0))
writer.write(th.Join(1))
# Set player names
writer.write(th.PlayerName(0, "Alice"))
writer.write(th.PlayerName(1, "Bob"))
# Players spawn
writer.write(th.PlayerNew(0, 100, 200))
writer.write(th.PlayerNew(1, 200, 100))
# Game events
writer.write(th.NetMessage(0, "Hello everyone!"))
writer.write(th.TickSkip(30)) # 30 ticks passed
# Players leave
writer.write(th.PlayerOld(1))
writer.write(th.Drop(1, "timeout"))
# Save the recording
writer.save("game_recording.teehistorian")
Batch Processing¶
def create_recording(events, filename):
"""Create a teehistorian from a list of events."""
with th.create() as writer:
writer.set_header("comment", f"Generated from {len(events)} events")
writer.write_all(events)
writer.save(filename)
# Usage
events = [
th.Join(0),
th.PlayerName(0, "Player"),
th.PlayerNew(0, 0, 0),
]
create_recording(events, "batch_recording.teehistorian")
Streaming to Network¶
import socket
def stream_teehistorian(writer, host, port):
"""Stream teehistorian data over network."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.connect((host, port))
data = writer.getvalue()
sock.sendall(data)
# Usage
writer = th.create()
writer.write(th.Join(0))
stream_teehistorian(writer, "localhost", 8080)
Error Handling¶
try:
with th.create() as writer:
writer.set_header("server_name", "My Server")
writer.write(th.Join(0))
# Trying to modify header after writing chunks raises an error
writer.set_header("comment", "This will fail") # ValueError
except ValueError as e:
print(f"Header modification error: {e}")
try:
writer = th.create()
writer.write(th.Join(0))
# Using a closed writer
writer._closed = True
writer.write(th.Drop(0, "quit")) # ValueError
except ValueError as e:
print(f"Closed writer error: {e}")
Thread Safety¶
The TeehistorianWriter is not thread-safe. If you need to write from multiple threads, use appropriate synchronization:
import threading
writer = th.create()
lock = threading.Lock()
def write_chunk(chunk):
with lock:
writer.write(chunk)
# Use write_chunk from multiple threads
Performance Tips¶
- Use context managers - They ensure proper cleanup and automatic EOS writing
- Batch writes - Use
write_all()for better performance when writing many chunks - Pre-allocate headers - Set all headers during creation or before first write
- Stream large files - Use
writeto()with file objects instead of keeping everything in memory