# greaseweazle/track.py
#
# Written & released by Keir Fraser <keir.xen@gmail.com>
#
# This is free and unencumbered software released into the public domain.
# See the file COPYING for more details, or visit <http://unlicense.org>.
import binascii
import itertools as it
from bitarray import bitarray
from greaseweazle.flux import WriteoutFlux
from greaseweazle import optimised
# Precompensation to apply to a MasterTrack for writeout.
class Precomp:
MFM = 0
FM = 1
GCR = 2
TYPESTRING = [ 'MFM', 'FM', 'GCR' ]
def __str__(self):
return "Precomp: %s, %dns" % (Precomp.TYPESTRING[self.type], self.ns)
def __init__(self, type, ns):
self.type = type
self.ns = ns
def apply(self, bits, bit_ticks, scale):
t = self.ns * scale
if self.type == Precomp.MFM:
for i in bits.itersearch(bitarray('10100', endian='big')):
bit_ticks[i+2] -= t
bit_ticks[i+3] += t
for i in bits.itersearch(bitarray('00101', endian='big')):
bit_ticks[i+2] += t
bit_ticks[i+3] -= t
# This is primarily for GCR and FM which permit adjacent 1s (and
# have correspondingly slower bit times). However it may be useful
# for illegal MFM sequences too, especially on Amiga (custom syncwords,
# 4us-bitcell tracks). Normal MFM should not trigger these patterns.
for i in bits.itersearch(bitarray('110', endian='big')):
bit_ticks[i+1] -= t
bit_ticks[i+2] += t
for i in bits.itersearch(bitarray('011', endian='big')):
bit_ticks[i+1] += t
bit_ticks[i+2] -= t
# A pristine representation of a track, from a codec and/or a perfect image.
class MasterTrack:
@property
def bitrate(self):
return len(self.bits) / self.time_per_rev
# bits: Track bitcell data, aligned to the write splice (bitarray or bytes)
# time_per_rev: Time per revolution, in seconds (float)
# bit_ticks: Per-bitcell time values, in unitless 'ticks'
# splice: Location of the track splice, in bitcells, after the index
# weak: List of (start, length) weak ranges
def __init__(self, bits, time_per_rev, bit_ticks=None, splice=0, weak=[]):
if isinstance(bits, bytes):
self.bits = bitarray(endian='big')
self.bits.frombytes(bits)
else:
self.bits = bits
self.time_per_rev = time_per_rev
self.bit_ticks = bit_ticks
self.splice = splice
self.weak = weak
self.precomp = None
def __str__(self):
s = "\nMaster Track: splice @ %d\n" % self.splice
s += (" %d bits, %.1f kbit/s"
% (len(self.bits), self.bitrate))
if self.bit_ticks:
s += " (variable)"
s += ("\n %.1f ms / rev (%.1f rpm)"
% (self.time_per_rev * 1000, 60 / self.time_per_rev))
if len(self.weak) > 0:
s += "\n %d weak range" % len(self.weak)
if len(self.weak) > 1: s += "s"
s += ": " + ", ".join(str(n) for _,n in self.weak) + " bits"
#s += str(binascii.hexlify(self.bits.tobytes()))
return s
def flux_for_writeout(self, cue_at_index=True):
return self.flux(for_writeout=True, cue_at_index=cue_at_index)
def flux(self, for_writeout=False, cue_at_index=True):
# We're going to mess with the track data, so take a copy.
bits = self.bits.copy()
bitlen = len(bits)
# Also copy the bit_ticks array (or create a dummy one), and remember
# the total ticks that it contains.
bit_ticks = self.bit_ticks.copy() if self.bit_ticks else [1] * bitlen
ticks_to_index = sum(bit_ticks)
# Weak regions need special processing for correct flux representation.
for s,n in self.weak:
e = s + n
assert 0 < s < e < bitlen
pattern = bitarray(endian="big")
if n < 400:
# Short weak regions are written with no flux transitions.
# Actually we insert a flux transition every 32 bitcells, else
# we risk triggering Greaseweazle's No Flux Area generator.
pattern.frombytes(b"\x80\x00\x00\x00")
bits[s:e] = (pattern * (n//32+1))[:n]
else:
# Long weak regions we present a fuzzy clock bit in an
# otherwise normal byte (16 bits MFM). The byte may be
# interpreted as
# MFM 0001001010100101 = 12A5 = byte 0x43, or
# MFM 0001001010010101 = 1295 = byte 0x47
pattern.frombytes(b"\x12\xA5")
bits[s:e] = (pattern * (n//16+1))[:n]
for i in range(0, n-10, 16):
x, y = bit_ticks[s+i+10], bit_ticks[s+i+11]
bit_ticks[s+i+10], bit_ticks[s+i+11] = x+y*0.5, y*0.5
# To prevent corrupting a preceding sync word by effectively
# starting the weak region early, we start with a 1 if we just
# clocked out a 0.
bits[s] = not bits[s-1]
# Similarly modify the last bit of the weak region.
bits[e-1] = not(bits[e-2] or bits[e])
if cue_at_index:
# Rotate data to start at the index.
index = -self.splice % bitlen
if index != 0:
bits = bits[index:] + bits[:index]
bit_ticks = bit_ticks[index:] + bit_ticks[:index]
splice_at_index = index < 4 or bitlen - index < 4
else:
splice_at_index = False
if not for_writeout:
# Do not extend the track for reliable writeout to disk.
pass
elif not cue_at_index:
# We write the track wherever it may fall (uncued).
# We stretch the track with extra header gap bytes, in case the
# drive spins slow and we need more length to create an overlap.
# Thus if the drive spins slow, the track gets a longer header.
pos = 4
# We stretch by 10 percent, which is way more than enough.
rep = bitlen // (10 * 32)
bit_ticks = bit_ticks[pos:pos+32] * rep + bit_ticks[pos:]
bits = bits[pos:pos+32] * rep + bits[pos:]
elif splice_at_index:
# Splice is at the index (or within a few bitcells of it).
# We stretch the track with extra footer gap bytes, in case the
# drive motor spins slower than expected and we need more filler
# to get us to the index pulse (where the write will terminate).
# Thus if the drive spins slow, the track gets a longer footer.
pos = (self.splice - 4) % bitlen
# We stretch by 10 percent, which is way more than enough.
rep = bitlen // (10 * 32)
bit_ticks = bit_ticks[:pos] + bit_ticks[pos-32:pos] * rep
bits = bits[:pos] + bits[pos-32:pos] * rep
else:
# Splice is not at the index. We will write more than one
# revolution, and terminate the second revolution at the splice.
# For the first revolution we repeat the track header *backwards*
# to the very start of the write. This is in case the drive motor
# spins slower than expected and the write ends before the original
# splice position.
# Thus if the drive spins slow, the track gets a longer header.
bit_ticks += bit_ticks[:self.splice-4]
bits += bits[:self.splice-4]
pos = self.splice+4
fill_pattern = bits[pos:pos+32]
while pos >= 32:
pos -= 32
bits[pos:pos+32] = fill_pattern
if for_writeout and self.precomp is not None:
self.precomp.apply(bits, bit_ticks,
ticks_to_index / (self.time_per_rev*1e9))
# Convert the stretched track data into flux.
bit_ticks_i = iter(bit_ticks)
flux_list = []
flux_ticks = 0
for bit in bits:
flux_ticks += next(bit_ticks_i)
if bit:
flux_list.append(flux_ticks)
flux_ticks = 0
if flux_ticks and for_writeout:
flux_list.append(flux_ticks)
# Package up the flux for return.
flux = WriteoutFlux(ticks_to_index, flux_list,
ticks_to_index / self.time_per_rev,
index_cued = cue_at_index,
terminate_at_index = splice_at_index)
return flux
# Track data generated from flux.
class RawTrack:
def __init__(self, clock, data):
self.clock = clock
self.clock_max_adj = 0.10
self.pll_period_adj = 0.05
self.pll_phase_adj = 0.60
self.bitarray = bitarray(endian='big')
self.timearray = []
self.revolutions = []
self.import_flux_data(data)
def __str__(self):
s = "\nRaw Track: %d revolutions\n" % len(self.revolutions)
for rev in range(len(self.revolutions)):
b, _ = self.get_revolution(rev)
s += "Revolution %u (%u bits): " % (rev, len(b))
s += str(binascii.hexlify(b.tobytes())) + "\n"
b = self.bitarray[sum(self.revolutions):]
s += "Tail (%u bits): " % (len(b))
s += str(binascii.hexlify(b.tobytes())) + "\n"
return s[:-1]
def get_revolution(self, nr):
start = sum(self.revolutions[:nr])
end = start + self.revolutions[nr]
return self.bitarray[start:end], self.timearray[start:end]
def get_all_data(self):
return self.bitarray, self.timearray
def import_flux_data(self, data):
flux = data.flux()
freq = flux.sample_freq
clock = self.clock
clock_min = self.clock * (1 - self.clock_max_adj)
clock_max = self.clock * (1 + self.clock_max_adj)
index_iter = it.chain(iter(map(lambda x: x/freq, flux.index_list)),
[float('inf')])
# Make sure there's enough time in the flux list to cover all
# revolutions by appending a "large enough" final flux value.
tail = max(0, sum(flux.index_list) - sum(flux.list) + clock*freq*2)
flux_iter = it.chain(flux.list, [tail])
try:
optimised.flux_to_bitcells(
self.bitarray, self.timearray, self.revolutions,
index_iter, flux_iter,
freq, clock, clock_min, clock_max,
self.pll_period_adj, self.pll_phase_adj)
except AttributeError:
flux_to_bitcells(
self.bitarray, self.timearray, self.revolutions,
index_iter, flux_iter,
freq, clock, clock_min, clock_max,
self.pll_period_adj, self.pll_phase_adj)
def flux_to_bitcells(bit_array, time_array, revolutions,
index_iter, flux_iter,
freq, clock_centre, clock_min, clock_max,
pll_period_adj, pll_phase_adj):
nbits = 0
ticks = 0.0
clock = clock_centre
to_index = next(index_iter)
for x in flux_iter:
# Gather enough ticks to generate at least one bitcell.
ticks += x / freq
if ticks < clock/2:
continue
# Clock out zero or more 0s, followed by a 1.
zeros = 0
while True:
# Check if we cross the index mark.
to_index -= clock
if to_index < 0:
revolutions.append(nbits)
nbits = 0
to_index += next(index_iter)
nbits += 1
ticks -= clock
time_array.append(clock)
if ticks >= clock/2:
zeros += 1
bit_array.append(False)
else:
bit_array.append(True)
break
# PLL: Adjust clock frequency according to phase mismatch.
if zeros <= 3:
# In sync: adjust clock by a fraction of the phase mismatch.
clock += ticks * pll_period_adj
else:
# Out of sync: adjust clock towards centre.
clock += (clock_centre - clock) * pll_period_adj
# Clamp the clock's adjustment range.
clock = min(max(clock, clock_min), clock_max)
# PLL: Adjust clock phase according to mismatch.
new_ticks = ticks * (1 - pll_phase_adj)
time_array[-1] += ticks - new_ticks
ticks = new_ticks
# Local variables:
# python-indent: 4
# End: