coseda.map package

Submodules

coseda.map.findscanlines module

coseda.map.findscanlines.compare_strokes(leftstrokes, rightstrokes, refined_line_breaks)[source]
coseda.map.findscanlines.cross_correlate_lines(input_path, line_bounds1, line_bounds2, data_key)[source]
coseda.map.findscanlines.find_closest_minimum(x_data, y_data, target_x)[source]

Find the closest minimum in y_data to the target_x in x_data

coseda.map.findscanlines.find_closest_stroke(frame, strokes)[source]
coseda.map.findscanlines.find_line_breaks(input_path)[source]
coseda.map.findscanlines.find_strokes_by_xpos(input_path, comparison_window=10, ignore_window=10)[source]
coseda.map.findscanlines.find_strokes_by_ypos(input_path, threshold_multiplier=10)[source]
coseda.map.findscanlines.get_line_bounds_from_index(streakdirection, index)[source]
coseda.map.findscanlines.get_three_line_bounds(input_path, frame_index)[source]
coseda.map.findscanlines.moving_average(data, window_size)[source]

Smooth data by performing a moving average.

coseda.map.findscanlines.plot_correlation(correlation, title)[source]
coseda.map.findscanlines.plot_line_profiles(input_path, index, window_size=5)[source]
coseda.map.findscanlines.plot_stagepos_y_with_breaks(input_path)[source]
coseda.map.findscanlines.plot_stagepos_y_with_breaks_debug(input_path)[source]
coseda.map.findscanlines.write_combined_stroke_info(input_path, rightstrokes, leftstrokes, refined_line_breaks)[source]

coseda.map.mapping module

class coseda.map.mapping.Disk(x, y, diameter, frame_index)[source]

Bases: object

coseda.map.mapping.are_touching(disk1, disk2)[source]
coseda.map.mapping.cluster_segments(input_path, threshold, beam_diameter)[source]
coseda.map.mapping.distance(disk1, disk2)[source]
coseda.map.mapping.find_clusters(disks)[source]
coseda.map.mapping.find_matrix(input_path)[source]
coseda.map.mapping.find_seeds(input_path, clusters)[source]
coseda.map.mapping.find_touching_disks(input_path, threshold, beam_diameter, interlaceoffset=None)[source]
coseda.map.mapping.get_line_bounds(input_path, frame_index, zdim='nPeaks')[source]
coseda.map.mapping.has_common_angle(seed_frame_index, neighbor_frame_index, hdf5_data, angle_tolerance=5)[source]
coseda.map.mapping.interlacecorrection(h5file_path, zdim, offset=None, on_pick_event_handler=None)[source]
coseda.map.mapping.linearize_map(input_path, threshold, beam_diameter)[source]
coseda.map.mapping.plot_clusters(input_path, disks, clusters)[source]
coseda.map.mapping.plot_crystamorphusvac(input_path, offset=None)[source]
coseda.map.mapping.plot_crystamorphusvac_overlap(input_path)[source]
coseda.map.mapping.plot_map_peaks(h5file_path, zdim, on_pick_event_handler=None)[source]
coseda.map.mapping.plot_nPeaks_histogram(input_path)[source]
coseda.map.mapping.refine_clusters(input_path, clusters, cutoff, angle_tolerance)[source]
coseda.map.mapping.write_cluster_ids(input_path, clusters, disks)[source]

coseda.map.streaks module

Streak (scanline) assignment from the stage trajectory.

Reconstructs, for every frame, which scan row (streak) it belongs to and its position within that row, by segmenting the stage motion on the sign of the x-velocity. A “sweep” is a maximal run where the (windowed) horizontal speed is above the measured noise floor and sign(vx) is constant: direction reversals split rows regardless of x amplitude (so partial / mid-sweep first rows are handled), while the parked lead-in and turnarounds (vx ~ 0) fall out between sweeps. Only complete sweeps (spanning ~the full x width) are kept.

Writes three datasets under entry/data:

streak_id per (dense) frame: row index, or -1 where not scanning. streak_frame per (dense) frame: position within the row; resets to

frame_id_start each streak and counts missing frames so a dropped frame keeps its column slot.

streak_endpoints compound table, one row per streak, with the start/end

stage coordinates and direction.

This is the replacement for the streakdirection produced by coseda.map.findscanlines.

coseda.map.streaks.add_scalebar(ax, data_to_m=1.0, frac=0.15, loc='lower left', color='white')[source]

Draw a 1-2-5 ‘nice’ scalebar on a map axis (labelled nm / µm / mm).

The bar is ~``frac`` of the x-range; data_to_m converts axis data units to metres (coseda stage positions are metres -> 1.0). A translucent dark box keeps it legible over any colormap. Returns the artist, or None if unavailable.

coseda.map.streaks.assign_streaks(input_path, window=17, noise_k=10.0, gap=3, min_frames=20, min_span_frac=0.7, frame_id_start=1, write=True)[source]

Assign streak ids / within-streak frame ids for a coseda HDF5 file.

Reads the stage trajectory (stagepos_x_refined/stagepos_y_refined, falling back to the unrefined datasets), detects complete sweeps with find_sweeps(), and builds the dense per-frame streak_id / streak_frame labels plus a streak_endpoints table. Frames whose images were stripped (index == -1) are still labelled when their stage positions are present, so full-length intensity maps can be regenerated.

Parameters:
  • input_path (str or pathlib.Path) – Path to the coseda HDF5 file.

  • window – Passed through to find_sweeps().

  • noise_k – Passed through to find_sweeps().

  • gap – Passed through to find_sweeps().

  • min_frames – Passed through to find_sweeps().

  • min_span_frac – Passed through to find_sweeps().

  • frame_id_start (int) – Value of streak_frame at each streak’s first frame.

  • write (bool) – If True, write streak_id / streak_frame / streak_endpoints into entry/data (replacing any existing versions). If False, only compute and return them.

Returns:

streak_id, streak_frame (dense int32 arrays), endpoints (structured array), spans and extras (from find_sweeps()).

Return type:

dict

coseda.map.streaks.find_sweeps(x, y, window=17, noise_k=10.0, gap=3, min_frames=20, min_span_frac=0.7)[source]

Segment the stage trajectory into complete scan sweeps.

Parameters:
  • x (array_like) – Stage x/y position per (valid) frame.

  • y (array_like) – Stage x/y position per (valid) frame.

  • window (int) – Smoothing width (frames) applied to positions before differentiating; averages out encoder quantization. Forced odd.

  • noise_k (float) – Motion threshold is noise_k * median(|vy|) – the noise floor is measured from the vertical speed, which rests at the shared x/y noise level while a row is scanned in x.

  • gap (int) – Bridge brief sub-threshold dips of up to this many frames within a sweep.

  • min_frames (int) – Ignore runs shorter than this.

  • min_span_frac (float) – A sweep is “complete” if its x-span is at least this fraction of the median sweep x-span (drops partial / mid-sweep rows).

Returns:

  • kept (list of dict) – Complete sweeps with streak_id, start_i, stop_i (valid-frame index, half-open) and direction (+1/-1).

  • extras (dict) – raw (all sweeps before the completeness filter), dropped, vx/vy (windowed speeds) and thr – for diagnostics.

coseda.map.streaks.load_atlas(input_path, zdim='frame_mean_intensities', group='atlas')[source]

Load a saved atlas from entry/<group>/<zdim>, or None if absent.

coseda.map.streaks.plot_rasterized_map(input_path, zdim='frame_mean_intensities', cmap='viridis', vmin_pct=1.0, vmax_pct=99.0, ax=None, **kwargs)[source]

Render rasterize_map() as an image (drop-in for the old scatter map).

coseda.map.streaks.rasterize_arrays(streak_id, x, y, z, valid=None, out_cols=None, max_gap_frac=3.0, left_streak_offset_m=0.0, streak_directions=None)[source]

Rasterize per-frame arrays into a regular physical grid (array-level core).

Rows are streaks (from streak_id); within each streak z is resampled onto a common physical-x grid using the measured x. Because x/y are absolute stage coordinates, both sweep directions share the same axis – no velocity model, flip or backlash needed. valid (bool, per frame) selects the frames to use; it is always AND-ed with streak_id >= 0 and finite coords/values. Rows are ordered by measured y so the raster is upright.

Returns {raster, extent, row_y, streak_ids} with NaN where there is no data.

coseda.map.streaks.rasterize_map(input_path, zdim='frame_mean_intensities', out_cols=None, max_gap_frac=3.0, left_streak_offset_um=0.0)[source]

Rasterize a scan file onto a regular physical grid using measured coords.

Thin file wrapper over rasterize_arrays(): reads the stage trajectory, streak_id and the zdim value. Full-length zdim arrays are used on the dense frame axis so atlases can be regenerated after stripping. Compact per-image arrays are mapped onto the dense axis via index and only available images contribute.

Requires assign_streaks to have been run (needs streak_id).

coseda.map.streaks.save_atlas(input_path, zdim='frame_mean_intensities', group='atlas', **kwargs)[source]

Rasterize zdim and persist it into the file (convenience wrapper).

coseda.map.streaks.write_atlas(input_path, zdim, out, group='atlas')[source]

Persist an already-computed raster into entry/<group>/<zdim>.

Stores the 2-D raster plus the physical x (per column) and y (per row) coordinates, with extent and provenance as group attrs, so the atlas can be re-rendered or exported later without recomputing. Overwrites any existing atlas for this zdim. out is a dict from rasterize_arrays() / rasterize_map() (raster, extent, row_y, streak_ids).

Module contents

Mapping utilities and scanline analysis helpers.